MCP Middleware for Servers: Auth, Logging, Rate Limits

MCP middleware is a function that wraps a server's request handling. Each middleware receives the request context and a call_next function: code before call_next runs on the way in, code after it runs on the way out, and raising an error refuses the request before the tool runs. Middleware is a feature of server frameworks such as FastMCP and Promptise, not of the MCP specification, so the API differs between them.

@server.middleware                       # Promptise: any async (ctx, call_next) callable
async def timing(ctx, call_next):
    start = time.perf_counter()
    result = await call_next(ctx)        # the rest of the chain, then the tool
    ctx.logger.info("%s took %.3fs", ctx.tool_name, time.perf_counter() - start)
    return result

Key takeaways:

• Middleware runs in registration order. The first one added is the outermost layer: first in, last out.

• Order is behavior. A rate limiter added before authentication can't tell callers apart, and an audit log placed inside a rate limiter never sees refused calls.

• Refuse by raising, not by returning. A raised error becomes a structured error the agent can read.

• Use built-ins for the common cases. Both frameworks ship auth, rate limiting, caching, timeouts and logging.

• Middleware state lives in one process. Rate-limit buckets and caches need shared storage when you run replicas.

An MCP client sends JSON-RPC requests: tools/call, resources/read, prompts/get, plus listing and session requests. The server framework routes each request to a handler. Middleware is a chain of wrappers around that handler, often called the onion model:

tools/call get_order {"order_id": "A-1001"}
        │
        ▼
┌─ AuditLog ──────────────────────────────────────────────┐   1st added: sees every call
│ ┌─ AuthMiddleware ────────────────────────────────────┐ │   2nd: sets ctx.client_id
│ │ ┌─ RateLimitMiddleware ───────────────────────────┐ │ │   3rd: bucket per client
│ │ │ ┌─ CacheMiddleware ───────────────────────────┐ │ │ │   4th: entry per client
│ │ │ │ ┌─ TimeoutMiddleware ─────────────────────┐ │ │ │ │   5th: times the handler only
│ │ │ │ │  guards (roles=) ──▶ tool handler       │ │ │ │ │
│ │ │ │ └─────────────────────────────────────────┘ │ │ │ │
│ │ │ └─────────────────────────────────────────────┘ │ │ │
│ │ └─────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
        │  result (or error) travels back out through every layer
        ▼
exception handlers ──▶ serialized result or structured error ──▶ client

A middleware can do four things:

• Act before the call: authenticate, check a quota, start a timer, write a trace span.

• Act after the call: log the result, store it in a cache, record a metric.

• Short-circuit: raise an error and never call call_next, so the tool doesn't run.

• Transform: change the arguments on the way in or the result on the way out.

All layers share one context object per request. That is how an outer layer can read the client ID that an inner AuthMiddleware set, after call_next returns.

Frameworks differ in which requests middleware sees. In Promptise, the chain runs for tool calls, resource reads and prompt requests, and ctx.request_type tells them apart. Listing requests don't pass through it: MCPServer(hide_unauthorized_tools=True) filters lists per caller instead. In FastMCP, hooks such as on_message, on_request, on_call_tool and on_list_tools cover every message. The official MCP Python SDK 2.x also has a Server.middleware list of (ctx, call_next) callables, which its source marks as provisional.

Middleware inside a server is not the only place for these rules. MCP gateways such as Traefik Hub's MCP middleware sit in front of servers and apply OAuth and policy checks to each request. A gateway suits many servers that you don't all own. In-server middleware suits rules that need the tool's own context, such as its roles, annotations or arguments.

Use middleware for rules that apply to every tool or every caller. Use per-tool settings when the rule depends on the tool, such as who may call it or how long it may take. Keep business logic in the tools themselves. And keep middleware fast: it runs on every call, so a network round trip per request adds latency everywhere.

• Python 3.10 or later and Promptise Foundry 1.4.3 or later. The steps need no API key, because they call the server in-process.

