MCP Python SDKのデコレータでツールを定義する実装パターン

MCP Python SDKのMCPServerはFastMCPから何が変わったのか

MCP Python SDKはv2で高水準サーバークラスの名前を変えました。from mcp.server.fastmcp import FastMCPというimport pathは非推奨ではなく削除され、from mcp.server import MCPServerに置き換わっています。pip install "mcp[cli]"(Python 3.10以上が必要)で入るのは現在この2.x系列です。v1系はメンテナンスモードに移り、重大な修正とセキュリティパッチだけが入ります。ライブラリがmcpに依存していて移行の準備ができていないなら、mcp>=1.28,<2のように上限を付けておくと、未固定の解決でも1.x系に留まります。

デコレータそのものの書き方はv1とほぼ変わりません。@mcp.tool()・@mcp.resource()・@mcp.prompt()は同じ引数を受け取り、入力スキーマも変わらず型ヒントから生成されます(@mcp.resource()だけsecurity=キーワードが増えました)。変わったのは周辺です。mcp.server.fastmcp.*配下はmcp.server.mcpserver.*に移り、ctx.fastmcpはctx.mcp_serverに、get_context()は廃止されてctx: Contextという引数を宣言する形に一本化され、例外の基底クラスFastMCPErrorはMCPServerErrorになりました。

サーバーを立ててClaude Codeに接続するところまでの基礎手順はMCPサーバー自作ガイドに譲ります。ここでは、MCPServerのデコレータと型ヒントでどこまで書けるかを、mcp 2.3.0で動かした出力とあわせて見ていきます。

デコレータ1つで型ヒントがJSON Schemaになる

@mcp.tool()を関数に付けるだけで、その関数はツールになります。SDKが関数から読み取るのは3つだけです。関数名がツール名、docstringがツール説明、型ヒントが引数のスキーマになります。関数名が登録名の既定値になる点は、Dagsterの@dg.asset(アセット名の既定が関数名)とも共通します。Dagster側の実装例はClaude CodeでDagsterのアセットを定義するにあります。

from mcp.server import MCPServer
 
mcp = MCPServer("Bookshop")
 
 
@mcp.tool()
def search_books(query: str, limit: int) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r} (showing up to {limit})."

型ヒントは単なるドキュメントではなく契約です。クライアントが"limit": "ten"のような不正な値を送ると、関数が実行される前にSDKが拒否します。引数にデフォルト値を与えれば、それだけでスキーマ上も必須項目から外れます。同じ「型ヒントを契約にする」考え方は、SQLAlchemyのMapped型でモデルを書くときにも表れます(Claude CodeでSQLAlchemyのモデルを実装する手順)。

制約や説明を足したいときはAnnotatedとPydanticのFieldを組み合わせます。

from typing import Annotated, Literal
from pydantic import Field
from mcp.server import MCPServer
 
mcp = MCPServer("Bookshop")
 
 
@mcp.tool()
def search_books(
    query: Annotated[str, Field(description="Title or author to search for.")],
    limit: Annotated[int, Field(ge=1, le=50)] = 10,
    genre: Literal["fiction", "non-fiction", "poetry"] | None = None,
) -> str:
    """Search the catalog by title or author."""
    where = f" in {genre}" if genre else ""
    return f"Found 3 books matching {query!r}{where} (showing up to {limit})."

Field(ge=1, le=50)は"minimum": 1, "maximum": 50としてスキーマに載り、limit=999のような呼び出しは関数が動く前に弾かれます。エラーメッセージはツール結果としてモデルに返るので、モデルはそれを読んで正しい値で呼び直せます。

ただし本文がモデルに届くのは入力検証のエラーとToolErrorのメッセージです。ツール内でそれ以外の例外が出ると、モデルが読むのはError executing tool <name>だけになります。MCPError(v1のMcpError)はツール結果ではなくプロトコルエラーとして扱われます。

実際にmcp 2.3.0で上のツールを登録し、インメモリのClientからtools/listを呼ぶと、入力スキーマは次の形で返りました。limitとgenreは既定値を持つのでrequiredにはqueryしか並ばず、Literalはenumに、| NoneはnullとのanyOfになります。

