MCP Server Caching, Timeouts and Circuit Breakers in Python

Wrap the tool in @cached to reuse recent answers, add TimeoutMiddleware so no call waits forever, and add CircuitBreakerMiddleware so the server stops calling an API that keeps failing. With Promptise Foundry that's a few lines:

import httpx

from promptise.mcp.server import CircuitBreakerMiddleware, InMemoryCache, MCPServer, TimeoutMiddleware, ToolError, cached

server = MCPServer("inventory")
server.add_middleware(CircuitBreakerMiddleware(failure_threshold=3, recovery_timeout=10))
server.add_middleware(TimeoutMiddleware(default_timeout=3))

http = httpx.AsyncClient(base_url="http://127.0.0.1:8340", timeout=1.0)


@server.tool(read_only_hint=True)
@cached(ttl=30, backend=InMemoryCache(max_size=10_000))
async def get_stock(sku: str) -> dict:
    """How many units of a product are on hand, for example for "SKU-7"."""
    response = await http.get(f"/stock/{sku}")
    if response.status_code >= 500:
        raise ToolError("The inventory service is unavailable.", code="UPSTREAM_UNAVAILABLE", retryable=True)
    return response.json()


if __name__ == "__main__":
    server.run()

The same question asked twice, through a real MCP client:

 163.1 ms  {"sku": "SKU-7", "name": "Widget 7", "on_hand": 19}
   1.8 ms  {"sku": "SKU-7", "name": "Widget 7", "on_hand": 19}

The first call went to the API. The second came from the cache. If the API fails three calls in a row, the breaker opens and the tool answers instantly with an error for ten seconds, instead of making the agent wait on a service that is down. The rest of this guide builds the full version and shows where each pattern helps and what it costs.

Each pattern guards against a different failure, and the order matters. Middleware runs in the order you add it, so the first one wraps all the others. Here's the path a call takes in the finished server:

• Python 3.10 or newer.

• Promptise Foundry 1.3.1 or newer from PyPI. httpx, used here to call the API, comes with it.

• An OpenAI API key, only for the agent run near the end. Other providers work by changing the model string, as listed in Models & Providers.

pip install "promptise>=1.3.1"
export OPENAI_API_KEY="sk-..."

Nothing here needs Docker or Redis. If you run several server processes, swap InMemoryCache for RedisCache, which needs pip install redis and a Redis server.

You'll build an inventory server with two tools: get_stock, which every customer shares, and get_price, which returns each customer's own contract price.

You can't test resilience against an API that always works. This small server stands in for yours. In its default flaky mode, half of all requests fail with a 503 and one in ten takes 2.5 seconds. Its random numbers are seeded, so runs are repeatable. You can switch it to healthy, down or hang while it runs:

"""A deliberately unreliable inventory API, standing in for the service behind your MCP server.

Modes (switch with POST /admin/mode?set=<mode>):
  flaky    50% of requests fail with 503, 10% take 2.5 seconds, the rest answer in about 150 ms
  healthy  every request answers in about 150 ms
  down     every request fails with 503 after about 150 ms
  hang     every request hangs for 30 seconds
"""

import json
import random
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

STOCK = {f"SKU-{n}": {"name": f"Widget {n}", "on_hand": (n * 37) % 120} for n in range(1, 21)}
CONTRACT_DISCOUNT = {"acme": 0.10, "globex": 0.25}