• For the FastMCP section, FastMCP 4.1.2 in a separate virtual environment. FastMCP 4 requires MCP SDK 2.x, while Promptise 1.4.3 pins mcp<2.0.

• About 20 minutes.

pip install "promptise>=1.4.3"

A middleware in Promptise is any async callable that takes (ctx, call_next). Register a function with @server.middleware or an instance with server.add_middleware(). This server has three: a function, a timing class, and a class that refuses tools tagged write:

"""Three middlewares around two tools: the order they run in, and a short-circuit."""
import asyncio
import time

from promptise.mcp.server import MCPServer, ToolError
from promptise.mcp.server.testing import TestClient

server = MCPServer(name="orders")


@server.middleware
async def outer(ctx, call_next):
    """A function middleware. Code after call_next only runs when the call succeeds."""
    print(f"outer   -> {ctx.tool_name}")
    result = await call_next(ctx)
    print(f"outer   <- {ctx.tool_name}")
    return result


class Timing:
    """A class middleware: any async callable taking (ctx, call_next) works."""

    async def __call__(self, ctx, call_next):
        start = time.perf_counter()
        print(f"timing  -> {ctx.tool_name}")
        try:
            return await call_next(ctx)
        finally:  # runs on success and on error
            print(f"timing  <- {ctx.tool_name} after {1000 * (time.perf_counter() - start):.0f} ms")


class ReadOnlyMode:
    """Short-circuits: refuses tools tagged 'write' without running them."""

    def __init__(self, enabled: bool) -> None:
        self.enabled = enabled

    async def __call__(self, ctx, call_next):
        if self.enabled and "write" in ctx.state["tool_def"].tags:
            raise ToolError("The server is in read-only mode.", code="READ_ONLY", retryable=True)
        return await call_next(ctx)


server.add_middleware(Timing())
server.add_middleware(ReadOnlyMode(enabled=True))


@server.tool()
async def get_order(order_id: str) -> dict:
    """Look up one order by ID."""
    print("  handler: get_order runs")
    await asyncio.sleep(0.1)
    return {"order_id": order_id, "status": "shipped"}


@server.tool(tags=["write"])
async def cancel_order(order_id: str) -> str:
    """Cancel an order."""
    print("  handler: cancel_order runs")
    return f"{order_id} cancelled"


async def main() -> None:
    client = TestClient(server)
    for tool in ("get_order", "cancel_order"):
        result = await client.call_tool(tool, {"order_id": "A-1001"})
        print(f"result: {result[0].text}\n")


asyncio.run(main())

outer   -> get_order
timing  -> get_order
  handler: get_order runs
timing  <- get_order after 101 ms
outer   <- get_order
result: {"order_id": "A-1001", "status": "shipped"}

outer   -> cancel_order
timing  -> cancel_order
timing  <- cancel_order after 0 ms
result: {
  "error": {
    "code": "READ_ONLY",
    "message": "The server is in read-only mode.",
    "retryable": true
  }
}

Three things to notice:

• The onion order. outer was registered first, so it runs first on the way in and last on the way out.

• Short-circuiting. ReadOnlyMode raised a ToolError, so cancel_order never ran, and the client got a structured READ_ONLY error with retryable: true.

• Code after `call_next` doesn't run on errors. outer printed no <- line for cancel_order, because the error skipped the rest of its body. Timing used try/finally, so it logged both calls. Any middleware that must record every call, such as an audit log, needs finally or an except that re-raises.

TestClient runs calls through the server's full pipeline (validation, middleware, guards and the handler) without a transport, which makes it the fastest way to test a middleware chain.

The rest of this guide builds one server, orders_server.py, with four tools and a stack of five middlewares. Authentication comes early because other layers depend on knowing the caller. AuthMiddleware with APIKeyAuth maps each key to a client ID and roles:

KEYS = {
    os.environ.get("SUPPORT_KEY", "sk-test-support"): {"client_id": "support-bot", "roles": ["support"]},
    os.environ.get("REPORT_KEY", "sk-test-report"): {"client_id": "report-bot", "roles": ["viewer"]},
}
server.add_middleware(AuthMiddleware(APIKeyAuth(keys=KEYS)))                # 2. who is calling

The tools opt in with auth=True, and refund_order also declares roles=["support"].

A small driver, demo.py, calls the server in-process through TestClient, sending a key per call the way an HTTP client sends the x-api-key header:

client = TestClient(server)
out = await client.call_tool("get_order", {"order_id": "A-1001"}, headers={"x-api-key": "sk-test-report"})

no key                       AUTHENTICATION_ERROR: Missing API key  [0 ms]
wrong key                    AUTHENTICATION_ERROR: Invalid API key  [0 ms]
report-bot get_order         {"order_id": "A-1001", "status": "shipped", "total": 42.0}  [203 ms]
report-bot refund_order      ACCESS_DENIED: Requires any of roles [support], but client has [viewer]  [1 ms]
support-bot refund_order     {"order_id": "A-1001", "refunded": 5.0}  [0 ms]

AuthMiddleware only authenticates tools marked auth=True (or every tool with MCPServer(require_auth=True)). The roles=["support"] check is a guard, and guards run after the whole middleware chain, right before the handler. That's why report-bot passed authentication and was then refused with ACCESS_DENIED. For JWT, OAuth and scopes, see MCP server authentication with JWT and OAuth.

An audit log should record every call, including the ones that were refused. Put it first, so it is the outermost layer. This custom middleware, AuditLog, writes one JSON line per call. It is a class because it holds configuration (the file path); its __call__ method is the middleware:

    async def __call__(self, ctx, call_next):
        start = time.perf_counter()
        entry = {
            "request_id": ctx.request_id,
            "type": ctx.request_type,  # "tool", "resource" or "prompt"
            "name": ctx.tool_name,
            # Argument names only: values can hold personal data.
            "args": sorted(ctx.state.get("_tool_arguments", {})),
        }
        try:
            result = await call_next(ctx)
            entry["status"] = "ok"
            return result
        except MCPError as exc:  # refusals and tool errors carry a code
            entry.update(status="error", code=exc.code)
            raise
        except Exception as exc:  # raw exceptions, before exception handlers map them
            entry.update(status="error", code=type(exc).__name__)
            raise
        finally:
            entry["client"] = ctx.client_id  # set by AuthMiddleware, further in
            entry["ms"] = round(1000 * (time.perf_counter() - start), 1)
            with open(self.path, "a") as f:
                f.write(json.dumps(entry) + "\n")

Four calls, then the file:

no key                       AUTHENTICATION_ERROR: Missing API key  [0 ms]
report-bot get_order         {"order_id": "A-1001", "status": "shipped", "total": 42.0}  [202 ms]
report-bot refund_order      ACCESS_DENIED: Requires any of roles [support], but client has [viewer]  [0 ms]
unknown order                ORDER_NOT_FOUND: No order A-9999 (Check the order ID. IDs look like A-1001.)  [201 ms]

audit.jsonl:
{"request_id": "f9d0d6d8cc1b", "type": "tool", "name": "get_order", "args": ["order_id"], "status": "error", "code": "AUTHENTICATION_ERROR", "client": null, "ms": 0.0}
{"request_id": "d363e9d9deec", "type": "tool", "name": "get_order", "args": ["order_id"], "status": "ok", "client": "report-bot", "ms": 202.2}
{"request_id": "24c1566dc3ec", "type": "tool", "name": "refund_order", "args": ["amount", "order_id"], "status": "error", "code": "ACCESS_DENIED", "client": "report-bot", "ms": 0.1}
{"request_id": "81bc1e765953", "type": "tool", "name": "get_order", "args": ["order_id"], "status": "error", "code": "OrderNotFound", "client": "report-bot", "ms": 200.6}

