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 сюди входить і позначка _meta з serverInfo, яку 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.
• Він огортає кожне вхідне повідомлення (server/discover, initialize, запити, сповіщення, невідомі методи) і виконується від зовнішнього до внутрішнього.
• ctx.request_id is None — так відрізняють сповіщення від запиту.
• Викиньте виняток замість виклику call_next, щоб відхилити одне повідомлення; з'єднання вціліє.
• Власне трасування OpenTelemetry у SDK — теж middleware, і воно вже в списку. Див. OpenTelemetry.
• Уся ця поверхня попередня. Спостерігайте через неї; не будуйте на ній.

Це все, що огортає запит. Авторизація — те, що вирішує, чи запит узагалі буде виконано.