Middleware

Middleware (промежуточный слой) — это одна асинхронная функция, которая оборачивает каждое сообщение, приходящее на сервер.

Её пишут в виде async (ctx, call_next) и добавляют в server.middleware. Вот и весь API.

MCPServer принимает список при создании (MCPServer(name, middleware=[...])) и предоставляет его как mcp.middleware; низкоуровневый Server предоставляет тот же список как server.middleware. В примере ниже используется низкоуровневый Server; если конструкция Server(name, on_call_tool=...) вам незнакома, сначала прочитайте Низкоуровневый Server.

Middleware для замера времени

Один сервер, один инструмент, один слой middleware, который пишет в лог, сколько заняло каждое сообщение:

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 — тот же ServerRequestContext, который получают обработчики. ctx.method — строка метода как есть; ctx.params — параметры как есть, до какой-либо валидации.
• call_next(ctx) запускает остаток цепочки: валидацию, поиск обработчика, сам обработчик. Верните то, что вернул он, — и ответ останется нетронутым.
• try/finally здесь намеренно: обработчик, выбросивший исключение, всё равно замеряется, потому что сбой доходит до middleware в виде исключения из call_next.
• server.middleware.append(...) регистрирует его. Список выполняется начиная с внешнего слоя, так что middleware[0] — ближайший к сети.

Попробуйте сами

Подключите клиент, запросите список инструментов, вызовите один из них. В логе три строки:

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

Вызовов было два, а строк три. Первая — server/discover: запрос, который клиент отправил, чтобы установить подключение, ещё до того, как вы что-либо запросили.

В этом и суть. Middleware оборачивает каждое входящее сообщение:

• Установку подключения: server/discover или, в сессии старого поколения, initialize и notifications/initialized.
• Каждый запрос и каждое уведомление. Для уведомления ctx.request_id is None, call_next(ctx) возвращает None, а всё, что вернёте вы, отбрасывается.
• Даже метод, для которого у сервера нет обработчика: call_next выбрасывает MCPError(-32601, "Method not found") сквозь middleware по пути к клиенту.

Что можно делать внутри

В порядке возрастания того, насколько стоит задуматься, прежде чем это делать:

• Наблюдать. Замерять, считать, логировать. Пример выше.
• Отклонять. Выбросьте MCPError вместо вызова call_next(ctx) — и на это одно сообщение придёт ответ с ошибкой JSON-RPC. Подключение не рвётся; следующее сообщение проходит. Именно так сервер ограничивает subscriptions/listen для каждого вызывающего: раздел Кому разрешено наблюдать на странице о подписках разбирает это пошагово.
• Переписывать. ctx — это dataclass: await call_next(dataclasses.replace(ctx, params=...)) передаёт остальной цепочке не те параметры, что прислал клиент. Никогда не делайте этого с initialize: результат, который получает клиент, строится из переписанных параметров, но состояние подключения сервер фиксирует по исходным параметрам из сети. Стороны могут завершить рукопожатие, расходясь в том, о чём они договорились.
• Отвечать. Верните результат, не вызывая call_next(ctx), — и он уйдёт клиенту как ваш ответ. call_next отдаёт готовую сетевую форму, а конвейер никогда не правит то, что вы возвращаете, так что вся обёртка целиком на вас: на подключении поколения 2026 сюда входит отметка serverInfo в _meta, которую SDK добавляет к результатам обработчиков, но не к вашим.

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

Единственный слой middleware, включённый по умолчанию

SDK поставляет ровно один слой middleware, и он уже в списке вашего сервера: тот, что создаёт спан OpenTelemetry для каждого сообщения. Его не нужно добавлять, и чаще всего о нём не приходится думать. Пока не установлен экспортёр, он ничего не делает, и у него есть своя страница: OpenTelemetry.

Итоги

• Middleware — это async (ctx, call_next) -> result; его передают как MCPServer(middleware=[...]) (или добавляют в mcp.middleware), а в низкоуровневом Server добавляют в server.middleware.
• Middleware оборачивает каждое входящее сообщение (server/discover, initialize, запросы, уведомления, неизвестные методы) и выполняется начиная с внешнего слоя.
• ctx.request_id is None — так уведомление отличают от запроса.
• Чтобы отклонить одно сообщение, выбросьте исключение вместо вызова call_next; подключение это переживёт.
• Собственная трассировка OpenTelemetry в SDK — тоже middleware, уже в списке. См. OpenTelemetry.
• Весь этот интерфейс предварительный. Наблюдайте с его помощью; не стройте на нём.

Это всё, что оборачивает запрос. А решает, будет ли запрос вообще выполнен, Авторизация.