The log has the refused calls with their codes, and the caller's ID even though AuthMiddleware runs inside AuditLog: the finally block reads ctx.client_id after the inner layers set it. A call with no key logs "client": null.

Look at the last line. The client received ORDER_NOT_FOUND, but the audit log says OrderNotFound. Exception handlers map exceptions to errors after the middleware chain, so middleware sees the raw exception. For expected errors, raise a ToolError with a code in the tool, and both will match.

For logs you must be able to trust later, Promptise also ships AuditMiddleware, which chains each entry to the previous one with an HMAC so that edits and deletions are detectable. See AI agent audit logs. For metrics, traces and dashboards, MetricsMiddleware, OTelMiddleware and PrometheusMiddleware follow the same pattern, covered in MCP server monitoring.

Agents retry fast and loop, so a server that agents call needs a rate limit per caller. RateLimitMiddleware keeps a token bucket per client, keyed on ctx.client_id. That only works when it runs after authentication, so the server adds it third, with a burst of five calls and a refill of one per second: RateLimitMiddleware(rate_per_minute=60, burst=5).

Seven quick calls from report-bot, then one from support-bot:

report-bot call 1            {"order_id": "A-1002", "status": "paid", "total": 19.5}  [203 ms]
report-bot call 2            {"order_id": "A-1002", "status": "paid", "total": 19.5}  [0 ms]
report-bot call 3            {"order_id": "A-1002", "status": "paid", "total": 19.5}  [0 ms]
report-bot call 4            {"order_id": "A-1002", "status": "paid", "total": 19.5}  [0 ms]
report-bot call 5            {"order_id": "A-1002", "status": "paid", "total": 19.5}  [0 ms]
report-bot call 6            RATE_LIMIT_EXCEEDED: Rate limit exceeded (Wait 1 second before retrying.)  [0 ms]
report-bot call 7            RATE_LIMIT_EXCEEDED: Rate limit exceeded (Wait 1 second before retrying.)  [0 ms]
support-bot call 1           {"order_id": "A-1002", "status": "paid", "total": 19.5}  [203 ms]

The bucket held five tokens, so calls 6 and 7 were refused with a retryable error that says how long to wait. support-bot has its own bucket and was not affected. Calls 2 to 5 took 0 ms because they were cache hits (the cache from step 5 is already in the stack), and they still used tokens: the limiter sits outside the cache. Put the limiter inside the cache if you only want to limit calls that reach the tool.

Now add the same limiter before AuthMiddleware, as wrong_order.py does:

server log: MCPServer 'orders': RateLimitMiddleware runs before AuthMiddleware for tools ['ping']: no caller is identified yet when it picks a bucket, so every caller shares one 'global' bucket. Add AuthMiddleware first (middleware runs in the order it was added), or pass key_func= to RateLimitMiddleware if one shared bucket is intended.
sk-a: pong
sk-a: pong
sk-a: pong
sk-b: RATE_LIMIT_EXCEEDED

No caller is known when the limiter runs, so all callers share one bucket, and bot-b was refused because bot-a used it up. Promptise logs a warning for this order when the server is built. Per-tool limits (@server.tool(rate_limit="10/min")), tenants and sizing are covered in rate limiting MCP tools per client.

Caching and timeouts go innermost, close to the handler. CacheMiddleware stores results per tool and arguments, and by default per caller. TimeoutMiddleware gives every tool a default time limit. The server adds them fourth and fifth: CacheMiddleware(ttl=30) and TimeoutMiddleware(default_timeout=2.0). Two calls to get_order from report-bot, one from support-bot, then the slow report:

report-bot (miss)            {"order_id": "A-1001", "status": "shipped", "total": 42.0}  [202 ms]
report-bot (hit)             {"order_id": "A-1001", "status": "shipped", "total": 42.0}  [0 ms]
support-bot (own entry)      {"order_id": "A-1001", "status": "shipped", "total": 42.0}  [201 ms]
sales_report                 TIMEOUT: Tool 'sales_report' timed out after 2.0s (The tool did not finish within 2.0s. It may depend on a slow or unavailable service: retry later, or continue without it.)  [2002 ms]

