Middleware

middleware एक async function है जो server को मिलने वाले हर message को wrap करता है।

इसे आप async (ctx, call_next) के रूप में लिखते हैं और server.middleware में append करते हैं। पूरा API बस इतना ही है।

MCPServer यह list construction के समय लेता है (MCPServer(name, middleware=[...])) और इसे mcp.middleware के रूप में उपलब्ध कराता है; low-level Server वही list server.middleware के रूप में देता है। नीचे दिए गए उदाहरण low-level Server इस्तेमाल करते हैं; अगर Server(name, on_call_tool=...) आपके लिए नया है, तो पहले Low-level Server पढ़ें।

Timing middleware

एक server, एक tool, एक middleware जो log करता है कि हर 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 वही ServerRequestContext है जो आपके handlers को मिलता है। ctx.method raw method string है; ctx.params raw params हैं, किसी भी validation से पहले।
• call_next(ctx) बाकी chain चलाता है: validation, handler lookup, आपका handler। जो उसने लौटाया वही लौटा दें, तो response जस का तस रहता है।
• try/finally जानबूझकर है: जो handler raise करता है उसका समय भी मापा जाता है, क्योंकि failure आपके middleware तक call_next से निकले exception के रूप में पहुँचती है।
• server.middleware.append(...) इसे register करता है। list outermost-first चलती है, इसलिए middleware[0] वह है जो wire के सबसे नज़दीक है।

इसे आज़माएँ

client connect करें, tools की सूची लें, एक को call करें। आपके log में तीन lines हैं:

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

आपने दो calls किए और तीन lines मिलीं। पहली server/discover है: वह request जो client ने connection तैयार करने के लिए भेजी, आपके कुछ माँगने से पहले।

यही असली बात है। middleware हर inbound message को wrap करता है:

• connection setup: server/discover, या legacy session पर initialize और notifications/initialized।
• हर request और हर notification जो server तक पहुँचे। notification के लिए ctx.request_id is None होता है, call_next(ctx) None लौटाता है, और आप जो भी लौटाएँ वह फेंक दिया जाता है। (2026-07-28 वाले streamable-HTTP path पर client का notification POST transport पर ही 202 से acknowledge हो जाता है और कभी dispatch नहीं होता, इसलिए वह middleware तक भी नहीं पहुँचता; वह revision HTTP पर कोई client-to-server notifications define ही नहीं करता।)
• वह method भी जिसके लिए server के पास कोई handler नहीं है: call_next MCPError(-32601, "Method not found") को client की ओर जाते हुए आपके middleware के बीच से raise करता है।

concurrency की सीमा

middleware के लिए call_next(ctx) call करना ज़रूरी नहीं है। इसकी जगह MCPError raise करें और वह एक message अस्वीकार कर दिया जाता है: connection बना रहता है और अगला message निकल जाता है।

मान लें कि हर search चार connections वाले pool का एक connection थामे रखता है। यह middleware चार tool calls को एक साथ चलने देता है और पाँचवें को अस्वीकार कर देता है:

from typing import Any

from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from mcp.server.context import CallNext, HandlerResult, ServerMiddleware
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

# MCP defines no "busy" error, so this server picks its own code.
SERVER_BUSY = 1


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}.")])


def max_concurrent_tool_calls(limit: int) -> ServerMiddleware[Any]:
    running = 0

    async def middleware(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult:
        nonlocal running
        if ctx.method != "tools/call":
            return await call_next(ctx)
        if running >= limit:
            raise MCPError(code=SERVER_BUSY, message=f"Server busy: tool call limit reached ({limit} in progress).")
        running += 1
        try:
            return await call_next(ctx)
        finally:
            running -= 1

    return middleware


server = Server("Bookshop", on_list_tools=on_list_tools, on_call_tool=on_call_tool)
server.middleware.append(max_concurrent_tool_calls(4))

• सिर्फ़ tools/call गिना जाता है, इसलिए tool calls अस्वीकार करते समय भी server server/discover और tools/list का जवाब देता रहता है।
• MCP कोई "server busy" error code define नहीं करता, इसलिए SERVER_BUSY इस server का अपना है।
• अस्वीकार करने से client को तुरंत पता चल जाता है कि server overloaded है। अगर आप callers को इंतज़ार कराना बेहतर समझते हैं, तो इसकी जगह call_next(ctx) के चारों ओर anyio.CapacityLimiter hold करें।

raise किया गया MCPError client application को जाता है, model को नहीं। अगर message model को पढ़ना चाहिए, तो इसकी जगह is_error=True वाला tool result लौटाएँ: यही नीचे वाला जवाब दें है।

इसके अंदर आप क्या कर सकते हैं

इस क्रम में कि आपको कितना हिचकना चाहिए, कम से ज़्यादा की ओर:

• देखें। समय मापें, गिनें, log करें। ऊपर वाला timing middleware।
• अस्वीकार करें। call_next(ctx) call करने के बजाय MCPError raise करें और उस एक message का जवाब JSON-RPC error से दिया जाता है। connection बना रहता है; अगला message निकल जाता है। ऊपर वाली concurrency की सीमा। server हर caller के लिए subscriptions/listen को भी इसी तरह gate करता है: Subscriptions page पर यह तय करना कि कौन देख सकता है इसे चरण दर चरण समझाता है।
• फिर से लिखें। ctx dataclass है: await call_next(dataclasses.replace(ctx, params=...)) बाकी chain को client के भेजे params से अलग params देता है। initialize के साथ ऐसा कभी न करें: client को जो result वापस मिलता है वह आपके बदले हुए params से बनता है, लेकिन server अपनी connection state मूल wire params से commit करता है। दोनों पक्ष handshake इस असहमति के साथ पूरा कर सकते हैं कि उन्होंने क्या negotiate किया।
• जवाब दें। call_next(ctx) call किए बिना result लौटाएँ और वह आपके response के रूप में client को जाता है। call_next आपको तैयार wire form देता है, और pipeline आप जो लौटाते हैं उसे कभी patch नहीं करता, इसलिए पूरा envelope आपका है: 2026 पीढ़ी के connection पर इसमें serverInfo का _meta stamp शामिल है, जिसे SDK handler results में जोड़ता है पर आपके results में नहीं।

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

वह एक middleware जो default रूप से चालू आता है

SDK ठीक एक middleware साथ देता है, और वह पहले से आपके server की list में है: वह जो हर message के लिए OpenTelemetry span emit करता है। आप इसे append नहीं करते, और ज़्यादातर समय इसके बारे में सोचते भी नहीं। जब तक आप कोई exporter install नहीं करते यह no-op है, और इसका अपना page है: OpenTelemetry।

सारांश

• middleware async (ctx, call_next) -> result है, जिसे MCPServer(middleware=[...]) के रूप में पास किया जाता है (या mcp.middleware में append किया जाता है), और low-level Server पर server.middleware में append किया जाता है।
• यह server तक पहुँचने वाले हर inbound message को wrap करता है (server/discover, initialize, requests, notifications, अनजान methods) और outermost-first चलता है।
• ctx.request_id is None से आप notification और request में फ़र्क करते हैं।
• एक message को अस्वीकार करने के लिए call_next call करने के बजाय raise करें; connection बचा रहता है।
• SDK का अपना OpenTelemetry tracing भी एक middleware है, जो पहले से list में है। देखें OpenTelemetry।
• पूरा surface provisional है। इससे देखें; इस पर निर्माण न करें।

request को wrap करने वाली हर चीज़ बस इतनी ही है। Authorization वह है जो तय करता है कि request को चलने दिया जाए भी या नहीं।