state = {"mode": "flaky", "requests": 0, "failed": 0}
rng = random.Random(42)
lock = threading.Lock()


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):  # keep the console quiet
        pass

    def send(self, status: int, body: dict) -> None:
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_POST(self):
        url = urlparse(self.path)
        if url.path == "/admin/mode":
            state["mode"] = parse_qs(url.query)["set"][0]
            return self.send(200, {"mode": state["mode"]})
        if url.path == "/admin/reset":
            with lock:
                state.update(requests=0, failed=0)
                rng.seed(42)
            return self.send(200, {"reset": True})
        self.send(404, {"error": "not found"})

    def do_GET(self):
        url = urlparse(self.path)
        if url.path == "/admin/stats":
            return self.send(200, state)

        with lock:
            state["requests"] += 1
            roll = rng.random()
        mode = state["mode"]

        if mode == "hang":
            time.sleep(30)
        time.sleep(0.15)
        if mode == "down" or (mode == "flaky" and roll < 0.5):
            with lock:
                state["failed"] += 1
            return self.send(503, {"error": "inventory backend unavailable"})
        if mode == "flaky" and roll < 0.6:
            time.sleep(2.35)

        parts = url.path.strip("/").split("/")
        if len(parts) != 2 or parts[1] not in STOCK:
            return self.send(404, {"error": "unknown sku"})
        kind, sku = parts
        item = STOCK[sku]
        if kind == "stock":
            return self.send(200, {"sku": sku, **item})
        if kind == "price":
            customer = parse_qs(url.query).get("customer", [""])[0]
            list_price = 10.0 + int(sku.split("-")[1])
            price = round(list_price * (1 - CONTRACT_DISCOUNT.get(customer, 0.0)), 2)
            return self.send(200, {"sku": sku, "customer": customer, "price_eur": price})
        self.send(404, {"error": "not found"})


if __name__ == "__main__":
    print("inventory API on http://127.0.0.1:8340 (mode: flaky)", flush=True)
    ThreadingHTTPServer(("127.0.0.1", 8340), Handler).serve_forever()

python flaky_inventory.py

Here's the tool most people write first. It's correct, and it has no defences:

import os

import httpx

from promptise.mcp.server import MCPServer

INVENTORY_URL = os.environ.get("INVENTORY_URL", "http://127.0.0.1:8340")

server = MCPServer("inventory-naive")
http = httpx.AsyncClient(base_url=INVENTORY_URL)  # httpx's default timeout: 5 seconds


@server.tool(read_only_hint=True)
async def get_stock(sku: str) -> dict:
    """How many units of a product are on hand, for example for "SKU-7"."""
    response = await http.get(f"/stock/{sku}")
    response.raise_for_status()
    return response.json()


if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=8342)

A benchmark script asked it about 10 products, three times each, the way an agent re-checks things during a conversation. Then it asked about 10 more while the API hung:

naive     flaky  answered 14/30  median    161 ms  p95   2511 ms  total  11.9 s  upstream requests  30  errors {'INTERNAL_ERROR': 16}
…
naive     hang   answered 0/10  median   5018 ms  p95   5025 ms  total  50.2 s  upstream requests  10  errors {'INTERNAL_ERROR': 10}

Fewer than half the calls succeeded, and every failure reached the model as "An internal error occurred." with "retryable": false, which tells the agent nothing useful. During the hang, each call waited for httpx's five-second default before failing: 50 seconds of the agent waiting for nothing.

Many failures are momentary, so the cheapest fix is to try again. Wait a little longer before each new try (exponential backoff), and pick a random wait up to that limit (full jitter), so many clients don't all retry at the same moment. Promptise doesn't ship a retry helper for tool code, and you don't need one:

http = httpx.AsyncClient(base_url=INVENTORY_URL, timeout=1.0)


@server.on_shutdown
async def close_http():
    await http.aclose()


async def fetch(path: str, params: dict | None = None, attempts: int = 4) -> httpx.Response:
    """GET from the inventory API. Retry 5xx answers and network errors with exponential backoff."""
    for attempt in range(attempts):
        try:
            response = await http.get(path, params=params)
            if response.status_code < 500:
                return response
        except httpx.TransportError:  # connection refused, reset, or the 1 second timeout
            pass
        if attempt < attempts - 1:
            await asyncio.sleep(random.uniform(0, 0.1 * 2**attempt))  # full jitter: up to 0.1, 0.2, 0.4 s
    raise ToolError(
        f"The inventory service didn't answer after {attempts} tries.",
        code="UPSTREAM_UNAVAILABLE",
        retryable=True,
    )

Three choices matter here:

• Only retry what can succeed next time. A 503 or a dropped connection might; a 404 won't, so 4xx answers return straight away.