The second call from report-bot was a hit. support-bot got its own entry, because the cache scope defaults to the client. That prevents one caller from receiving data computed for another. refund_order sets cache=False, and tools annotated destructive_hint=True are never cached either. Place CacheMiddleware after AuthMiddleware: if it runs first, it can't tell callers apart, so it skips caching for authenticated tools and logs a warning.

sales_report sleeps for five seconds, so the timeout cut it off at two with a retryable TIMEOUT and a suggestion the model can act on. Because the timeout is the innermost layer, it times only the tool, not the audit log or an approval gate in front of it. For one tool that needs longer, set @server.tool(timeout=30). Circuit breakers for failing upstreams are in MCP server caching, timeouts and circuit breakers.

An agent decides what to do next from the error text, so errors should say what went wrong and whether to retry, without leaking internals. Map your domain exceptions with an exception handler:

@server.exception_handler(OrderNotFound)
async def order_not_found(ctx, exc):
    return ToolError(
        str(exc),
        code="ORDER_NOT_FOUND",
        retryable=False,
        suggestion="Check the order ID. IDs look like A-1001.",
    )

export_orders raises a RuntimeError with a database URL in the message, as a careless library might:

unknown order                ORDER_NOT_FOUND: No order A-9999 (Check the order ID. IDs look like A-1001.)  [203 ms]
server log: Unhandled error in tool 'export_orders' (RuntimeError: connect failed: postgres://admin:s3cret@10.0.0.5/orders)
bug in a tool                INTERNAL_ERROR: An internal error occurred.  [1 ms]
bad argument                 VALIDATION_ERROR: Invalid input: amount: Input should be a valid number, unable to parse string as a number (Check the parameter types and constraints in the tool description.)  [1 ms]

The mapped error carries a code and a suggestion. The unmapped one reached the client as a generic INTERNAL_ERROR, while the server log kept the full message. That default keeps connection strings and file paths out of the model's context. Arguments that fail validation still pass through the middleware chain, so authentication, rate limits and the audit log count them, and the VALIDATION_ERROR names the field without echoing its value.

The full server. Run it with python orders_server.py to serve it over Streamable HTTP on 127.0.0.1:8080:

"""An orders MCP server with a middleware stack: audit, auth, rate limits, caching, timeouts."""
import asyncio
import json
import os
import time

from promptise.mcp.server import (
    APIKeyAuth,
    AuthMiddleware,
    CacheMiddleware,
    MCPError,
    MCPServer,
    RateLimitMiddleware,
    TimeoutMiddleware,
    ToolError,
)

AUDIT_PATH = os.environ.get("AUDIT_PATH", "audit.jsonl")

# In production, load keys from a secret store. Rich keys carry roles for guards.
KEYS = {
    os.environ.get("SUPPORT_KEY", "sk-test-support"): {"client_id": "support-bot", "roles": ["support"]},
    os.environ.get("REPORT_KEY", "sk-test-report"): {"client_id": "report-bot", "roles": ["viewer"]},
}

ORDERS = {
    "A-1001": {"status": "shipped", "total": 42.00},
    "A-1002": {"status": "paid", "total": 19.50},
}


class OrderNotFound(Exception):
    """A domain error raised by the order store."""


class AuditLog:
    """Outermost middleware: one JSON line per call, allowed or refused."""

    def __init__(self, path: str) -> None:
        self.path = path

    async def __call__(self, ctx, call_next):
        start = time.perf_counter()
        entry = {
            "request_id": ctx.request_id,
            "type": ctx.request_type,  # "tool", "resource" or "prompt"
            "name": ctx.tool_name,
            # Argument names only: values can hold personal data.
            "args": sorted(ctx.state.get("_tool_arguments", {})),
        }
        try:
            result = await call_next(ctx)
            entry["status"] = "ok"
            return result
        except MCPError as exc:  # refusals and tool errors carry a code
            entry.update(status="error", code=exc.code)
            raise
        except Exception as exc:  # raw exceptions, before exception handlers map them
            entry.update(status="error", code=type(exc).__name__)
            raise
        finally:
            entry["client"] = ctx.client_id  # set by AuthMiddleware, further in
            entry["ms"] = round(1000 * (time.perf_counter() - start), 1)
            with open(self.path, "a") as f:
                f.write(json.dumps(entry) + "\n")


server = MCPServer(name="orders")
server.add_middleware(AuditLog(AUDIT_PATH))                                 # 1. sees every call
server.add_middleware(AuthMiddleware(APIKeyAuth(keys=KEYS)))                # 2. who is calling
server.add_middleware(RateLimitMiddleware(rate_per_minute=60, burst=5))     # 3. per client, needs 2
server.add_middleware(CacheMiddleware(ttl=30))                              # 4. per client, needs 2
server.add_middleware(TimeoutMiddleware(default_timeout=2.0))               # 5. bounds the handler


@server.exception_handler(OrderNotFound)
async def order_not_found(ctx, exc):
    return ToolError(
        str(exc),
        code="ORDER_NOT_FOUND",
        retryable=False,
        suggestion="Check the order ID. IDs look like A-1001.",
    )


@server.tool(auth=True, read_only_hint=True)
async def get_order(order_id: str) -> dict:
    """Look up one order by ID."""
    await asyncio.sleep(0.2)  # a slow database
    if order_id not in ORDERS:
        raise OrderNotFound(f"No order {order_id}")
    return {"order_id": order_id, **ORDERS[order_id]}


@server.tool(auth=True, roles=["support"], destructive_hint=True, cache=False)
async def refund_order(order_id: str, amount: float) -> dict:
    """Refund part or all of an order. Support staff only."""
    if order_id not in ORDERS:
        raise OrderNotFound(f"No order {order_id}")
    return {"order_id": order_id, "refunded": amount}


@server.tool(auth=True)
async def sales_report(month: str) -> dict:
    """Build the sales report for a month, such as 2026-09."""
    await asyncio.sleep(5)  # far slower than the 2-second timeout
    return {"month": month, "orders": 2}


@server.tool(auth=True)
async def export_orders() -> str:
    """Export all orders as CSV."""
    raise RuntimeError("connect failed: postgres://admin:s3cret@10.0.0.5/orders")


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

Then http_check.py connects a real MCP client over HTTP, MCPClient(url="http://127.0.0.1:8080/mcp", api_key="sk-test-report"), lists the tools and calls get_order twice and refund_order once:

tools: ['get_order', 'refund_order', 'sales_report', 'export_orders']
get_order: {"order_id": "A-1001", "status": "shipped", "total": 42.0}
get_order: {"order_id": "A-1001", "status": "shipped", "total": 42.0}
refund_order: {   "error": {     "code": "ACCESS_DENIED",     "message": "Requires any of roles [support], but client has [viewer]",     "retryable": false,     "details": {       "guard": "HasRole",       "tool": "refund_order"     }   } }

The second get_order came from the cache, and the audit file got three lines, one per tool call. tools/list is not in the audit log, because Promptise middleware doesn't run for listing requests.

FastMCP has had middleware since version 2.9. You subclass Middleware and override hooks. A request passes through on_message, then on_request, then an operation hook such as on_call_tool, on_read_resource or on_list_tools. Middleware runs in the order it was added, like in Promptise. The same audit idea with FastMCP 4.1.2's built-in error handling and rate limiting:

"""The same idea in FastMCP 4: a custom audit hook plus built-in error handling and rate limits."""
import asyncio
import logging
import sys

from fastmcp import Client, FastMCP
from fastmcp.exceptions import ToolError
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.server.middleware.error_handling import ErrorHandlingMiddleware
from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware

logging.basicConfig(stream=sys.stdout, level=logging.ERROR, format="server log: %(message)s")
logging.getLogger("fastmcp").setLevel(logging.CRITICAL)  # keep the demo output short


class AuditLog(Middleware):
    """Runs for every message; prints the method, and for tool calls the outcome."""

    async def on_message(self, context: MiddlewareContext, call_next):
        print(f"audit  -> {context.method}")
        return await call_next(context)

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        name, args = context.message.name, sorted(context.message.arguments or {})
        try:
            result = await call_next(context)
            print(f"audit  <- {name}{args} ok")
            return result
        except Exception as exc:
            print(f"audit  <- {name}{args} {type(exc).__name__}: {exc}")
            raise


mcp = FastMCP("orders")
mcp.add_middleware(ErrorHandlingMiddleware(logger=logging.getLogger("errors")))  # first in, last out
mcp.add_middleware(RateLimitingMiddleware(max_requests_per_second=0.2, burst_capacity=4))
mcp.add_middleware(AuditLog())


@mcp.tool
async def get_order(order_id: str) -> dict:
    """Look up one order by ID."""
    if order_id != "A-1001":
        raise ToolError(f"No order {order_id}")
    return {"order_id": order_id, "status": "shipped"}


async def main() -> None:
    async with Client(mcp) as client:  # in-memory transport, same middleware chain
        print("tools:", [t.name for t in await client.list_tools()])
        for order_id in ("A-1001", "A-9999", "A-1001"):
            try:
                result = await client.call_tool("get_order", {"order_id": order_id})
                print("client:", result.data)
            except Exception as exc:
                print(f"client: {type(exc).__name__}: {exc}")


asyncio.run(main())

audit  -> server/discover
audit  -> tools/list
tools: ['get_order']
audit  -> tools/call
audit  <- get_order['order_id'] ok
client: {'order_id': 'A-1001', 'status': 'shipped'}
audit  -> tools/call
audit  <- get_order['order_id'] ToolError: No order A-9999
server log: Error in tools/call: ToolError: No order A-9999
client: MCPError: Internal error: No order A-9999
server log: Error in tools/call: RateLimitError: Rate limit exceeded for client: global
client: MCPError: Rate limit exceeded for client: global

Four things in this output are worth knowing before you copy the pattern:

• FastMCP rate limiting counts every request. RateLimitingMiddleware hooks on_request, so the discovery and tools/list requests used two of the four tokens, and the third tool call was refused.

• Per-client limits need a client ID. Without get_client_id, every caller shares the global bucket, so one busy agent can use up the limit for everyone. Pass a function that takes the middleware context and returns a key, such as the client ID from the caller's access token.

• The refused call never reached the audit hook. The limiter was added before AuditLog, so it is outside it. Add your audit middleware first if it must record refusals.

• `ErrorHandlingMiddleware` changed the tool error. With its default transform_errors=True, the tool's ToolError reached the client as MCPError: Internal error: No order A-9999, a protocol error, instead of a tool error. With transform_errors=False, the same run printed client: ToolError: No order A-9999.

FastMCP's hooks reach further: you can filter tool lists, act on initialize, or observe notifications. Promptise gives you more governance out of the box and checks the most common ordering mistakes. If you are choosing a server framework, see FastMCP alternatives.

Use a standard order

A stack that works for most servers, outermost first: audit or request logging, authentication, rate limiting, concurrency limits, caching, circuit breaker, timeout, then guards and the handler. Error shaping belongs at the edges: Promptise's exception handlers already run outside the chain, and in FastMCP the error-handling middleware goes first.

Share state across replicas

Token buckets, caches and circuit-breaker counters live in process memory. With three replicas behind a load balancer, each client gets three times the limit. Use CacheMiddleware(backend=RedisCache(url=...)) for a shared cache, pass RateLimitMiddleware a limiter backed by shared storage, or enforce limits at a gateway. Every production concern of a server is collected in building a production MCP server.

Scope middleware to routers and request types

Not every rule belongs on every tool. An MCPRouter can carry its own middleware, which runs after the server's middleware and only for that router's tools, for example a tighter timeout for a group of database tools. Inside one middleware, branch on ctx.request_type ("tool", "resource" or "prompt") when a rule only makes sense for tool calls. The routers and middleware docs show both.

Keep middleware fast and safe

Promptise compiles each tool's middleware chain once at start-up, so the chain itself costs little. What you do inside it is the cost. Avoid a database or network call per request in middleware unless you cache it. Never log argument values or results by default, because tool inputs often hold personal data and tokens. Log names, sizes and codes instead.

Test the order

Ordering bugs are silent. Write tests that call the server through TestClient and assert the behavior you rely on: an unauthenticated call is refused, two clients have separate buckets, a refused call is in the audit log. Testing an MCP server shows the patterns.

Middleware is a framework feature, not part of the MCP specification, so middleware written for FastMCP doesn't run on a Promptise server or the reverse. Promptise middleware doesn't see listing or session requests. For per-caller tool lists you use hide_unauthorized_tools, and connection-level rules belong in the HTTP layer or a gateway. The built-in rate limiter and in-memory cache are per process. Middleware also only sees requests that reach the MCP layer: with require_auth=True on an HTTP transport, a request with a bad token is rejected with 401 before that, so it appears in the HTTP access log, not in your audit middleware. And middleware enforces rules around tools; it doesn't make an unsafe tool safe. Keep destructive tools narrow and require approval for them.

What is MCP middleware?

MCP middleware is code that wraps every request an MCP server handles. Each layer gets the request context and a call_next function, so it can act before the tool runs, act after it, or refuse the call by raising an error. Servers use it for authentication, audit logging, rate limits, caching, timeouts and error handling. It is a feature of server frameworks, not of the MCP protocol.

How do you add rate limiting to an MCP server?

Add a rate-limiting middleware that keeps a token bucket per caller, and place it after authentication so it knows who is calling. In Promptise use RateLimitMiddleware(rate_per_minute=..., burst=...) or a per-tool rate_limit="10/min". In FastMCP use RateLimitingMiddleware with get_client_id, keeping in mind that it counts every request, including tool listing. Return a retryable error that says how long to wait.

How do you audit log MCP tool calls?

Register an audit middleware as the outermost layer so it records every call, including refused ones. Log the request ID, tool name, caller, status, error code and duration in a finally block. Log argument names, not values, because tool inputs often contain personal data. For tamper-evident logs, chain entries with an HMAC, as Promptise's AuditMiddleware does.

In what order should MCP middleware run?

Outermost first: logging or audit, then authentication, then rate and concurrency limits, then caching, then timeouts closest to the tool. Authentication must come before anything that keys on the caller, such as per-client rate limits and caches. In both FastMCP and Promptise, the middleware added first runs first on the way in and last on the way out.

Is middleware part of the MCP specification?

No. The MCP specification defines the messages between clients and servers, not how a server processes them. FastMCP's documentation says its middleware is specific to FastMCP. Promptise, FastMCP and the official Python SDK each have their own middleware API, so middleware doesn't move between frameworks without changes.

Can middleware change tool arguments or results?

Yes. A middleware can rewrite the request before calling call_next or transform the result it returns. Use this with care: rewriting arguments hides what the model asked for, and changing results can confuse the agent. Good uses are redacting secrets from results, truncating oversized responses and adding metadata. Validation of arguments belongs in the tool's schema.

• Rate limit each client and tenant in depth: LLM tool rate limiting per client.

• Add metrics, traces and a dashboard: MCP server monitoring.

• Replace API keys with JWT or OAuth: MCP server authentication.

• Put it together: build a production MCP server.

• Review the security side: MCP security best practices.

• Reference: routers and middleware in the Promptise docs.