Middleware

Eine Middleware ist eine einzelne async-Funktion, die jede Nachricht umschließt, die dein Server empfängt.

Du schreibst sie als async (ctx, call_next) und hängst sie an server.middleware an. Das ist die ganze API.

MCPServer nimmt die Liste bei der Konstruktion entgegen (MCPServer(name, middleware=[...])) und stellt sie als mcp.middleware bereit; der Low-Level-Server stellt dieselbe Liste als server.middleware bereit. Das Beispiel unten verwendet den Low-Level-Server; wenn Server(name, on_call_tool=...) neu für dich ist, lies zuerst Der Low-Level-Server.

Eine Timing-Middleware

Ein Server, ein Tool, eine Middleware, die loggt, wie lange jede Nachricht gedauert hat:

import logging
import time

from mcp.server import Server, ServerRequestContext
from mcp.server.context import CallNext, HandlerResult
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

logger = logging.getLogger(__name__)


async def on_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(
        tools=[
            Tool(
                name="search_books",
                description="Search the catalog by title or author.",
                input_schema={
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
            )
        ]
    )


async def on_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    query = (params.arguments or {})["query"]
    return CallToolResult(content=[TextContent(type="text", text=f"Found 3 books matching {query!r}.")])


async def log_timing(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult:
    start = time.perf_counter()
    try:
        return await call_next(ctx)
    finally:
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.info("%s took %.1f ms", ctx.method, elapsed_ms)


server = Server("Bookshop", on_list_tools=on_list_tools, on_call_tool=on_call_tool)
server.middleware.append(log_timing)

• ctx ist derselbe ServerRequestContext, den deine Handler erhalten. ctx.method ist der rohe Methoden-String; ctx.params sind die rohen Params, vor jeder Validierung.
• call_next(ctx) führt den Rest der Kette aus: Validierung, die Handler-Suche, deinen Handler. Gib zurück, was es zurückgegeben hat, und die Response bleibt unverändert.
• Das try/finally ist Absicht: Ein Handler, der eine Exception auslöst, wird trotzdem gemessen, denn der Fehlschlag erreicht deine Middleware als Exception aus call_next.
• server.middleware.append(...) registriert sie. Die Liste läuft von außen nach innen, also ist middleware[0] diejenige, die am nächsten an der Leitung sitzt.

Ausprobieren

Verbinde einen Client, liste die Tools auf, rufe eines auf. Dein Log hat drei Zeilen:

server/discover took 18.3 ms
tools/list took 0.1 ms
tools/call took 0.1 ms

Du hast zwei Aufrufe gemacht und drei Zeilen bekommen. Die erste ist server/discover: der Request, den der Client zum Aufbau der Verbindung geschickt hat, bevor du irgendetwas angefordert hast.

Genau darum geht es. Middleware umschließt jede eingehende Nachricht:

• Den Verbindungsaufbau: server/discover, oder initialize und notifications/initialized in einer Legacy-Session.
• Jeden Request und jede Benachrichtigung. Bei einer Benachrichtigung gilt ctx.request_id is None, call_next(ctx) gibt None zurück, und was immer du zurückgibst, wird verworfen.
• Sogar eine Methode, für die der Server keinen Handler hat: call_next wirft den MCPError(-32601, "Method not found") durch deine Middleware hindurch auf dem Weg zum Client.

Was du in einer Middleware tun kannst

In aufsteigender Reihenfolge danach, wie sehr du zögern solltest:

• Beobachten. Miss es, zähle es, logge es. Das Beispiel oben.
• Ablehnen. Wirf einen MCPError statt call_next(ctx) aufzurufen, und diese eine Nachricht wird mit einem JSON-RPC-Fehler beantwortet. Die Verbindung bleibt bestehen; die nächste Nachricht geht durch. So beschränkt ein Server subscriptions/listen pro Aufrufer: Entscheiden, wer zusehen darf auf der Seite Abonnements führt es Schritt für Schritt vor.
• Umschreiben. ctx ist eine Dataclass: await call_next(dataclasses.replace(ctx, params=...)) reicht dem Rest der Kette andere Params weiter, als der Client geschickt hat. Tu das nie mit initialize: Das Ergebnis, das der Client zurückbekommt, wird aus deinen umgeschriebenen Params gebaut, aber der Server legt seinen Verbindungszustand anhand der ursprünglichen Params von der Leitung fest. Beide Seiten können den Handshake beenden und sich dabei uneinig sein, was sie ausgehandelt haben.
• Antworten. Gib ein Ergebnis zurück, ohne call_next(ctx) aufzurufen, und es geht als deine Response an den Client. call_next reicht dir die fertige Form für die Leitung, und die Pipeline bessert nie nach, was du zurückgibst – der ganze Umschlag gehört also dir: Auf einer Verbindung der 2026er-Generation gehört dazu der serverInfo-Stempel in _meta, den das SDK an Handler-Ergebnisse anfügt, an deine aber nicht.

ValueError: 'initialize' is handled by the server runner and cannot be overridden;
use Server.middleware to observe or wrap initialization

Die eine Middleware, die standardmäßig aktiv ist

Das SDK liefert genau eine Middleware mit, und sie steht bereits auf der Liste deines Servers: die, die für jede Nachricht einen OpenTelemetry-Span ausgibt. Du hängst sie nicht an, und meistens denkst du gar nicht an sie. Sie tut nichts, bis du einen Exporter installierst, und sie hat ihre eigene Seite: OpenTelemetry.

Zusammenfassung

• Eine Middleware ist async (ctx, call_next) -> result, übergeben als MCPServer(middleware=[...]) (oder an mcp.middleware angehängt) und beim Low-Level-Server an server.middleware angehängt.
• Sie umschließt jede eingehende Nachricht (server/discover, initialize, Requests, Benachrichtigungen, unbekannte Methoden) und läuft von außen nach innen.
• An ctx.request_id is None unterscheidest du eine Benachrichtigung von einem Request.
• Wirf eine Exception, statt call_next aufzurufen, um eine einzelne Nachricht abzulehnen; die Verbindung überlebt.
• Das OpenTelemetry-Tracing des SDK ist ebenfalls eine Middleware und steht schon auf der Liste. Siehe OpenTelemetry.
• Die ganze Oberfläche ist provisorisch. Beobachte damit; baue nicht darauf.

Das ist alles, was einen Request umschließt. Autorisierung entscheidet, ob der Request überhaupt laufen darf.