• Give each try a short timeout. timeout=1.0 on the client means a 2.5-second response counts as a failure and gets retried, rather than eating the whole budget.

• Fail with a clear, retryable error. When all four tries fail, the model gets UPSTREAM_UNAVAILABLE with "retryable": true instead of an internal error.

Agents ask the same thing more often than you'd think: they re-check before acting, and several agents ask about the same popular products. @cached stores each result for ttl seconds, keyed by the server, the function, its arguments and, by default, the caller. Stock levels are the same for every customer, so get_stock shares one entry per product with scope="shared":

stock_cache = InMemoryCache(max_size=10_000)


@server.tool(read_only_hint=True, max_concurrent=4)
@cached(ttl=30, backend=stock_cache, scope="shared")  # the same for every customer
async def get_stock(sku: str) -> dict:
    """How many units of a product are on hand, for example for "SKU-7". Up to 30 seconds old."""
    response = await fetch(f"/stock/{sku}")
    if response.status_code == 404:
        return {"sku": sku, "found": False}
    return response.json()

Put @cached below @server.tool(), so it wraps your function. max_size stops the cache from growing without limit: when it's full, the least recently used entry is dropped. Here are four calls against the flaky API, timed by the client over HTTP:

get_stock(SKU-7)    166.0 ms  {"sku": "SKU-7", "name": "Widget 7", "on_hand": 19}
get_stock(SKU-7)      2.8 ms  {"sku": "SKU-7", "name": "Widget 7", "on_hand": 19}
get_stock(SKU-7)      2.2 ms  {"sku": "SKU-7", "name": "Widget 7", "on_hand": 19}
get_stock(SKU-8)    999.8 ms  {"sku": "SKU-8", "name": "Widget 8", "on_hand": 56}
requests that reached the inventory API: 5

The repeat calls for SKU-7 never left the server. SKU-8 took about a second because its first three tries failed and the fourth worked: five API requests in total, and the agent saw four clean answers. Errors aren't cached, so a failed call is retried in full next time. Note that None results aren't cached either: the cache treats None as "nothing stored", so return an empty dict or list instead. Since 1.3.0, identical calls that arrive while the first one is still running wait for it and share its result, so a burst of the same question costs one API request.

The tool's description says "Up to 30 seconds old" on purpose. The model reads it, and it's the honest contract: a cache trades freshness for speed.

get_price returns each customer's contract price, so the cache key must include the caller. Since 1.3.0 that's the default: @cached keys every entry on the caller (scope="client": the token issuer, tenant and client ID), so get_price needs nothing extra:

@server.tool(read_only_hint=True, max_concurrent=4)
@cached(ttl=300, backend=price_cache)  # keyed per caller by default
async def get_price(sku: str) -> dict:
    """Your contract price for a product in EUR. Up to 5 minutes old."""
    response = await fetch(f"/price/{sku}", params={"customer": get_context().client_id})
    if response.status_code == 404:
        return {"sku": sku, "found": False}
    return response.json()

Two customers, each asking twice, each with their own API key:

acme       166.2 ms  {"sku": "SKU-7", "customer": "acme", "price_eur": 15.3}
globex     159.4 ms  {"sku": "SKU-7", "customer": "globex", "price_eur": 12.75}
acme         2.1 ms  {"sku": "SKU-7", "customer": "acme", "price_eur": 15.3}
globex       2.4 ms  {"sku": "SKU-7", "customer": "globex", "price_eur": 12.75}

Each customer gets their own cached price. The client ID comes from AuthMiddleware, set up in the complete server below. The tenant is part of the key too when the caller has one, and scope="tenant" shares entries between all clients of a tenant.

This script checks three cache setups that went wrong in 1.2.1, with two callers, acme and globex:

middleware = make_server("cache-middleware")
middleware.add_middleware(CacheMiddleware(ttl=300))


@middleware.tool()
async def stock(sku: str) -> str:
    return f"stock of {sku} for {get_context().client_id}"

1. plain @cached, per-caller answer
   acme  : SKU-7 price for acme
   globex: SKU-7 price for globex
