미들웨어

미들웨어는 서버가 받는 모든 메시지를 감싸는 하나의 async 함수입니다.

async (ctx, call_next) 형태로 작성해서 server.middleware에 추가하면 됩니다. 이것이 API의 전부입니다.

MCPServer는 생성 시 목록을 받아(MCPServer(name, middleware=[...])) mcp.middleware로 노출하고, 저수준 Server는 같은 목록을 server.middleware로 노출합니다. 아래 예제는 저수준 Server를 사용합니다. Server(name, on_call_tool=...)가 처음이라면 저수준 Server를 먼저 읽으세요.

시간을 재는 미들웨어

서버 하나, 도구 하나, 그리고 메시지마다 걸린 시간을 로그로 남기는 미들웨어 하나입니다.

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는 어떤 검증도 거치기 전의 원시 params입니다.
• call_next(ctx)는 체인의 나머지, 즉 검증, 핸들러 조회, 작성한 핸들러를 실행합니다. 반환된 값을 그대로 반환하면 응답은 손대지 않은 채로 나갑니다.
• try/finally는 의도적인 선택입니다. 예외를 일으키는 핸들러도 시간이 측정됩니다. 실패는 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로, 아무것도 요청하기 전에 클라이언트가 연결을 설정하려고 보낸 요청입니다.

바로 이것이 핵심입니다. 미들웨어는 들어오는 모든 메시지를 감쌉니다.

• 연결 설정. server/discover이거나, 레거시 세션에서는 initialize와 notifications/initialized입니다.
• 모든 요청과 모든 알림. 알림의 경우 ctx.request_id is None이고, call_next(ctx)는 None을 반환하며, 무엇을 반환하든 버려집니다.
• 서버에 핸들러가 없는 메서드까지도 포함됩니다. call_next가 MCPError(-32601, "Method not found")를 일으키고, 이 예외는 클라이언트로 가는 길에 미들웨어를 통과합니다.

미들웨어 안에서 할 수 있는 일

망설임이 적게 필요한 것부터 순서대로 나열합니다.

• 관찰. 시간을 재고, 횟수를 세고, 로그를 남기세요. 위의 예제가 이에 해당합니다.
• 거부. call_next(ctx)를 호출하는 대신 MCPError를 일으키면 그 메시지 하나에 JSON-RPC 오류로 응답합니다. 연결은 유지되고 다음 메시지는 그대로 통과합니다. 서버가 호출자별로 subscriptions/listen을 제한하는 방법이 바로 이것입니다. 구독 페이지의 누가 지켜볼 수 있는지 정하기에서 단계별로 설명합니다.
• 재작성. ctx는 데이터클래스입니다. await call_next(dataclasses.replace(ctx, params=...))는 체인의 나머지에 클라이언트가 보낸 것과 다른 params를 넘깁니다. initialize에는 절대 이렇게 하지 마세요. 클라이언트가 돌려받는 결과는 재작성한 params로 만들어지지만, 서버는 원래 와이어 params를 기준으로 연결 상태를 확정합니다. 양쪽이 무엇을 협상했는지 서로 다르게 이해한 채로 핸드셰이크를 마칠 수 있습니다.
• 응답. 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

기본으로 켜져 있는 단 하나의 미들웨어

SDK에는 미들웨어가 정확히 하나 포함되어 있으며, 이미 서버의 목록에 들어 있습니다. 모든 메시지마다 OpenTelemetry 스팬을 내보내는 미들웨어입니다. 직접 추가할 필요가 없고, 대부분의 경우 신경 쓸 필요도 없습니다. 익스포터를 설치하기 전까지는 아무 일도 하지 않으며, 별도의 페이지가 있습니다. OpenTelemetry를 참고하세요.

요약

• 미들웨어는 async (ctx, call_next) -> result 형태이며, MCPServer(middleware=[...])로 전달하거나(또는 mcp.middleware에 추가하거나) 저수준 Server에서는 server.middleware에 추가합니다.
• 들어오는 모든 메시지(server/discover, initialize, 요청, 알림, 알 수 없는 메서드)를 감싸며 바깥쪽부터 실행됩니다.
• ctx.request_id is None으로 알림과 요청을 구분합니다.
• call_next를 호출하는 대신 예외를 일으키면 메시지 하나를 거부합니다. 연결은 유지됩니다.
• SDK 자체의 OpenTelemetry 트레이싱도 미들웨어이며, 이미 목록에 있습니다. OpenTelemetry를 참고하세요.
• 이 표면 전체가 잠정적입니다. 관찰하는 데 사용하고, 그 위에 무언가를 쌓지는 마세요.

요청을 감싸는 것은 이것이 전부입니다. 요청이 애초에 실행될 수 있는지를 결정하는 것은 인가입니다.