ミドルウェア

機械翻訳

このページは英語版ドキュメントから自動翻訳されたものであり、正式な版は英語版のページです。不自然な箇所があれば、翻訳についてで報告の方法を説明しています。

ミドルウェアとは、サーバーが受け取るすべてのメッセージを包み込む 1 つの非同期関数です。

async (ctx, call_next) の形で書き、server.middleware に追加します。API はこれだけです。

Warning

ミドルウェアのリストは、ソース上で暫定（provisional）とマークされています。シグネチャやセマンティクスは 2.x のマイナーリリースで変わる可能性があります。メッセージを「観察」する（計時、ログ、トレース）ため、あるいは「拒否」するために使ってください。サーバーの土台にはしないでください。

MCPServer は構築時にこのリストを受け取り（MCPServer(name, middleware=[...])）、mcp.middleware として公開します。低レベルの Server も同じリストを server.middleware として公開します。以下の例では低レベルの Server を使います。Server(name, on_call_tool=...) に馴染みがなければ、先に低レベルの Server を読んでください。

計時ミドルウェア

サーバー 1 つ、ツール 1 つ、そして各メッセージにかかった時間をログに出すミドルウェア 1 つです。

server.py

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 は意図的なものです。ハンドラーが例外を送出しても計時されます。失敗は call_next から出てくる例外としてミドルウェアに届くからです。
server.middleware.append(...) で登録します。リストは外側から順に実行されるので、middleware[0] が通信路に最も近いミドルウェアです。

試してみる

クライアントを接続し、ツールを一覧し、1 つ呼び出してください。ログには 3 行出ます。

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

呼び出しは 2 回なのに、行は 3 つです。最初の行は 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 を送出すると、そのメッセージ 1 つに JSON-RPC エラーで応答します。接続は維持され、次のメッセージは通ります。サーバーが呼び出し側ごとに subscriptions/listen を制御するのはこの方法です。サブスクリプションのページの誰が監視できるかを決めるで順を追って説明しています。
書き換える。 ctx はデータクラスです。await call_next(dataclasses.replace(ctx, params=...)) とすると、クライアントが送ったものとは異なるパラメーターをチェーンの残りに渡せます。initialize に対しては決して行わないでください。クライアントが受け取る結果は書き換えたパラメーターから組み立てられますが、サーバーは元の通信路上のパラメーターから接続状態を確定します。両者が、ネゴシエートした内容について食い違ったままハンドシェイクを終える可能性があります。
応答する。 call_next(ctx) を呼ばずに結果を返すと、それがレスポンスとしてクライアントへ送られます。call_next が渡してくるのは完成した送信形式であり、パイプラインは返したものに一切手を加えないので、エンベロープ全体が自分の責任になります。2026 年世代の接続ではこれに serverInfo の _meta スタンプが含まれます。SDK はハンドラーの結果にはこれを付けますが、ミドルウェアが返すものには付けません。

Check

initialize もミドルウェアが包むものの 1 つであり、ミドルウェアはそのための「唯一の」フックです。add_request_handler で乗っ取ろうとすると、SDK は拒否します。

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

Warning

initialize はインラインで処理されます。ミドルウェアチェーンが返るまで、サーバーはそれ以上の受信メッセージを読みません。そのため、initialize の処理中にサーバーからクライアントへのリクエスト（ctx.session.send_request(...) やエリシテーション（elicitation））を await すると、接続がデッドロックします。待っているレスポンスは決して読まれないからです。送りっぱなしの通知は問題ありません。

デフォルトで有効な唯一のミドルウェア

SDK が同梱するミドルウェアはちょうど 1 つで、すでにサーバーのリストに載っています。すべてのメッセージに対して OpenTelemetry のスパンを発行するミドルウェアです。自分で追加する必要はなく、ほとんどの場合は意識することもありません。エクスポーターをインストールするまでは何もしません。専用のページがあります。OpenTelemetry を参照してください。

Info

ASGI ミドルウェアを書いたことがあれば、この形はもう知っています。Starlette の (scope, receive, send) が (ctx, call_next) になり、トランスポートの「後」で、生の HTTP リクエストではなくデコード済みのメッセージに対して動きます。2 つは組み合わせられます。streamable_http_app() 上の Starlette ミドルウェアは HTTP を見て、こちらは MCP を見ます。

まとめ

ミドルウェアは async (ctx, call_next) -> result です。MCPServer(middleware=[...]) として渡すか（または mcp.middleware に追加し）、低レベルの Server では server.middleware に追加します。
受信するすべてのメッセージ（server/discover、initialize、リクエスト、通知、未知のメソッド）を包み、外側から順に実行されます。
ctx.request_id is None で、通知とリクエストを見分けます。
call_next を呼ぶ代わりに例外を送出すると、メッセージを 1 つ拒否できます。接続は維持されます。
SDK 自身の OpenTelemetry トレースもミドルウェアであり、すでにリストに載っています。OpenTelemetry を参照してください。
この仕組み全体が暫定です。観察には使っても、その上に何かを築かないでください。

リクエストを包むものはこれですべてです。認可は、そもそもそのリクエストを実行させるかどうかを決めるものです。