低レベルの Server
@mcp.tool() はひとつの層です。その下には 2 つ目のサーバークラス Server があり、生の MCP を話します。プロトコルオブジェクトを渡すと、それをそのまま通信路に載せます。
MCPServer はその上に作られています。便利な層が邪魔になるときは、下に降ります。
- Python のシグネチャから導出されたものではなく、正確なスキーマ（ファイルから読み込んだもの、データベースから生成したもの）を出力する必要がある。
- 結果を完全に制御する必要がある。
_meta、is_error、structured_contentのすべてのキー。 - MCP が定義していないメソッドを扱う必要がある。
それ以外はすべて、MCPServer のままで構いません。
同じツールを手書きする
これは ツール が @mcp.tool() を使って 9 行で書いている search_books ツールから、糖衣構文を取り除いたものです。
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)
app = server.streamable_http_app()
変わったのは 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 を構築し、まさにこのようなハンドラーをそこに登録します。
試してみる
mcp dev と mcp run は MCPServer しか受け付けないので、このサーバーは自分で起動します。server.py の最後の行がここから普通の ASGI アプリを組み立て、uvicorn がそれを実行します。
uvicorn server:app --port 8000
Inspector でも、どのクライアントでも、http://localhost:8000/mcp に向けてください。
import asyncio
from mcp import Client
async def main() -> None:
async with Client("http://localhost:8000/mcp") 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 の産物です。この層では、通信上に現れるものはすべて自分が載せたものです。
テストでは uvicorn もポートも省けます。Client(server) は MCPServer を受け取るのとまったく同じように、低レベルの Server をインプロセスで受け取ります。テスト がまさにそのパターンです。
何もチェックされない
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 で振り分けます。
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 を載せます。どちらも自分の責任です。
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 を送出します。スキーマを約束するのは簡単ですが、守るのは自分の仕事です。戻り値の型とスキーマの全段階については 構造化出力 を参照してください。
ダイアレクトは JSON Schema 2020-12
input_schema と output_schema は JSON Schema であり、MCP 仕様 がそのダイアレクトを定めています。$schema キーのないスキーマは JSON Schema 2020-12 です。MCPServer が生成するスキーマはこのデフォルトに依存しています（Pydantic は 2020-12 を書き出し、キーを省略します）。手書きの dict にも同じ基準が適用されるので、2020-12 の語彙がすべて使えます。
from mcp.server import Server, ServerRequestContext
from mcp.types import CallToolRequestParams, CallToolResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool
FIND_BOOK = Tool(
name="find_book",
description="Find one book by ISBN, or by title and author.",
input_schema={
"type": "object",
"properties": {
"isbn": {"type": "string", "pattern": "^[0-9]{13}$"},
"title": {"type": "string"},
"author": {"type": "string"},
},
"oneOf": [{"required": ["isbn"]}, {"required": ["title", "author"]}],
"additionalProperties": False,
},
)
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[FIND_BOOK])
async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
args = params.arguments or {}
found = f"ISBN {args['isbn']}" if "isbn" in args else f"{args['title']!r} by {args['author']}"
return CallToolResult(content=[TextContent(type="text", text=f"Found {found} on shelf C-3.")])
server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
input_schemaのルートは"type": "object"でなければなりません。その横に置くoneOf、additionalProperties、anyOf、if/then/else、prefixItems、ローカルな$refを伴う$defs、そのほかの 2020-12 キーワードは、書いたとおりにクライアントに届きます。$schemaキーは不要です。古いドラフトにオプトインする場合にだけ追加してください。この SDK のClientはツールのoutput_schemaに照らしてstructured_contentを検証しますが、バリデーターは$schemaから選び、キーがなければ 2020-12 を使います。
_meta：モデルではなくアプリケーションのために
content は答えのうちモデルが読む部分です。structured_content は同じ答えを型付きデータにしたものです。_meta は 3 つ目のチャネルで、答えの一部ではまったくなく、クライアントアプリケーションのために結果に同乗するデータです。
レコード ID、トレース ID など、UI が必要としプロンプトが必要としないものに使います。
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 する型についてジェネリックです。一度アノテーションを付ければ、そのオブジェクトは現れる場所すべてで型が付きます。
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 はそれ以外のすべてを扱います。
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があります。そのハンドラーは stdio と、ハンドシェイク世代の HTTP 接続で発火します。2026-07-28の Streamable HTTP 経路では、クライアントの通知 POST は202で受理されるだけでディスパッチされません。このリビジョンは HTTP 上でのクライアントからサーバーへの通知を定義していないからです。 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が公開するケイパビリティは、どのハンドラーを登録したかから導出されます。
クライアントが両方のサーバーを同じように扱ったのは、両者がまさに同じプロトコルだからであり、それこそが要点です。さらに下の層はクラスですらありません。ミドルウェア です。