Running MCP in Production: Scaling, Caching and Operations

From One Process to Many

Streamable HTTP, Testing and the MCP Inspector ran one server process. Real deployments run several, for capacity and so one crash doesn’t take the service down. The 2026-07-28 protocol is built for that: every request carries what it needs, so any instance can answer it. This lesson finds the places where that still breaks.

• Required MCP concepts: what a reverse proxy must forward, the request-state key every instance must share, state that lives outside the process, and cache hints.
• Optional production infrastructure: operational controls for the authorization from MCP Authorization (key rotation, token lifetimes, audit logs), and the same server on Kubernetes.

Everything ran in Docker with python:3.12-slim plus pip install mcp==2.3.0, and nginx:1.29-alpine. The Kubernetes part used minikube v1.38.1.

Part 1: Required MCP Concepts

Two instances and a shared database

# notes_server.py - one instance of a notes server; run two behind nginx
import os, sqlite3, sys
from typing import Annotated
from pydantic import BaseModel
from mcp.server import MCPServer
from mcp.server.mcpserver import Elicit, RequestStateSecurity, Resolve

NAME, PORT = sys.argv[1], int(sys.argv[2])
KEY = os.environ.get("STATE_KEY")  # unset: each process makes its own random key
mcp = MCPServer("notes", request_state_security=RequestStateSecurity(keys=[KEY]) if KEY else None)
db = sqlite3.connect("notes.db", check_same_thread=False, isolation_level=None)  # shared by both
db.execute("CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, text TEXT)")


class Confirm(BaseModel):
    confirm: bool


@mcp.tool()
def add_note(text: str) -> str:
    """Save a note."""
    return f"{NAME} saved note {db.execute('INSERT INTO notes (text) VALUES (?)', (text,)).lastrowid}"


def ask(note_id: int) -> Elicit[Confirm]:
    print(f"{NAME} resolving delete_note({note_id})", file=sys.stderr)
    (text,) = db.execute("SELECT text FROM notes WHERE id = ?", (note_id,)).fetchone()
    return Elicit(f"Delete note {note_id} ({text!r})?", Confirm)  # identical on every instance


@mcp.tool()
def delete_note(note_id: int, ok: Annotated[Confirm, Resolve(ask)]) -> str:
    """Delete a note after the user confirms."""
    db.execute("DELETE FROM notes WHERE id = ?", (note_id,))
    return f"{NAME} deleted note {note_id}, {db.execute('SELECT COUNT(*) FROM notes').fetchone()[0]} left"


mcp.run(transport="streamable-http", host="127.0.0.1", port=PORT)

Each instance is the same file, started with a name and a port: python -u notes_server.py A 8101 and python -u notes_server.py B 8102. They bind to 127.0.0.1 and keep notes in one SQLite file, so whatever A writes, B reads. SQLite is fine for two processes on one machine. Across machines you’d use a database server; the point is that no instance keeps state of its own. delete_note asks for confirmation through elicitation, the multi round-trip request from Elicitation, Sampling and Roots.

nginx runs in the same network namespace and round-robins requests between them:

# nginx.conf - round-robin MCP requests across two instances
events {}
http {
    upstream mcp {
        server 127.0.0.1:8101;
        server 127.0.0.1:8102;
    }
    server {
        listen 8080;
        location /mcp {
            proxy_pass http://mcp;
            proxy_http_version 1.1;
            proxy_buffering off;  # let streamed (SSE) responses through as they arrive
        }
    }
}

The 421 from the Host check

This probe sends one well-formed server/discover request through nginx four times:

# lb_probe.py - send one well-formed MCP request through nginx, four times
import httpx

META = {"io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": {}}
BODY = {"jsonrpc": "2.0", "id": 1, "method": "server/discover", "params": {"_meta": META}}
HEADERS = {"Accept": "application/json, text/event-stream", "MCP-Protocol-Version": "2026-07-28",
           "Mcp-Method": "server/discover"}
for _ in range(4):
    r = httpx.post("http://localhost:8080/mcp", json=BODY, headers=HEADERS)
    print(r.status_code, r.reason_phrase, r.text[:40])

$ python lb_probe.py; grep -c 'Invalid Host header: mcp' a.log b.log
421 Misdirected Request Invalid Host header
421 Misdirected Request Invalid Host header
421 Misdirected Request Invalid Host header
421 Misdirected Request Invalid Host header
a.log:2
b.log:2

