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 beim Erzeugen 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 gebraucht 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 Parameter, vor jeder Validierung.
• call_next(ctx) führt den Rest der Kette aus: Validierung, die Suche nach dem Handler, deinen Handler. Gib zurück, was es zurückgegeben hat, und die Response bleibt unverändert.
• Das try/finally ist Absicht: Auch ein Handler, der eine Exception auslöst, wird gemessen, denn der Fehler 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] die Middleware, die der Leitung am nächsten ist.

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 geschickt hat, um die Verbindung aufzubauen, bevor du irgendetwas angefordert hast.

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

• Den Verbindungsaufbau: server/discover, oder initialize und notifications/initialized auf 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 innerhalb einer Middleware möglich ist

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

• Beobachten. Messen, zählen, loggen. Das Beispiel oben.
• Ablehnen. Löse einen MCPError aus, 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 schränkt ein Server subscriptions/listen pro Aufrufer ein: Entscheiden, wer zusehen darf auf der Seite Abonnements geht das Schritt für Schritt durch.
• Umschreiben. ctx ist eine Dataclass: await call_next(dataclasses.replace(ctx, params=...)) reicht dem Rest der Kette andere Parameter weiter, als der Client geschickt hat. Tu das niemals bei initialize: Das Ergebnis, das der Client zurückbekommt, wird aus deinen umgeschriebenen Parametern gebaut, aber der Server legt seinen Verbindungszustand anhand der ursprünglichen Parameter von der Leitung fest. Beide Seiten können den Handshake abschließen und sich dabei uneinig sein, was sie ausgehandelt haben.
• Beantworten. 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 schließt das den _meta-Stempel serverInfo ein, den das SDK an Handler-Ergebnisse anhängt, 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 erzeugt. 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.
• Löse eine Exception aus, statt call_next aufzurufen, um eine einzelne Nachricht abzulehnen; die Verbindung überlebt.
• Das eigene OpenTelemetry-Tracing des SDK ist ebenfalls eine Middleware, die schon auf der Liste steht. Siehe OpenTelemetry.
• Die gesamte 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.