{"type": "object", "properties": {"query": {"description": "Title or author to search for.", "title": "Query", "type": "string"}, "limit": {"default": 10, "maximum": 50, "minimum": 1, "title": "Limit", "type": "integer"}, "genre": {"anyOf": [{"enum": ["fiction", "non-fiction", "poetry"], "type": "string"}, {"type": "null"}], "default": null, "title": "Genre"}}, "required": ["query"], "title": "search_booksArguments"}

limitに999を渡したときのツール結果は、is_errorが真で、本文は次のように始まります。

Error executing tool search_books: 1 validation error for search_booksArguments
limit
  Input should be less than or equal to 50 [type=less_than_equal, input_value=999, input_type=int]
    For further information visit https://errors.pydantic.dev/2.13/v/less_than_equal

引数が増えてきたらPydanticのBaseModelを1つの引数にまとめる書き方もでき、ネストしたモデル・モデルのリストも組み合わせ自由です。I/Oを伴うツールはasync defで書き、そうでなければ素のdefのままで構いません。v2ではdefのツールは自動的にワーカースレッドで実行されるため、イベントループを塞ぎません。

名前・表示名・ヒントをデコレータで上書きする

関数名やdocstringから導かれる値は、デコレータの引数で置き換えられます。titleはUIに出す人間向けの名前、annotationsはクライアント向けの動作ヒントです。

from mcp.server import MCPServer
from mcp.types import ToolAnnotations
 
mcp = MCPServer("Bookshop")
 
 
@mcp.tool(
    title="Search the catalog",
    annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False),
)
def search_books(query: str) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r}."

read_only_hint=Trueは「何も変更しない」、open_world_hint=Falseは「このカタログのような閉じた集合だけを相手にする」という宣言です。destructive_hintとidempotent_hintは書き込みを行うツール向けで、仕様上は読み取り専用でないツールにだけ意味を持ちます。クライアントが「実行前にユーザーへ確認するか」を決める材料になりますが、あくまでヒントで、セキュリティ境界にはなりません。mcp 2.3.0でtools/listを呼ぶと、上のread_only_hintとopen_world_hintがannotationsにそのまま載ってきます。

戻り値の型注釈がoutput_schemaになる

ツールが返す値にも型ヒントを付けると、それがそのまま構造化出力のスキーマになります。戻り値の型注釈が出力スキーマそのものという設計です。

from mcp.server import MCPServer
 
mcp = MCPServer("Weather")
READINGS = {"London": 17, "Cairo": 34}
 
 
@mcp.tool()
def get_temperature(city: str) -> int:
    """Current temperature in a city, in whole degrees Celsius."""
    return READINGS[city]

intのようなスカラー型はJSONオブジェクトではないので、{"result": 17}という形にラップされます。content(モデルが読むテキスト)とstructured_content(呼び出し元アプリケーションが読む型付きデータ)の両方が、この1つの戻り値から自動生成される仕組みです。

PydanticのBaseModel・TypedDict・dataclassを返り値の型にすると、ラップなしでそのオブジェクトの構造がスキーマになります。3通りの書き方のどれを使っても生成されるスキーマは同じで、既存のコードベースに合わせて選べます。list[T]は{"result": [...]}にラップされますが、dict[str, T]だけは元からJSONオブジェクトなのでラップされません。

戻り値は送信前にこのスキーマへ検証されます。上流のAPIレスポンスが必須フィールドを欠いていれば、その場でツールエラーになりモデルへ伝わるので、欠けたデータをそのままクライアントへ流すことはありません。逆に戻り値の型注釈が型チェッカーのためだけのものなら@mcp.tool(structured_output=False)でこの仕組み自体を無効化できます。1つだけ落とし穴があります。__init__の中だけでフィールドを設定し、クラス本体には型注釈を書いていないクラスを返すと、SDKはエラーを出さずに構造化出力を諦め、モデルにはオブジェクトのrepr文字列が渡ります。mcp 2.3.0で、__init__でだけnameとonlineを設定するStationクラスを返してみました。output_schemaもstructured_contentもNoneです。モデルが読む本文は"<__main__.Station object at 0x...>"でした。原因が分かりにくい不具合になりやすいので、返すクラスにはクラス本体に型注釈を書くのが確実です。同じ関数に@mcp.tool(structured_output=True)を付けると、この沈黙はエラーに変わります。importした時点でInvalidSignature: Function f: return type <class '__main__.S'> is not serializable for structured outputと出るので、実行前に気づけます。