Both instances got two requests each, and both refused them. Bound to 127.0.0.1, the SDK turns on DNS-rebinding protection and only answers requests whose Host header is localhost:<port> or 127.0.0.1:<port>. It’s a defence against a web page in your browser reaching a server on localhost. nginx replaced the client’s Host: localhost:8080 with the upstream’s name, mcp, which isn’t allowed.

The obvious fix, proxy_set_header Host $host;, isn’t enough:

$ python lb_probe.py; grep -h 'Invalid Host' a.log b.log | sort | uniq -c
421 Misdirected Request Invalid Host header
421 Misdirected Request Invalid Host header
421 Misdirected Request Invalid Host header
421 Misdirected Request Invalid Host header
      4 Invalid Host header: localhost

$host is the host name without the port, and the SDK’s allowed list is localhost:*, a host with a port. $http_host passes the client’s header through exactly:

# nginx-fixed.conf - the same, passing the client's Host header through
events {}
http {
    upstream mcp {
        server 127.0.0.1:8101;
        server 127.0.0.1:8102;
    }
    server {
        listen 8080;
        location /mcp {
            proxy_pass http://mcp;
            proxy_set_header Host $http_host;  # keep "localhost:8080", port included
            proxy_http_version 1.1;
            proxy_buffering off;  # let streamed (SSE) responses through as they arrive
        }
    }
}

