コンテンツにスキップ

低レベルの Server

機械翻訳

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

@mcp.tool() はひとつの層です。その下には 2 つ目のサーバークラス Server があり、生の MCP を話します。プロトコルオブジェクトを渡すと、それをそのまま通信路に載せます。

MCPServer はその上に作られています。便利な層が邪魔になるときは、下に降ります。

  • Python のシグネチャから導出されたものではなく、正確なスキーマ（ファイルから読み込んだもの、データベースから生成したもの）を出力する必要がある。
  • 結果を完全に制御する必要がある。_meta、is_error、structured_content のすべてのキー。
  • MCP が定義していないメソッドを扱う必要がある。

それ以外はすべて、MCPServer のままで構いません。

同じツールを手書きする

これは ツール が @mcp.tool() を使って 9 行で書いている search_books ツールから、糖衣構文を取り除いたものです。

server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
    return CallToolResult(content=[TextContent(type="text", text=text)])


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)

変わったのは 3 つで、それが低レベル API のすべてです。

  • ハンドラーはコンストラクターのパラメーターです。 on_list_tools= と on_call_tool= を Server(...) に渡します。この層にデコレーターはなく、すべてのハンドラーが同じ形 async (ctx, params) -> result をしています。
  • 入力スキーマは自分で書きます。 Tool.input_schema は素の JSON Schema の dict です。誰も型ヒントから導出してはくれません。導出元になる型ヒントがないからです。
  • 結果は自分で組み立てます。 CallToolResult(content=[TextContent(...)]) を手で書きます。ラップされるものも、変換されるものも、戻り値のアノテーションから推論されるものもありません。

params はパース済みのリクエストです。CallToolRequestParams からは .name と .arguments が取れます。ctx は ServerRequestContext です。クライアントに話しかけるための ctx.session、ctx.lifespan_context、ctx.request_id、そして受信したリクエストの _meta である ctx.meta があります。

Info

FastAPI を使ったことがあれば、この関係はもう知っています。MCPServer はデコレーターと型ヒントの層で、Server はその下の Starlette です。両者は競合するものではありません。MCPServer は Server を構築し、まさにこのようなハンドラーをそこに登録します。

試してみる

これには Inspector がありません。mcp dev と mcp run は MCPServer しか受け付けないからです。インメモリの Client は気にしません。MCPServer を受け取るのとまったく同じように、低レベルの Server を受け取ります。

main.py
import asyncio

from mcp import Client

from server import server


async def main() -> None:
    async with Client(server) as client:
        result = await client.call_tool("search_books", {"query": "dune", "limit": 5})
        print(result.content)


asyncio.run(main())

[TextContent(type='text', text="Found 3 books matching 'dune' (showing up to 5).", annotations=None, meta=None)]

@mcp.tool() 版が出力したのと同じテキストです。正直に言うと、違いが 2 つあります。

  • result.structured_content は None です。高レベルのサーバーは -> str を {"result": ...} にラップしてくれますが、ここでは自分で組み立てなかったものを誰も組み立ててくれません。
  • list_tools は自分で打ち込んだスキーマを一字一句そのまま返します。高レベル版にはすべてのプロパティに "title": "Query" があり、ルートに "title": "search_booksArguments" がありました。Pydantic の産物です。この層では、通信上に現れるものはすべて自分が載せたものです。

何もチェックされない

MCPServer は、生成したスキーマに照らして呼び出しを検証し、関数が実行される前に不正な引数を拒否します（ツール）。

Server はそれをしません。input_schema はクライアントに「公開」されますが、params.arguments に「適用」されることは決してありません。

Check

limit なしで search_books を呼び出すと、args["limit"] が KeyError を送出します。クライアントに見えるのは次のとおりです。

MCPError: Internal server error

JSON-RPC エラー、コード -32603、メッセージは意図的に一般的なものです。SDK はトレースバックをリモートの呼び出し側に漏らしません。モデルは自分が何を間違えたのか知ることができないので、再試行もできません。（テストでは、raise_exceptions=True を指定すると代わりに本当の例外が表に出ます。テスト を参照してください。）

これは一般化できます。低レベルのハンドラーから送出された例外は常にプロトコルエラーであり、is_error=True のツール結果になることはありません。モデルに失敗を読ませて回復させたいなら、params.arguments を自分で検証し、CallToolResult(content=[TextContent(...)], is_error=True) を返してください。この 2 種類の失敗が エラーの処理 の主題です。

2 つのツール、1 つのハンドラー

on_call_tool はサーバー上のすべてのツールの唯一の入り口です。params.name で振り分けます。

server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
)