ContextとResolveでモデルに見せない引数を注入する

すべての引数がモデルから来るわけではありません。リクエストの情報やサーバー内部の状態が欲しいときはContext型の引数を宣言するだけで、SDKが実行時に注入します。

from mcp.server import MCPServer
from mcp.server.mcpserver import Context
 
mcp = MCPServer("Bookshop")
 
 
@mcp.tool()
def search_books(query: str, ctx: Context) -> str:
    """Search the catalog by title or author."""
    return f"[request {ctx.request_id}] Found 3 books matching {query!r}."

引数名は何でも構いません(ctx・context・cのどれでも)。SDKは型注釈だけで見つけるので、モデルに公開される入力スキーマにはctxが一切現れません。

Contextからは、リクエストIDのctx.request_id、クライアントへの通知チャンネルのctx.session、サーバー自身のリソースを内部から読むctx.read_resource(uri)などが使えます。HTTP系のトランスポートではctx.headersでリクエストヘッダーも読めますが、ヘッダーはクライアントが自由に設定できる入力です。ロケールや機能フラグには使えても、呼び出し元の本人確認には使えません。

v2で新しく増えたのがResolveです。モデルに直接答えさせたくない値(在庫数のような内部データ、ユーザーにしか答えられない確認事項)は、Annotated[Type, Resolve(resolver_fn)]という形で宣言すると、ツール本体が呼ばれる前に自分で書いた関数がその値を用意してくれます。

from typing import Annotated
from pydantic import BaseModel
from mcp.server import MCPServer
from mcp.server.mcpserver import Resolve
 
mcp = MCPServer("Bookshop")
INVENTORY = {"Dune": 7, "Neuromancer": 0}
 
 
class Stock(BaseModel):
    title: str
    copies: int
 
 
async def check_stock(title: str) -> Stock:
    return Stock(title=title, copies=INVENTORY.get(title, 0))
 
 
@mcp.tool()
async def reserve_book(title: str, stock: Annotated[Stock, Resolve(check_stock)]) -> str:
    """Reserve a copy of a book."""
    if stock.copies == 0:
        return f"{title!r} is out of stock."
    return f"Reserved {title!r} ({stock.copies - 1} copies left)."

呼び出しの流れは次のとおりです。

resolverは別のresolverにも依存でき、同じresolverを複数の引数が必要としても、実行されるのは1回の呼び出しにつき1回です。依存グラフはツールの登録時に解析されます。分類できない引数や、resolver同士の循環があると、クライアントが接続する前に起動時点でInvalidSignatureになります。

Resolveはユーザーへの質問(elicitation)にも使えます。resolverがElicit(...)を返すと、SDKは接続方式に応じて質問を届けます。旧来のクライアントにはその場でのelicitationリクエストとして、2026-07-28リビジョンのクライアントにはmulti-round-trip(サーバーからの能動リクエストを廃した新方式)として、同じツール本体のまま両方を成立させます。このmulti-round-trip自体の仕組みはMCPのMRTRとはで扱っています。旧来のctx.elicit()もレガシー接続では引き続き動きますが、2026-07-28の接続ではNoBackChannelErrorになるため、新しく書くツールはResolveに寄せておくとどちらの接続方式でも動きます。

実行時にツールを増やして通知する

ツールの一覧は、import時に固定されるものではありません。mcp.add_tool(fn)は、@mcp.tool()と同じ規則で名前・説明・スキーマを導いて、素の関数をツールとして登録します。

from mcp.server import MCPServer
from mcp.server.mcpserver import Context
 
mcp = MCPServer("Bookshop")
 
 
def recommend_book(genre: str) -> str:
    """Recommend a book in the given genre."""
    return f"In {genre}, try 'Dune'."
 
 
@mcp.tool()
async def enable_recommendations(ctx: Context) -> str:
    """Switch on the recommendation tool."""
    mcp.add_tool(recommend_book)
    await ctx.session.send_tool_list_changed()
    return "Recommendations are now available."

mcp 2.3.0でインメモリのClientから確かめました。enable_recommendationsより先にrecommend_bookを呼ぶと、is_errorが真で本文はUnknown tool: recommend_bookです。呼んだ後は、同じ呼び出しが成功します。

