> ## Documentation Index
> Fetch the complete documentation index at: https://mcp-use.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

<AgentInstructions>

## Submitting Feedback

If you encounter incorrect, outdated, or confusing documentation on this page, submit feedback:

POST https://mcp-use.com/docs/feedback

```json
{
  "path": "/python/server/middleware",
  "feedback": "Description of the issue"
}
```

Only submit feedback when you have something specific and actionable to report.

</AgentInstructions>

# Middleware

> Intercept and process MCP requests with middleware

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

## Quick Start

```python theme={null}
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      |

<Tip>
  **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.
</Tip>

### Hook nesting

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

```python theme={null}
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
```

<Tip>
  Use `on_request` for logic that applies to all requests. Use specific hooks when you only care about certain operations.
</Tip>

## 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              |

<Note>
  Context is immutable. Use `context.copy()` to pass data downstream:

  ```python theme={null}
  enriched = context.copy(metadata={**context.metadata, "user_id": "123"})
  return await call_next(enriched)
  ```
</Note>

## Examples

<Tabs>
  <Tab title="Authentication">
    Reject requests without a valid API key:

    ```python theme={null}
    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)
    ```
  </Tab>

  <Tab title="Rate Limiting">
    Limit requests per session:

    ```python theme={null}
    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)
    ```
  </Tab>

  <Tab title="Connection Guard">
    Reject clients during the MCP handshake:

    ```python theme={null}
    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)
    ```
  </Tab>
</Tabs>

## Middleware Order

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

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

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

## 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

```python theme={null}
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

<Card title="middleware_example.py" icon="github" href="https://github.com/mcp-use/mcp-use/blob/main/libraries/python/examples/server/middleware_example.py">
  Complete working server with logging, auth, rate limiting, and validation middleware.
</Card>