Middleware

Un middleware est une fonction asynchrone qui enveloppe chaque message que votre serveur reçoit.

Vous l’écrivez sous la forme async (ctx, call_next) et vous l’ajoutez à server.middleware. C’est toute l’API.

MCPServer reçoit la liste à la construction (MCPServer(name, middleware=[...])) et l’expose sous mcp.middleware ; le Server bas niveau expose la même liste sous server.middleware. L’exemple ci-dessous utilise le Server bas niveau ; si Server(name, on_call_tool=...) est nouveau pour vous, lisez d’abord Le Server bas niveau.

Un middleware de chronométrage

Un serveur, un outil, un middleware qui journalise le temps pris par chaque message :

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 est le même ServerRequestContext que celui que reçoivent vos gestionnaires (handlers). ctx.method est la chaîne de méthode brute ; ctx.params contient les paramètres bruts, avant toute validation.
• call_next(ctx) exécute le reste de la chaîne : la validation, la recherche du gestionnaire, votre gestionnaire. Renvoyez ce qu’il a renvoyé et la réponse reste intacte.
• Le try/finally est délibéré : un gestionnaire qui lève une exception est tout de même chronométré, car l’échec atteint votre middleware sous la forme de l’exception qui sort de call_next.
• server.middleware.append(...) l’enregistre. La liste s’exécute de l’extérieur vers l’intérieur, donc middleware[0] est celui qui est le plus proche de la liaison.

Essayer

Connectez un client, listez les outils, appelez-en un. Votre journal contient trois lignes :

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

Vous avez fait deux appels et obtenu trois lignes. La première est server/discover : la requête que le client a envoyée pour établir la connexion, avant que vous ne demandiez quoi que ce soit.

C’est tout l’intérêt. Le middleware enveloppe chaque message entrant :

• La mise en place de la connexion : server/discover, ou initialize et notifications/initialized sur une session historique.
• Chaque requête et chaque notification. Pour une notification, ctx.request_id is None, call_next(ctx) renvoie None, et tout ce que vous renvoyez est ignoré.
• Même une méthode pour laquelle le serveur n’a pas de gestionnaire : call_next lève MCPError(-32601, "Method not found") à travers votre middleware en route vers le client.

Ce que vous pouvez y faire

Du geste le plus anodin à celui devant lequel vous devriez le plus hésiter :

• Observer. Chronométrer, compter, journaliser. C’est l’exemple ci-dessus.
• Refuser. Levez une MCPError au lieu d’appeler call_next(ctx) et ce message-là reçoit pour réponse une erreur JSON-RPC. La connexion reste ouverte ; le message suivant passe. C’est ainsi qu’un serveur contrôle l’accès à subscriptions/listen appelant par appelant : la section Décider qui peut observer de la page Abonnements détaille la démarche.
• Réécrire. ctx est une dataclass : await call_next(dataclasses.replace(ctx, params=...)) transmet au reste de la chaîne d’autres paramètres que ceux envoyés par le client. Ne faites jamais cela pour initialize : le résultat que le client reçoit en retour est construit à partir de vos paramètres réécrits, mais le serveur fixe l’état de sa connexion à partir des paramètres d’origine reçus sur la liaison. Les deux côtés peuvent terminer la poignée de main (handshake) en désaccord sur ce qu’ils ont négocié.
• Répondre. Renvoyez un résultat sans appeler call_next(ctx) et il part au client comme votre réponse. call_next vous remet la forme finale telle qu’elle circule sur la liaison, et le pipeline ne retouche jamais ce que vous renvoyez ; toute l’enveloppe est donc à votre charge : sur une connexion de génération 2026, cela inclut l’estampille _meta serverInfo, que le SDK ajoute aux résultats des gestionnaires mais pas aux vôtres.

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

Le seul middleware activé par défaut

Le SDK fournit exactement un middleware, et il figure déjà dans la liste de votre serveur : celui qui émet un span OpenTelemetry pour chaque message. Vous ne l’ajoutez pas et, la plupart du temps, vous n’y pensez pas. Il ne fait rien tant que vous n’installez pas d’exporteur, et il a sa propre page : OpenTelemetry.

Récapitulatif

• Un middleware est async (ctx, call_next) -> result, passé via MCPServer(middleware=[...]) (ou ajouté à mcp.middleware), et ajouté à server.middleware sur le Server bas niveau.
• Il enveloppe chaque message entrant (server/discover, initialize, requêtes, notifications, méthodes inconnues) et s’exécute de l’extérieur vers l’intérieur.
• ctx.request_id is None est ce qui distingue une notification d’une requête.
• Levez une exception au lieu d’appeler call_next pour refuser un message ; la connexion survit.
• Le traçage OpenTelemetry du SDK est lui aussi un middleware, déjà dans la liste. Voir OpenTelemetry.
• Toute cette surface est provisoire. Servez-vous-en pour observer ; ne construisez pas dessus.

C’est tout ce qui enveloppe une requête. Quant à savoir si la requête a seulement le droit de s’exécuter, c’est l’Autorisation qui en décide.