send_tool_list_changed()はnotifications/tools/list_changedを送る旧来の形です。2026-07-28の接続では、変更通知は購読用のストリーム(subscriptions/listen)にしか届きません。その接続にはawait ctx.notify_tools_changed()を使います。

Lifespanでプロセス全体の状態を持たせる

データベース接続やHTTPクライアントのように、サーバーの寿命いっぱい持ち回したい状態はlifespanに書きます。@asynccontextmanagerで1つのオブジェクトをyieldすると、それがサーバーの生きている間ずっとハンドラから見えます。

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
 
 
class Database:
    @classmethod
    async def connect(cls) -> "Database":
        return cls()
 
    async def disconnect(self) -> None: ...
 
    def query(self) -> int:
        return 42
 
 
@dataclass
class AppContext:
    db: Database
 
 
@asynccontextmanager
async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        await db.disconnect()
 
 
mcp = MCPServer("Bookshop", lifespan=app_lifespan)
 
 
@mcp.tool()
def count_books(genre: str, ctx: Context[AppContext]) -> str:
    """Count the books in a genre."""
    db = ctx.request_context.lifespan_context.db
    return f"{db.query()} books in {genre!r}."

yieldの前が起動処理、finallyの中が終了処理で、lifespanはサーバーの生存期間全体で1回だけ実行されます。ツール側はctx.request_context.lifespan_contextでyieldされたオブジェクトにアクセスします。Context[AppContext]のように型引数を付けると、この参照が型チェッカーの補完対象になります。@mcp.resource()や@mcp.prompt()の関数にも同じctx引数を宣言できます。型引数を付けない素のContextでは、lifespan_contextはdict[str, Any]として扱われます。lifespanを渡さなければSDKの既定が空のdictを返し、Noneにはなりません。

streamable HTTP経由でサーバーを立てる場合、v2ではlifespanが起動時に1回だけ実行され、その状態が全セッション・全リクエストで共有されます。

v1でlifespanの中に接続ごとのリソースを確保していた場合は、共有される前提に書き直さないと挙動が変わります。

v2移植でよくあるつまずき

v1からの移植で最初に踏むのはimportの置き換えですが、それを越えたあとにも動作が変わる箇所がいくつかあります。

トランスポート設定をMCPServer()に渡すとエラーになる。 MCPServer("x", port=9000)のように、v1ではFastMCPのコンストラクタに直接書けていたhost・port・stateless_httpなどは、v2ではrun()側の引数に移りました。MCPServer(...)はサーバーが何であるか(名前・instructions・lifespan・認証)だけを表し、どう配信するかはrun()が担当するという役割分担です。古いコンストラクタ引数をそのまま残すと、mcp 2.3.0ではTypeError: MCPServer.__init__() got an unexpected keyword argument 'port'で止まります。

同期関数がイベントループを塞まなくなった代わりに、スレッド前提のコードが崩れることがある。 defで書いたツール・リソース・resolverは自動的にワーカースレッドで実行されるようになりました。イベントループをブロックしないという恩恵の裏返しとして、スレッドアフィニティのあるコード(特定のスレッドでしか呼べないライブラリなど)を同期関数の中で使っていると、v1では問題なかった処理が別スレッドに移って壊れることがあります。async defのハンドラはこの変更の影響を受けません。

テストクライアントが2026-07-28を既定でネゴシエートする。 Client(mcp)でサーバーをインメモリ接続すると、v2は既定で2026-07-28リビジョンを話します。v1で通っていたctx.elicit()呼び出しをテストで叩くと、この新しい接続ではNoBackChannelErrorになって落ちます。移植したツールがctx.elicit()をまだ使っているなら、Resolveベースの書き方に寄せるか、テストクライアントをClient(mcp, mode="legacy")で明示的にレガシー接続へ固定してください。

まとめ

v1からの移植で書き換えるのは、周辺のAPIだけです。

新規に書くなら、モデルに見せたくない値は最初からContextかResolveで受けます。この形ならctx.elicit()が使えない2026-07-28の接続でも、同じツール本体が動きます。

Claudeへの接続やInspectorでの動作確認といった基礎手順はMCPサーバー自作ガイド、拡張機能ごとの対応状況の違いはMCP拡張機能のクライアント対応状況を比較する、キャッシュ制御のttlMs/cacheScopeはMCPのttlMsとcacheScopeでキャッシュ制御を仕様から理解するで扱っています。