ADD_BOOK = Tool(
    name="add_book",
    description="Add a book to the catalog.",
    input_schema={
        "type": "object",
        "properties": {"title": {"type": "string"}, "author": {"type": "string"}, "year": {"type": "integer"}},
        "required": ["title", "author", "year"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS, ADD_BOOK])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    if params.name == "search_books":
        text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
    elif params.name == "add_book":
        text = f"Added {args['title']!r} by {args['author']} ({args['year']})."
    else:
        raise ValueError(f"Unknown tool: {params.name}")
    return CallToolResult(content=[TextContent(type="text", text=text)])


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
  • list_tools は両方を公開します。call_tool は名前でディスパッチします。
  • else 分岐は重要です。Server は、一度もリストに載せていない名前への tools/call でも、そのままハンドラーに転送してしまいます。そこで例外を送出すると、呼び出しは上と同じ -32603 になります。

構造化出力を手書きする

Tool に output_schema を宣言し、結果に structured_content を載せます。どちらも自分の責任です。

server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
    output_schema={
        "type": "object",
        "properties": {"matches": {"type": "integer"}, "query": {"type": "string"}},
        "required": ["matches", "query"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    data = {"matches": 3, "query": args["query"]}
    return CallToolResult(
        content=[TextContent(type="text", text=f"Found 3 books matching {args['query']!r}.")],
        structured_content=data,
    )


server = Server("Bookshop", version="2.0.0", on_list_tools=list_tools, on_call_tool=call_tool)

呼び出すと、結果は両方の表現を持ちます。

{
  "content": [{"type": "text", "text": "Found 3 books matching 'dune'."}],
  "structuredContent": {"matches": 3, "query": "dune"},
  "isError": false,
  "resultType": "complete",
  "_meta": {"io.modelcontextprotocol/serverInfo": {"name": "Bookshop", "version": "2.0.0"}}
}

_meta ブロックはサーバーの識別スタンプです。SDK は 2026 年世代のすべての結果にこれを追加し、version はコンストラクターの値を使います（何も設定していないサーバーは空文字列を報告します）。自身を識別してはならないサーバーは、返す結果を所有するミドルウェアでこのキーを取り除けます。

サーバーは 2 つのフィールドを比較しません。この SDK の Client は比較します。宣言した output_schema を満たさない structured_content を返すと、call_tool は Invalid structured content returned by tool search_books で始まり、続けて jsonschema の失敗内容を引用する RuntimeError を送出します。スキーマを約束するのは簡単ですが、守るのは自分の仕事です。戻り値の型とスキーマの全段階については 構造化出力 を参照してください。

_meta：モデルではなくアプリケーションのために

content は答えのうちモデルが読む部分です。structured_content は同じ答えを型付きデータにしたものです。_meta は 3 つ目のチャネルで、答えの一部ではまったくなく、クライアントアプリケーションのために結果に同乗するデータです。

レコード ID、トレース ID など、UI が必要としプロンプトが必要としないものに使います。

server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
    output_schema={
        "type": "object",
        "properties": {"matches": {"type": "integer"}, "query": {"type": "string"}},
        "required": ["matches", "query"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    data = {"matches": 3, "query": args["query"]}
    return CallToolResult(
        content=[TextContent(type="text", text=f"Found 3 books matching {args['query']!r}.")],
        structured_content=data,
        _meta={"bookshop/record_ids": ["bk_17", "bk_42", "bk_99"]},
    )


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
  • 構築するときは通信路上の名前である _meta= を使います。クライアントは result.meta として読み出します。
  • キーには名前空間を付けてください（bookshop/record_ids）。io.modelcontextprotocol/* のキーはプロトコルが予約しています。

Warning

_meta はサーバーとクライアントアプリケーションの間の取り決めであり、何がモデルに届くかについての保証ではありません。何を描画するかはホストが決めます。ツール結果のどの部分にも、決して秘密情報を入れないでください。

ケイパビリティはハンドラーに従う

Server は、ハンドラーを渡したメソッド群だけを正確に公開します。上の Bookshop は on_list_tools と on_call_tool だけを渡しているので、接続したクライアントには次のように見えます。

{"tools": {"listChanged": false}}

resources も prompts もありません。裏付けるものがないからです。on_list_prompts を渡せば prompts が現れ、on_completion を渡せば completions が現れます。

MCPServer は、何かを登録したかどうかにかかわらず、常にツール、リソース、プロンプトを公開します。そのマネージャーが常に存在するからです。この層では、宣言とはコンストラクター呼び出しそのものです。

ライフスパンのジェネリック

Server は、そのライフスパンが yield する型についてジェネリックです。一度アノテーションを付ければ、そのオブジェクトは現れる場所すべてで型が付きます。

server.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)


@dataclass
class Catalog:
    books: list[str]

    def search(self, query: str) -> list[str]:
        return [title for title in self.books if query.lower() in title.lower()]


@asynccontextmanager
async def lifespan(server: Server[Catalog]) -> AsyncIterator[Catalog]:
    yield Catalog(books=["Dune", "Dune Messiah", "Children of Dune"])


SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
)


async def list_tools(ctx: ServerRequestContext[Catalog], params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext[Catalog], params: CallToolRequestParams) -> CallToolResult:
    matches = ctx.lifespan_context.search((params.arguments or {})["query"])
    text = f"Found {len(matches)} books: {', '.join(matches)}."
    return CallToolResult(content=[TextContent(type="text", text=text)])


server = Server("Bookshop", lifespan=lifespan, on_list_tools=list_tools, on_call_tool=call_tool)
  • ライフスパンは Callable[[Server[Catalog]], AbstractAsyncContextManager[Catalog]] です。async ジェネレーターに @asynccontextmanager を付けると、まさにそれが得られます。
  • yield したものが ctx.lifespan_context になり、ハンドラーに ServerRequestContext[Catalog] とアノテーションが付いているので、.search(...) が補完され、型チェックされます。
  • サーバーの起動時に一度入り、停止時に一度出ます。起動、後始末、そして同じ考え方の MCPServer 版については ライフスパン を参照してください。

lifespan= がなければ、ctx.lifespan_context は空の dict です。

独自のメソッド

コンストラクターは MCP が定義するメソッドを扱います。add_request_handler はそれ以外のすべてを扱います。

server.py
from pydantic import BaseModel

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    RequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
    return CallToolResult(content=[TextContent(type="text", text=text)])


class ReindexParams(RequestParams):
    full: bool = False


class ReindexResult(BaseModel):
    indexed: int


async def reindex(ctx: ServerRequestContext, params: ReindexParams) -> ReindexResult:
    return ReindexResult(indexed=3)


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
server.add_request_handler("bookshop/reindex", ReindexParams, reindex)
  • 最初の引数はメソッド文字列です。通知には対になる add_notification_handler があります。
  • params_type は、受信した params をハンドラーの実行前に検証するためのモデルです。つまり、カスタムメソッドはツールが受けられない検証を受けられます。_meta フィールドがほかのメソッドと同じようにパースされるよう、RequestParams をサブクラス化してください。
  • ハンドラーは BaseModel、dict、または None を返します。SDK がそれを JSON-RPC の結果にシリアライズします。

正直な注意点が 1 つあります。高レベルの Client には MCP が定義するメソッドの動詞しかないので、client.reindex() はありません。ベンダーメソッドは、その存在をすでに知っている相手のためのものです。一緒に配布するクライアントや、JSON-RPC を話す自前の別のサービスなどです。

自分のものにできないメソッドが 1 つあります。

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

ハンドシェイクはランナーのものです。server/discover、ping、その他すべての組み込みは自由に置き換えられます。

Tip

このエラーで言及されている Server.middleware は、initialize を含むすべての受信メッセージをラップします。新しいメソッドに応答するのではなく、トラフィックを観察したり書き換えたりしたいなら、ミドルウェア から始めてください。

その他のハンドラー

以下はどれも、ここまでで身につけた語彙で理解できる考え方です。それぞれに専用のページがあります。

  • on_call_tool、on_get_prompt、on_read_resource は、通常の結果の代わりに InputRequiredResult を返して呼び出しを一時停止し、クライアントに入力を求めることができます。マルチラウンドトリップ（multi-round-trip）リクエスト を参照してください。この層らしく、何も代わりにインストールされません。MCPServer はデフォルトで requestState を封印しますが、ここでは設定した request_state は書いたとおりに通信路を渡ります。server.middleware.append(RequestStateBoundary(RequestStateSecurity(keys=[...]), default_audience=server.name)) でオプトインするまではそうです。この 1 行（どちらの名前も mcp.server.request_state からインポートします）で、MCPServer が行うのとまったく同じ封印と検証が得られます（requestState の保護）。
  • on_list_resources、on_read_resource、on_list_prompts、on_get_prompt、on_completion は、ほかのプリミティブ向けの同じ (ctx, params) -> result の形です。
  • on_subscriptions_listen は 2026-07-28 の subscriptions/listen ストリームを提供します。SubscriptionBus の上に構築した ListenHandler を渡し、ほかのハンドラーからバスにイベントを発行してください。全体の組み立て方については サブスクリプション を参照してください。
  • server.streamable_http_app() は MCPServer のものと同じ Starlette アプリを返します。サーバーの実行 がほかの ASGI アプリをデプロイするのと同じ方法でデプロイしてください。この層には server.run(transport=...) はありません。server.run(read_stream, write_stream, server.create_initialization_options()) が 1 組のストリーム上で 1 つの接続を駆動し、その 1 行がすべてです。

まとめ

  • 低レベルの Server はハンドラーを on_* のコンストラクターパラメーターとして受け取ります。すべてのハンドラーは async (ctx, params) -> result です。
  • input_schema の dict は自分で書き、CallToolResult は自分で組み立てます。導出も、ラップも、検証も、代わりにしてくれるものはありません。
  • ハンドラー内の例外は -32603 のプロトコルエラーです。モデルが読めるツールエラーは、is_error=True を付けて自分で返す CallToolResult です。
  • 結果の _meta はモデルではなくクライアントアプリケーション宛てです。
  • Server[T] はライフスパンが yield するものについてジェネリックで、ctx.lifespan_context は型付きの T です。
  • add_request_handler(method, params_type, handler) は任意のメソッドを提供します。initialize は予約されています。
  • Server が公開するケイパビリティは、どのハンドラーを登録したかから導出されます。

Client(server) が両方のサーバーを同じように扱ったのは、両者がまさに同じプロトコルだからであり、それこそが要点です。さらに下の層はクラスですらありません。ミドルウェア です。