2. CacheMiddleware
   acme   SKU-7: stock of SKU-7 for acme
   globex SKU-9: stock of SKU-9 for globex
3. @cached with a ctx parameter, same caller, same sku
   acme  : stock of SKU-7, handler run 1
   acme  : stock of SKU-7, handler run 1
   acme  : stock of SKU-7, handler run 1

• Plain `@cached` on a per-caller tool keeps callers apart: globex got its own price. In 1.2.1 it got acme's.

• `CacheMiddleware` answered globex's question about SKU-9 with SKU-9, for globex. In 1.2.1 it returned acme's SKU-7.

• `@cached` on a tool with a `ctx: RequestContext` parameter now hits: injected parameters such as the request context are left out of the key. In 1.2.1 every call was a miss.

Retries make a call more likely to succeed, but they can also make a failing call last a long time. TimeoutMiddleware sets the most an agent will ever wait for one tool call:

server.add_middleware(TimeoutMiddleware(default_timeout=3))

You can override it per tool with @server.tool(timeout=...). Since 1.3.0 the per-tool value works without TimeoutMiddleware too, for async tools: a test tool with timeout=0.2 that slept for one second stopped at 0.20 s with TIMEOUT. In 1.2.1 it finished normally after 1.00 s.

Here's one call to each server while the API hangs:

naive     answered after 5.03 s:
{"error": {"code": "INTERNAL_ERROR", "message": "An internal error occurred.", "retryable": false}}
resilient answered after 3.02 s:
{
  "error": {
    "code": "TIMEOUT",
    "message": "Tool 'get_stock' timed out after 3s",
    "retryable": true,
    "suggestion": "The tool did not finish within 3s. It may depend on a slow or unavailable service: retry later, or continue without it."
  }
}

Think of it as two budgets: one second per try inside fetch, three seconds for the whole call. The overall budget wins, so the agent never waits longer than three seconds. The suggestion tells the model to retry later or carry on without the tool, and the code and retryable flag are what the agent acts on.

Retries and timeouts help with blips. During a real outage they make things worse: every call still spends its full three seconds failing, and your retries add load to a service that's trying to recover. A circuit breaker counts consecutive failures per tool. After failure_threshold of them it opens and refuses calls immediately. After recovery_timeout seconds it lets calls through again to test the service:

breaker = CircuitBreakerMiddleware(
    failure_threshold=3,
    recovery_timeout=10,
    excluded_tools={"inventory_status"},
)

When the breaker is open, it raises CircuitOpenError. Since 1.3.0 the model gets that as a CIRCUIT_OPEN error, marked retryable, with the wait in the message, the suggestion and details. In 1.2.1 it arrived as a non-retryable "An internal error occurred.", and this guide needed an exception handler to fix that. Now the default answer, here for a fourth call after three failures, is already one the agent can act on:

{
  "error": {
    "code": "CIRCUIT_OPEN",
    "message": "Circuit open for tool 'get_stock'. Retry after 10.0s.",
    "retryable": true,
    "suggestion": "'get_stock' is paused after repeated failures. Wait about 10s before calling it again, or continue without it.",
    "details": {
      "tool": "get_stock",
      "retry_after_seconds": 10.0
    }
  }
}

@server.exception_handler(CircuitOpenError) still works if you want your own wording. Give agents and operators a way to see the state, too. inventory_status is excluded from the breaker, so it always answers:

@server.tool(read_only_hint=True)
async def inventory_status() -> dict:
    """Whether the inventory tools are working ("closed"), paused ("open") or being retried ("half_open")."""
    return {tool: breaker.get_state(tool).value for tool in ("get_stock", "get_price")}

Here's a full cycle. The API goes down, comes back, and the breaker recovers:

--- inventory API is now healthy
SKU-7     200.0 ms  {"sku": "SKU-7", "name": "Widget 7", "on_hand": 19}
       circuit: closed
--- inventory API is now down
SKU-1     990.4 ms  {"code": "UPSTREAM_UNAVAILABLE", "message": "The inventory service didn't answer after 4 tries."}
       circuit: closed
