Middleware

Um middleware é uma única função async que envolve toda mensagem que o seu servidor recebe.

Você o escreve como async (ctx, call_next) e o adiciona ao fim de server.middleware. A API inteira é essa.

MCPServer recebe a lista na construção (MCPServer(name, middleware=[...])) e a expõe como mcp.middleware; o Server de baixo nível expõe a mesma lista como server.middleware. O exemplo abaixo usa o Server de baixo nível; se Server(name, on_call_tool=...) é novidade para você, leia O Server de baixo nível primeiro.

Um middleware de medição de tempo

Um servidor, uma ferramenta, um middleware que registra no log quanto tempo cada mensagem levou:

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 é o mesmo ServerRequestContext que os seus handlers recebem. ctx.method é a string bruta do método; ctx.params são os params brutos, antes de qualquer validação.
• call_next(ctx) executa o restante da cadeia: a validação, a busca do handler, o seu handler. Retorne o que ele retornou e a resposta fica intacta.
• O try/finally é proposital: um handler que lança exceção ainda é cronometrado, porque a falha chega ao seu middleware como a exceção que sai de call_next.
• server.middleware.append(...) faz o registro. A lista executa do mais externo para o mais interno, então middleware[0] é o que fica mais perto do fio.

Experimente

Conecte um cliente, liste as ferramentas, chame uma. O seu log tem três linhas:

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

Você fez duas chamadas e recebeu três linhas. A primeira é server/discover: a requisição que o cliente enviou para estabelecer a conexão, antes de você pedir qualquer coisa.

É justamente esse o ponto. O middleware envolve toda mensagem de entrada:

• O estabelecimento da conexão: server/discover, ou initialize e notifications/initialized em uma sessão legada.
• Toda requisição e toda notificação. Para uma notificação, ctx.request_id is None, call_next(ctx) retorna None e o que quer que você retorne é descartado.
• Até um método para o qual o servidor não tem handler: call_next lança o MCPError(-32601, "Method not found") através do seu middleware a caminho do cliente.

O que você pode fazer dentro de um

Em ordem crescente do quanto você deveria hesitar:

• Observar. Cronometre, conte, registre no log. O exemplo acima.
• Recusar. Lance um MCPError em vez de chamar call_next(ctx) e essa única mensagem é respondida com um erro JSON-RPC. A conexão continua de pé; a próxima mensagem passa. É assim que um servidor controla o acesso a subscriptions/listen por chamador: Decidindo quem pode observar, na página de Assinaturas, percorre o passo a passo.
• Reescrever. ctx é uma dataclass: await call_next(dataclasses.replace(ctx, params=...)) entrega ao restante da cadeia params diferentes dos que o cliente enviou. Nunca faça isso com initialize: o resultado que o cliente recebe de volta é construído a partir dos seus params reescritos, mas o servidor grava o estado da conexão a partir dos params originais do fio. Os dois lados podem terminar o handshake discordando sobre o que negociaram.
• Responder. Retorne um resultado sem chamar call_next(ctx) e ele vai para o cliente como a sua resposta. call_next entrega a você a forma final do fio, e o pipeline nunca altera o que você retorna, então o envelope inteiro é seu: em uma conexão da era 2026 isso inclui o carimbo _meta de serverInfo, que o SDK adiciona aos resultados dos handlers, mas não aos seus.

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

O único middleware que já vem ligado por padrão

O SDK traz exatamente um middleware, e ele já está na lista do seu servidor: o que emite um span do OpenTelemetry para cada mensagem. Você não o adiciona, e na maior parte do tempo nem pensa nele. Ele é um no-op até você instalar um exportador, e tem a própria página: OpenTelemetry.

Recapitulando

• Um middleware é async (ctx, call_next) -> result, passado como MCPServer(middleware=[...]) (ou adicionado a mcp.middleware) e adicionado a server.middleware no Server de baixo nível.
• Ele envolve toda mensagem de entrada (server/discover, initialize, requisições, notificações, métodos desconhecidos) e executa do mais externo para o mais interno.
• ctx.request_id is None é como você distingue uma notificação de uma requisição.
• Lance uma exceção em vez de chamar call_next para recusar uma mensagem; a conexão sobrevive.
• O tracing do OpenTelemetry do próprio SDK também é um middleware, já na lista. Veja OpenTelemetry.
• Toda essa superfície é provisória. Observe com ela; não construa em cima dela.

Isso é tudo o que envolve uma requisição. Autorização é o que decide se a requisição chega a ser executada.