Middleware

Middleware lets you intercept MCP requests to add logging, authentication, rate limiting, validation, or any cross-cutting logic.

Quick Start

from mcp_use.server import MCPServer
from mcp_use.server.middleware import Middleware

class LoggingMiddleware(Middleware):
    async def on_request(self, context, call_next):
        print(f"→ {context.method}")
        result = await call_next(context)
        print(f"← {context.method}")
        return result

server = MCPServer(
    name="my-server",
    middleware=[LoggingMiddleware()]
)

How It Works

Middleware executes in an onion model: each middleware wraps the next, with the handler at the center.

Request
  → Middleware A (before)
    → Middleware B (before)
      → Handler
    ← Middleware B (after)
  ← Middleware A (after)
Response

Each middleware can:

Inspect/modify the request before calling call_next
Inspect/modify the response after call_next returns
Short-circuit by returning early without calling call_next
Reject by raising an exception

Hooks

Override these methods to intercept specific request types:

Hook	When it runs
`on_request`	Every request (wraps all other hooks)
`on_initialize`	Client connection handshake
`on_call_tool`	Tool execution
`on_read_resource`	Resource reads
`on_get_prompt`	Prompt retrieval
`on_list_tools`	Tool listing
`on_list_resources`	Resource listing
`on_list_prompts`	Prompt listing
`on_set_logging_level`	Client sets minimum log level
`on_complete`	Completion/autocomplete requests

Typed context: Each hook receives a fully-typed context.message. For example, on_initialize gets ServerMiddlewareContext[InitializeRequestParams], so your editor knows exactly what fields are available (like context.message.clientInfo.name). No guessing, full autocomplete.

Hook nesting

When you override both on_request and a specific hook, they nest: on_request wraps the specific hook.

class MyMiddleware(Middleware):
    async def on_request(self, context, call_next):
        print("1. on_request before")
        result = await call_next(context)  # Calls on_call_tool (if tool request)
        print("4. on_request after")
        return result

    async def on_call_tool(self, context, call_next):
        print("2. on_call_tool before")
        result = await call_next(context)  # Calls handler
        print("3. on_call_tool after")
        return result

Use on_request for logic that applies to all requests. Use specific hooks when you only care about certain operations.

Context

Every hook receives a ServerMiddlewareContext with:

Field	Type	Description
`message`	Typed params	Request parameters (e.g., `CallToolRequestParams`)
`method`	`str`	MCP method name (e.g., `"tools/call"`)
`session_id`	`str \| None`	Client session ID (from `mcp-session-id` header)
`transport`	`str`	Transport type (`"stdio"`, `"streamable-http"`)
`timestamp`	`datetime`	Request timestamp
`headers`	`dict \| None`	HTTP headers (HTTP transports only)
`client_ip`	`str \| None`	Client IP (HTTP transports only)
`metadata`	`dict`	Custom data passed between middleware

Context is immutable. Use context.copy() to pass data downstream:

enriched = context.copy(metadata={**context.metadata, "user_id": "123"})
return await call_next(enriched)

Examples

Authentication
Rate Limiting
Connection Guard

Reject requests without a valid API key:

class AuthMiddleware(Middleware):
    async def on_call_tool(self, context, call_next):
        api_key = context.headers.get("x-api-key") if context.headers else None
        if not api_key or api_key != "secret":
            raise PermissionError("Invalid API key")
        return await call_next(context)

Limit requests per session:

from collections import defaultdict
from datetime import datetime

class RateLimitMiddleware(Middleware):
    def __init__(self, max_per_minute: int = 30):
        self.max = max_per_minute
        self.requests: dict[str, list[datetime]] = defaultdict(list)

    async def on_call_tool(self, context, call_next):
        sid = context.session_id or "anonymous"
        now = datetime.now()
        # Keep only requests from last minute
        self.requests[sid] = [
            t for t in self.requests[sid]
            if (now - t).total_seconds() < 60
        ]
        if len(self.requests[sid]) >= self.max:
            raise Exception("Rate limit exceeded")
        self.requests[sid].append(now)
        return await call_next(context)

Reject clients during the MCP handshake:

class ConnectionGuard(Middleware):
    async def on_initialize(self, context, call_next):
        client = context.message.clientInfo.name
        if client in ["blocked-client"]:
            raise ValueError(f"Client {client} not allowed")
        print(f"Connection from: {client}")
        return await call_next(context)

Middleware Order

Order matters. Middleware runs in the order added, with earlier middleware wrapping later ones.

server = MCPServer(
    middleware=[
        LoggingMiddleware(),      # 1. Outermost - sees all requests
        AuthMiddleware(),         # 2. Rejects unauthorized early
        RateLimitMiddleware(),    # 3. Limits request rate
        ValidationMiddleware(),   # 4. Innermost - validates data
    ]
)

Recommended order: Logging → Authentication → Rate limiting → Validation. This ensures logging sees all requests (including rejected ones) and auth rejects early before expensive operations.

Best Practices

Single responsibility: Each middleware does one thing
Fail fast: Reject invalid requests early, before expensive operations
Always call call_next: Unless intentionally short-circuiting
Re-raise exceptions: If you catch errors to log them, always re-raise

async def on_request(self, context, call_next):
    try:
        return await call_next(context)
    except Exception as e:
        print(f"Request failed: {e}")
        raise  # Always re-raise

Full Example

middleware_example.py

Complete working server with logging, auth, rate limiting, and validation middleware.

Get Started

MCP Server

MCP Client

MCP Agents

Development

Quick Start

How It Works

Hooks

Hook nesting

Context

Examples

Middleware Order

Best Practices

Full Example

middleware_example.py

​Quick Start

​How It Works

​Hooks

​Hook nesting

​Context

​Examples

​Middleware Order

​Best Practices

​Full Example

middleware_example.py

Quick Start

How It Works

Hooks

Hook nesting

Context

Examples

Middleware Order

Best Practices

Full Example