SKU-2    1051.9 ms  {"code": "UPSTREAM_UNAVAILABLE", "message": "The inventory service didn't answer after 4 tries."}
       circuit: closed
SKU-3     875.8 ms  {"code": "UPSTREAM_UNAVAILABLE", "message": "The inventory service didn't answer after 4 tries."}
       circuit: open
SKU-4       3.0 ms  {"code": "CIRCUIT_OPEN", "message": "Circuit open for tool 'get_stock'. Retry after 10.0s."}
       circuit: open
SKU-7       2.5 ms  {"code": "CIRCUIT_OPEN", "message": "Circuit open for tool 'get_stock'. Retry after 10.0s."}
       circuit: open
--- inventory API is now healthy
SKU-5       2.5 ms  {"code": "CIRCUIT_OPEN", "message": "Circuit open for tool 'get_stock'. Retry after 10.0s."}
       circuit: open
--- waiting 10 seconds for recovery_timeout
SKU-5     184.0 ms  {"sku": "SKU-5", "name": "Widget 5", "on_hand": 65}
       circuit: closed
SKU-6     158.0 ms  {"sku": "SKU-6", "name": "Widget 6", "on_hand": 102}
       circuit: closed

What to notice:

• Failing fast. After three failures, calls took a few milliseconds instead of a second, and no request reached the API.

• Cached data is refused too. SKU-7 was in the cache, but the breaker sits in front of the tool, so it never got there. That's the cost of this ordering.

• Healthy calls are refused until the timer runs out. The API was back for SKU-5's first try, and the breaker still said no. After ten seconds, the next call went through as the single test call, and its success closed the circuit. Since 1.3.0 exactly one call gets through while the breaker is half-open; calls arriving meanwhile get CIRCUIT_OPEN.

Since 1.3.0 the breaker counts only unexpected exceptions and retryable errors, such as UPSTREAM_UNAVAILABLE and TIMEOUT. A non-retryable ToolError for bad input, an auth denial, or a rate-limit or capacity refusal leaves the failure streak unchanged, so a model asking three times for a product that doesn't exist can't pause the tool for everyone. In 1.2.1 it could, which is why get_stock returns {"found": false} for a 404; that's still a friendlier answer than an error. CircuitBreakerMiddleware(is_failure=...) changes the rule.

The last two limits protect the API from too much traffic, rather than from failure. max_concurrent=4 on a tool allows four calls in flight at once. Ten parallel calls for products that weren't cached yet:

SKU-11 on hand: 47
SKU-12 on hand: 84
SKU-13 on hand: 1
SKU-14 on hand: 38
SKU-15 CONCURRENCY_LIMIT_EXCEEDED: Tool 'get_stock' at capacity (4 concurrent calls)
SKU-16 CONCURRENCY_LIMIT_EXCEEDED: Tool 'get_stock' at capacity (4 concurrent calls)
SKU-17 CONCURRENCY_LIMIT_EXCEEDED: Tool 'get_stock' at capacity (4 concurrent calls)
SKU-18 CONCURRENCY_LIMIT_EXCEEDED: Tool 'get_stock' at capacity (4 concurrent calls)
SKU-19 CONCURRENCY_LIMIT_EXCEEDED: Tool 'get_stock' at capacity (4 concurrent calls)
SKU-20 CONCURRENCY_LIMIT_EXCEEDED: Tool 'get_stock' at capacity (4 concurrent calls)
circuit: closed

Extra calls are refused with a retryable error, not queued. Since 1.3.0 the code is CONCURRENCY_LIMIT_EXCEEDED, so the agent can tell "busy, retry in a moment" from "over quota". Set the limit to what the API can take, and remember that a model often asks for several tools at once.

The last line, circuit: closed, matters too. Promptise adds a per-tool limiter automatically when a tool sets max_concurrent. Since 1.3.0 it puts the limiter in front of the first circuit breaker, and the breaker ignores capacity and rate-limit refusals in any order. In 1.2.1 the limiter landed inside the breaker, so a burst like this one opened the circuit. A test server with a breaker and a max_concurrent=2 tool, hit with eight parallel calls:

burst of 8: {'done': 2, 'CONCURRENCY_LIMIT_EXCEEDED': 6}
after burst: done 99
chain: ["PerToolConcurrencyLimiter", "CircuitBreakerMiddleware"]

No manual wiring needed. The 1.2.1 version of this guide added PerToolConcurrencyLimiter by hand, before the breaker. You can still add it yourself to choose its place; Promptise sees it's there and doesn't add a second one.

Rate limiting caps calls over time. TokenBucketLimiter(rate_per_minute=600, burst=60) allows bursts of 60 calls, refilled at ten a second. per_tool=True gives each client a separate bucket per tool:

server.add_middleware(RateLimitMiddleware(TokenBucketLimiter(rate_per_minute=600, burst=60), per_tool=True))

Eighty back-to-back calls from one client:

{
  "error": {
    "code": "RATE_LIMIT_EXCEEDED",
    "message": "Rate limit exceeded",
    "retryable": true,
    "suggestion": "Wait 1 second before retrying.",
    "details": {
      "retry_after_seconds": 0.031
    }
  }
}
{'answered': 66, 'RATE_LIMIT_EXCEEDED': 14}

The first 60 used up the burst, a few more fit in as the bucket refilled, and the rest were told to wait a second. The rate limiter sits in front of the cache, so cached answers count against the limit too. For a fixed limit on a single tool, @server.tool(rate_limit="100/min") works without any middleware.

The complete server

import asyncio
import os
import random

import httpx

from promptise.mcp.server import (
    APIKeyAuth,
    AuthMiddleware,
    CircuitBreakerMiddleware,
    InMemoryCache,
    MCPServer,
    RateLimitMiddleware,
    TimeoutMiddleware,
    TokenBucketLimiter,
    ToolError,
    cached,
    get_context,
)

INVENTORY_URL = os.environ.get("INVENTORY_URL", "http://127.0.0.1:8340")

server = MCPServer("inventory", version="1.0.0", require_auth=True)

breaker = CircuitBreakerMiddleware(
    failure_threshold=3,
    recovery_timeout=10,
    excluded_tools={"inventory_status"},
)

# Middleware runs in the order it's added: the first one wraps all the others.
server.add_middleware(
    AuthMiddleware(
        APIKeyAuth(
            keys={
                os.environ["INVENTORY_ACME_KEY"]: {"client_id": "acme"},
                os.environ["INVENTORY_GLOBEX_KEY"]: {"client_id": "globex"},
            }
        )
    )
)
server.add_middleware(RateLimitMiddleware(TokenBucketLimiter(rate_per_minute=600, burst=60), per_tool=True))
server.add_middleware(breaker)
server.add_middleware(TimeoutMiddleware(default_timeout=3))


http = httpx.AsyncClient(base_url=INVENTORY_URL, timeout=1.0)


@server.on_shutdown
async def close_http():
    await http.aclose()


async def fetch(path: str, params: dict | None = None, attempts: int = 4) -> httpx.Response:
    """GET from the inventory API. Retry 5xx answers and network errors with exponential backoff."""
    for attempt in range(attempts):
        try:
            response = await http.get(path, params=params)
            if response.status_code < 500:
                return response
        except httpx.TransportError:  # connection refused, reset, or the 1 second timeout
            pass
        if attempt < attempts - 1:
            await asyncio.sleep(random.uniform(0, 0.1 * 2**attempt))  # full jitter: up to 0.1, 0.2, 0.4 s
    raise ToolError(
        f"The inventory service didn't answer after {attempts} tries.",
        code="UPSTREAM_UNAVAILABLE",
        retryable=True,
    )


stock_cache = InMemoryCache(max_size=10_000)
price_cache = InMemoryCache(max_size=10_000)


@server.tool(read_only_hint=True, max_concurrent=4)
@cached(ttl=30, backend=stock_cache, scope="shared")  # the same for every customer
async def get_stock(sku: str) -> dict:
    """How many units of a product are on hand, for example for "SKU-7". Up to 30 seconds old."""
    response = await fetch(f"/stock/{sku}")
    if response.status_code == 404:
        return {"sku": sku, "found": False}
    return response.json()