$ python lb_probe.py
200 OK {"jsonrpc":"2.0","id":1,"result":{"cache
200 OK {"jsonrpc":"2.0","id":1,"result":{"cache
200 OK {"jsonrpc":"2.0","id":1,"result":{"cache
200 OK {"jsonrpc":"2.0","id":1,"result":{"cache

The request-state key

Now the real client, with a scripted user who always says yes:

# notes_client.py - talk to whichever instance nginx picks
import asyncio
from mcp import Client, types
from mcp.shared.exceptions import MCPError


async def say_yes(context, params: types.ElicitRequestParams) -> types.ElicitResult:
    print("  server asks:", params.message)
    return types.ElicitResult(action="accept", content={"confirm": True})


async def main():
    async with Client("http://localhost:8080/mcp", elicitation_callback=say_yes) as client:
        for text in ("buy milk", "call bob"):
            print((await client.call_tool("add_note", {"text": text})).content[0].text)
        try:
            print((await client.call_tool("delete_note", {"note_id": 1})).content[0].text)
        except MCPError as e:
            print("delete failed:", e.error.message)


asyncio.run(main())

$ python notes_client.py; grep -h resolving a.log b.log
B saved note 1
B saved note 2
  server asks: Delete note 1 ('buy milk')?
delete failed: Invalid or expired requestState
A resolving delete_note(1)

Saving notes worked, and A read the text of note 1 from the database B had written it to. The delete didn’t work. Instance A ran the resolver and asked the question, sealing where the call had got to into requestState with its own key. The client’s retry with the answer went to B, which couldn’t open it. Every MCPServer process makes a random key at startup unless you give it one. Start both with the same STATE_KEY. It must be at least 32 bytes, or RequestStateSecurity raises ValueError; secrets.token_hex(32) makes one:

$ python notes_client.py; grep -h resolving a.log b.log
B saved note 1
B saved note 2
  server asks: Delete note 1 ('buy milk')?
B deleted note 1, 1 left
A resolving delete_note(1)
B resolving delete_note(1)

The question came from A and the answer was handled by B: B ran the resolver again, matched the answer to the sealed request and deleted the note. In production the key comes from a secret store, never from source code. RequestStateSecurity(keys=[...]) takes a list so you can rotate it without breaking calls in flight: deploy [old, new], then [new, old], then [new] once the old states have expired (600 seconds by default).

grep also shows why the resolver’s output must be the same on every instance: B rebuilt the question to check the answer against it. Practice 1 shows what happens when it isn’t.

Cache hints

Hosts list a server’s tools often, sometimes before every model turn. A server can say how long a result stays fresh. This counts the tools/list requests that reached the server, using the tracing helper otel_setup.py from MCP Observability with OpenTelemetry Traces:

# cache_hint.py - let clients reuse the tool list for a minute
import asyncio
from otel_setup import spans
from mcp import Client
from mcp.server import MCPServer
from mcp.server.caching import CacheHint


def make_server(hints):
    mcp = MCPServer("calc", cache_hints=hints)

    @mcp.tool()
    def add(a: int, b: int) -> int:
        """Add two integers."""
        return a + b
    return mcp


async def main():
    for hints in (None, {"tools/list": CacheHint(ttl_ms=60_000, scope="public")}):
        spans.clear()
        async with Client(make_server(hints)) as client:
            for _ in range(3):
                await client.list_tools()
        reached = sum(s.name == "tools/list" for s in spans.get_finished_spans())
        print(f"{'with' if hints else 'no'} cache hint: 3 list_tools() calls, {reached} reached the server")


asyncio.run(main())

$ python cache_hint.py
no cache hint: 3 list_tools() calls, 3 reached the server
with cache hint: 3 list_tools() calls, 1 reached the server

With the hint, the Python SDK client answered the second and third calls from its cache. scope="public" says the result is the same for every user, so it may be shared across authorization contexts; use "private" for results that depend on who is asking. That only works if the tool list is stable: build the same tools in the same order on every instance, or clients may cache a list that another instance wouldn’t serve. Timeouts are the other half of talking to a fleet; they’re covered, with cancellation, in Logging, Progress and Cancellation in MCP.

Part 2: Optional Production Infrastructure

Operational controls for authorization

The authorization lesson built the flow. Running it needs four controls.

• Issuer and audience are configuration. In the lesson they’re constants. In production, set the issuer URL, the resource URL and the required scopes per environment, so a staging token is never valid in production. The audience check from that lesson is what enforces the split.
• Token lifetimes. Access tokens are bearer tokens, and the SDK doesn’t support DPoP yet, so keep them short (minutes) and let refresh tokens, rotated on every use, carry the session.
• Key rotation. The authorization server signs with a key it publishes in its JWKS. Rotate it by publishing the new key before using it, and retire the old one only after its last tokens have expired:

# rotate_keys.py - rotate the token-signing key without logging anyone out
import json, time
import jwt
from cryptography.hazmat.primitives.asymmetric import ec


def new_key(kid):
    key = ec.generate_private_key(ec.SECP256R1())
    return key, json.loads(jwt.algorithms.ECAlgorithm.to_jwk(key.public_key())) | {"kid": kid, "alg": "ES256"}


def verify(token, jwks):  # what the MCP server's verifier does with the published JWKS
    try:
        key = jwt.PyJWKSet.from_dict(jwks)[jwt.get_unverified_header(token)["kid"]]
        jwt.decode(token, key, algorithms=["ES256"], audience="http://localhost:8001/mcp")
        return "accepted"
    except (jwt.PyJWTError, KeyError) as e:
        return f"rejected ({type(e).__name__})"


old, old_jwk = new_key("2026-09")
new, new_jwk = new_key("2026-10")
claims = {"aud": "http://localhost:8001/mcp", "exp": int(time.time()) + 300}
tokens = {kid: jwt.encode(claims, k, "ES256", headers={"kid": kid}) for kid, k in (("2026-09", old), ("2026-10", new))}
for phase, keys in (("publish both", [old_jwk, new_jwk]), ("retire 2026-09", [new_jwk])):
    print(f"{phase}:", {kid: verify(t, {"keys": keys}) for kid, t in tokens.items()})

$ python rotate_keys.py
publish both: {'2026-09': 'accepted', '2026-10': 'accepted'}
retire 2026-09: {'2026-09': 'rejected (KeyError)', '2026-10': 'accepted'}

While both keys are published, tokens signed with either one verify, because the verifier picks the key by the token’s kid. Retire 2026-09 too early and every token still signed with it fails at once, as here, even though none of them has expired. Wait at least one access-token lifetime between the two steps.

• Audit logs. Record who called which tool and with what outcome. The SDK runs a middleware list around every request (the same mechanism its OpenTelemetry spans use), and get_access_token() returns the caller’s token:

# audit_server.py - the protected calc server, plus one audit line per request
import json, sys
from mcp.server.auth.middleware.auth_context import get_access_token
from protected_server import mcp


def is_tool_error(result):  # a tool result reaches middleware as a model or a plain dict
    return getattr(result, "is_error", False) or (isinstance(result, dict) and result.get("isError") is True)


async def audit(ctx, call_next):
    token, outcome = get_access_token(), "ok"
    try:
        result = await call_next(ctx)
        if is_tool_error(result):
            outcome = "tool_error"
        return result
    except Exception as e:
        outcome = type(e).__name__
        raise
    finally:
        print(json.dumps({"client": token.client_id if token else None, "method": ctx.method,
                          "tool": (ctx.params or {}).get("name"), "outcome": outcome}), file=sys.stderr, flush=True)


mcp._lowlevel_server.middleware.append(audit)  # provisional API in mcp 2.x
mcp.run(transport="streamable-http", host="localhost", port=8001)

$ python oauth_client.py; grep '^{' rs.log
client: authorize at http://localhost:9000/authorize
add -> {'result': 5}
{"client": "https://localhost:8443/client.json", "method": "server/discover", "tool": null, "outcome": "ok"}
{"client": "https://localhost:8443/client.json", "method": "tools/call", "tool": "add", "outcome": "ok"}
{"client": "https://localhost:8443/client.json", "method": "tools/list", "tool": null, "outcome": "ok"}

Each line names the OAuth client, the method, the tool and the outcome. Requests rejected for a missing or bad token never reach this middleware: the bearer-token check runs first and answers 401, so failed logins show up in the HTTP access log (here uvicorn’s), not in this one. Ship both to the same place.

On Kubernetes

The same calc server as a Deployment of two replicas behind a Service. The pod has to listen on every interface, which changes one security default:

# k8s_server.py - the calc server in a pod: listen on every interface, answer only for the Service
import os
from mcp.server import MCPServer
from mcp.server.mcpserver import RequestStateSecurity
from mcp.server.transport_security import TransportSecuritySettings

mcp = MCPServer("calc", request_state_security=RequestStateSecurity(keys=[os.environ["STATE_KEY"]]))


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b


# Binding 0.0.0.0 turns off the SDK's automatic localhost-only Host check, so set one yourself.
hosts = TransportSecuritySettings(allowed_hosts=["calc:*", "calc.default.svc.cluster.local:*"])
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, transport_security=hosts)

# calc.yaml - two replicas of the MCP server behind one Service, sharing a request-state key
apiVersion: v1
kind: Secret
metadata: {name: calc-state-key}
stringData: {STATE_KEY: "lab-only-change-me-use-a-real-random-key"}
---
apiVersion: apps/v1
kind: Deployment
metadata: {name: calc}
spec:
  replicas: 2
  selector: {matchLabels: {app: calc}}
  template:
    metadata: {labels: {app: calc}}
    spec:
      containers:
        - name: calc
          image: mcp-calc:1
          imagePullPolicy: Never          # the image was loaded into minikube, not pulled
          envFrom: [{secretRef: {name: calc-state-key}}]
          ports: [{containerPort: 8000}]
          readinessProbe: {tcpSocket: {port: 8000}}
---
apiVersion: v1
kind: Service
metadata: {name: calc}
spec:
  selector: {app: calc}
  ports: [{port: 8000, targetPort: 8000}]

# Dockerfile - the calc server image for the cluster
FROM python:3.12-slim
RUN pip install --no-cache-dir mcp==2.3.0 httpx
WORKDIR /app
COPY k8s_server.py k8s_probe.py ./
CMD ["python", "k8s_server.py"]

Build the image inside minikube and deploy it:

minikube start --driver=docker
minikube image build -t mcp-calc:1 .

$ kubectl apply -f calc.yaml
secret/calc-state-key created
deployment.apps/calc created
service/calc created
$ kubectl wait --for=condition=available deployment/calc --timeout=120s
deployment.apps/calc condition met
$ kubectl get deployment calc -o jsonpath='{.status.readyReplicas}/{.spec.replicas} replicas ready{"\n"}'
2/2 replicas ready

Then call it from a pod in the cluster, through the Service name:

# k8s_probe.py - run inside the cluster: call the Service, then try a forged Host header
import asyncio
import httpx
from mcp import Client

URL = "http://calc:8000/mcp"


async def main():
    async with Client(URL) as client:
        print("add via the Service ->", (await client.call_tool("add", {"a": 2, "b": 3})).structured_content)
    r = httpx.post(URL, json={}, headers={"Host": "evil.example", "Content-Type": "application/json",
                                          "Accept": "application/json, text/event-stream"})
    print("forged Host header ->", r.status_code, r.text)


asyncio.run(main())

$ kubectl run probe --rm -i -q --restart=Never --image=mcp-calc:1 --image-pull-policy=Never -- python k8s_probe.py
add via the Service -> {'result': 5}
forged Host header -> 421 Invalid Host header

The Service sends each connection to one of the ready replicas, and both replicas read the request-state key from one Secret, so either can finish a call the other started. The forged Host header got 421. That is only because of allowed_hosts: bound to 0.0.0.0, the SDK doesn’t turn on its localhost-only check, so without that setting the pod would answer any Host. Beyond this, Kubernetes adds nothing MCP-specific. Put the database outside the pods, set resource limits, and point the trace exporter from the observability lesson at your collector’s Service.

Practice

1. Word the question per instance. Change ask() in notes_server.py so the question starts with the instance’s name, start both instances with the shared key, and run notes_client.py.

# p24_1_server.py - notes_server.py, but each instance words the question its own way
import os, sqlite3, sys
from typing import Annotated
from pydantic import BaseModel
from mcp.server import MCPServer
from mcp.server.mcpserver import Elicit, RequestStateSecurity, Resolve

NAME, PORT = sys.argv[1], int(sys.argv[2])
KEY = os.environ.get("STATE_KEY")  # unset: each process makes its own random key
mcp = MCPServer("notes", request_state_security=RequestStateSecurity(keys=[KEY]) if KEY else None)
db = sqlite3.connect("notes.db", check_same_thread=False, isolation_level=None)  # shared by both
db.execute("CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, text TEXT)")


class Confirm(BaseModel):
    confirm: bool


@mcp.tool()
def add_note(text: str) -> str:
    """Save a note."""
    return f"{NAME} saved note {db.execute('INSERT INTO notes (text) VALUES (?)', (text,)).lastrowid}"


def ask(note_id: int) -> Elicit[Confirm]:
    print(f"{NAME} resolving delete_note({note_id})", file=sys.stderr)
    (text,) = db.execute("SELECT text FROM notes WHERE id = ?", (note_id,)).fetchone()
    return Elicit(f"{NAME} asks: delete note {note_id} ({text!r})?", Confirm)  # differs per instance


@mcp.tool()
def delete_note(note_id: int, ok: Annotated[Confirm, Resolve(ask)]) -> str:
    """Delete a note after the user confirms."""
    db.execute("DELETE FROM notes WHERE id = ?", (note_id,))
    return f"{NAME} deleted note {note_id}, {db.execute('SELECT COUNT(*) FROM notes').fetchone()[0]} left"


mcp.run(transport="streamable-http", host="127.0.0.1", port=PORT)

$ python notes_client.py 2>&1 | grep -o 'InputRequiredRoundsExceededError: .*rounds'; grep -c resolving a.log b.log
InputRequiredRoundsExceededError: Server returned InputRequiredResult for more than 10 rounds; raise input_required_max_rounds
a.log:6
b.log:5

2. A hint of zero. Change the cache hint to ttl_ms=0 and count the requests again.

# p24_2.py - a hint with ttl_ms=0 means "immediately stale"
import asyncio
from otel_setup import spans
from mcp import Client
from mcp.server import MCPServer
from mcp.server.caching import CacheHint


def make_server(hints):
    mcp = MCPServer("calc", cache_hints=hints)

    @mcp.tool()
    def add(a: int, b: int) -> int:
        """Add two integers."""
        return a + b
    return mcp


async def main():
    for hints in (None, {"tools/list": CacheHint(ttl_ms=0, scope="public")}):
        spans.clear()
        async with Client(make_server(hints)) as client:
            for _ in range(3):
                await client.list_tools()
        reached = sum(s.name == "tools/list" for s in spans.get_finished_spans())
        print(f"{'with' if hints else 'no'} cache hint: 3 list_tools() calls, {reached} reached the server")


asyncio.run(main())

$ python p24_2.py
no cache hint: 3 list_tools() calls, 3 reached the server
with cache hint: 3 list_tools() calls, 3 reached the server

3. Audit a failing tool. Attach the audit middleware to an in-memory server whose tool divides by zero, and log the tool, the arguments and the outcome.

# p24_3.py - the audit middleware on a server with a failing tool, in one process
import asyncio, json
from mcp import Client
from mcp.server import MCPServer
from mcp.server.auth.middleware.auth_context import get_access_token

mcp = MCPServer("calc")


@mcp.tool()
def divide(a: int, b: int) -> float:
    """Divide a by b."""
    return a / b


def is_tool_error(result):  # a tool result reaches middleware as a model or a plain dict
    return getattr(result, "is_error", False) or (isinstance(result, dict) and result.get("isError") is True)


async def audit(ctx, call_next):
    result = await call_next(ctx)
    token = get_access_token()
    outcome = "tool_error" if is_tool_error(result) else "ok"
    if ctx.method == "tools/call":
        print(json.dumps({"client": token.client_id if token else None, "tool": ctx.params["name"],
                          "arguments": ctx.params["arguments"], "outcome": outcome}))
    return result


async def main():
    mcp._lowlevel_server.middleware.append(audit)  # provisional API in mcp 2.x
    async with Client(mcp) as client:
        for b in (2, 0):
            await client.call_tool("divide", {"a": 1, "b": b})


asyncio.run(main())

$ python p24_3.py 2>/dev/null
{"client": null, "tool": "divide", "arguments": {"a": 1, "b": 2}, "outcome": "ok"}
{"client": null, "tool": "divide", "arguments": {"a": 1, "b": 0}, "outcome": "tool_error"}