@server.tool(read_only_hint=True, max_concurrent=4)
@cached(ttl=300, backend=price_cache)  # keyed per caller by default
async def get_price(sku: str) -> dict:
    """Your contract price for a product in EUR. Up to 5 minutes old."""
    response = await fetch(f"/price/{sku}", params={"customer": get_context().client_id})
    if response.status_code == 404:
        return {"sku": sku, "found": False}
    return response.json()


@server.tool(read_only_hint=True)
async def inventory_status() -> dict:
    """Whether the inventory tools are working ("closed"), paused ("open") or being retried ("half_open")."""
    return {tool: breaker.get_state(tool).value for tool in ("get_stock", "get_price")}


if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=8341)

Run it with two test keys, and it listens at http://127.0.0.1:8341/mcp:

export INVENTORY_ACME_KEY="acme-test-key"
export INVENTORY_GLOBEX_KEY="globex-test-key"
python inventory_server.py

The same benchmark as in step 2, now against both servers. The flaky run asks about 10 products three times each; the hang run asks about 10 more while the API hangs:

naive     flaky  answered 14/30  median    161 ms  p95   2511 ms  total  11.9 s  upstream requests  30  errors {'INTERNAL_ERROR': 16}
resilient flaky  answered 29/30  median      4 ms  p95   1091 ms  total   9.4 s  upstream requests  29  errors {'UPSTREAM_UNAVAILABLE': 1}
naive     hang   answered 0/10  median   5018 ms  p95   5025 ms  total  50.2 s  upstream requests  10  errors {'INTERNAL_ERROR': 10}
resilient hang   answered 0/10  median      3 ms  p95   3009 ms  total   9.0 s  upstream requests   9  errors {'TIMEOUT': 3, 'CIRCUIT_OPEN': 7}

The benchmark ran three times with the same counts each time, because the flaky API's randomness is seeded; the per-call timings moved by up to about 250 ms between runs, and the totals by up to half a second. An earlier set of runs, on a machine busy with other work, was slower across the board, and in one of them a 2.5-second answer ran into the 3-second timeout: timeouts turn a slow machine into errors, so leave headroom. Read the numbers honestly:

• Retries cost requests; the cache paid them back. Retries added requests and the cache removed about as many, by answering repeat questions itself: 29 requests in total, against the naive server's 30.

• The slowest calls are still slow. A p95 of about 1.1 seconds comes from first-time lookups that needed several tries. Caching helps the second question, not the first.

• During the outage, nothing got answered either way. The difference is that the resilient server gave up in 9 seconds instead of 50, and told the model to come back later.

Numbers are one thing; what matters is what the agent does with the errors. This agent connects to the server with acme's key, while the API is down:

import asyncio
import os

from promptise import build_agent
from promptise.config import HTTPServerSpec


async def main():
    agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={
            "inventory": HTTPServerSpec(
                url="http://127.0.0.1:8341/mcp",
                api_key=os.environ["INVENTORY_ACME_KEY"],
            ),
        },
        instructions="You answer stock questions for the warehouse team. Use the inventory tools.",
        trace_tools=True,
    )
    try:
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "How many units of SKU-3, SKU-4, SKU-9 and SKU-12 are on hand?"}]}
        )
        print(">>>", result["messages"][-1].content)
    finally:
        await agent.shutdown()


asyncio.run(main())

→ Invoking tool: get_stock with {'sku': 'SKU-3'}
→ Invoking tool: get_stock with {'sku': 'SKU-4'}
→ Invoking tool: get_stock with {'sku': 'SKU-9'}
→ Invoking tool: get_stock with {'sku': 'SKU-12'}
✖ get_stock error: UPSTREAM_UNAVAILABLE: The inventory service didn't answer after 4 tries.
✖ get_stock error: UPSTREAM_UNAVAILABLE: The inventory service didn't answer after 4 tries.
✖ get_stock error: UPSTREAM_UNAVAILABLE: The inventory service didn't answer after 4 tries.
✖ get_stock error: UPSTREAM_UNAVAILABLE: The inventory service didn't answer after 4 tries.
→ Invoking tool: inventory_status with {}
✔ Tool result from inventory_status: {"get_stock": "open", "get_price": "closed"}
>>> I couldn't fetch the on-hand quantities right now.

I tried to retrieve SKU-3, SKU-4, SKU-9 and SKU-12 but the inventory service did not respond. The inventory status shows get_stock is currently paused ("open"), so live stock queries are unavailable.

Options:
- I can retry now (or retry a few times).
- I can poll and notify you when the service becomes available.
- Or you can try again later.

Which would you prefer?

The model asked for all four products at once, which fits within max_concurrent=4. All four failed after their retries, which opened the breaker. Instead of retrying blindly, the agent checked inventory_status, saw the tool was paused, and told the user plainly. It didn't make up numbers. Its offer to poll and tell you when the service is back is the model's own idea, though: this agent has no tool for that, so instructions that list what it can actually do would keep such offers honest.

These hold for Promptise Foundry 1.3.1, checked against its source and the runs above:

• The in-memory cache belongs to one process. Run several workers and each has its own cache. RedisCache(url=..., prefix=...) shares one. Since 1.3.0 it sends TTLs in milliseconds, so a ttl under one second works. Checked against a Redis 7 server: two instances of the same server shared one cached entry, and an entry with ttl=0.3 was gone after half a second.

• Coalescing is per process. Five identical calls that arrive together share one upstream call within a process; a test showed the handler running once. Separate workers each make their own call.

• The breaker is per tool and shared by every client. One client's failures pause the tool for all of them, and two tools that call the same API each count their own failures.

• An open breaker blocks cache hits. With this middleware order, cached answers aren't served during an outage. Serving stale data when the API is down needs your own code.

How do I set an MCP server timeout?

On the server, add TimeoutMiddleware(default_timeout=...) and override per tool with @server.tool(timeout=...); since 1.3.0 the per-tool value works on async tools even without the middleware. Also set a timeout on your own HTTP client, shorter than the tool's, so one slow request can be retried within the budget. MCP clients have their own timeouts too (Promptise's MCPClient uses a 30-second HTTP timeout by default), so keep your server's deadline below the client's.

How do I add rate limiting to an MCP server?

Add RateLimitMiddleware with a TokenBucketLimiter(rate_per_minute=..., burst=...). Buckets are per client by default, include the tenant when there is one, and per_tool=True splits them per tool. For a fixed limit on one tool, @server.tool(rate_limit="100/min") is enforced automatically.

How do I retry with exponential backoff in Python?

Loop over a fixed number of attempts, and after each failure sleep a random time between zero and a base delay times two to the power of the attempt number, as fetch does above. Only retry errors that might succeed next time, such as 5xx answers, timeouts and dropped connections, and cap the total time with a timeout. Retries multiply the load on a struggling service, so pair them with a circuit breaker.

How does a circuit breaker work in Python?

It wraps calls to a dependency and counts failures in a row. Past a threshold it "opens" and rejects calls immediately for a cool-down period, then lets calls through to test whether the dependency has recovered, and "closes" again on success. In a Promptise MCP server, CircuitBreakerMiddleware does this per tool, and get_state() and reset() let you inspect or clear it.

Does caching work across several MCP server instances?

Not with InMemoryCache, which lives inside one process. Pass a RedisCache as the backend of @cached to share one cache across processes and machines. The same rules about per-caller keys apply.

• Caching & Performance: @cached, cache backends, rate limits, concurrency and timeouts.

• Resilience Patterns: the circuit breaker, health checks and exception handlers.

• Production Features: everything else for running a server in production.

• Routers & Middleware: how the middleware chain works and how to write your own.

• Multi-Tenancy: tenants, and keeping their data apart.

• How to Connect MCP Servers to Your AI Agent in Python: put an agent in front of this server.

• OpenAPI to MCP: Turn Any REST API into an MCP Server: generate the server for your API, then add these patterns.