コンテンツにスキップ

拡張機能

機械翻訳

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

拡張機能とは、1 つの識別子の下にまとめられた、オプトイン式の MCP の振る舞い一式です。

サーバー側では、ツール、リソース、新しいリクエストメソッドを提供でき、tools/call をラップすることもできます。クライアント側では、追加の tools/call の結果形状を引き受け(claim)、ベンダー通知を監視できます。それぞれの側が自分の capabilities.extensions でアドバタイズし、求めなかった人にとっては何も変わりません。これが契約です(SEP-2133)。そして黄金律が 1 つあります。拡張機能はデフォルトでオフです

拡張機能を使う

構築時にインスタンスを渡します。

server.py
from mcp.server.apps import Apps
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("demo", extensions=[Apps()])

これで完了です。サーバーは capabilities.extensions の下で io.modelcontextprotocol/ui をアドバタイズし、拡張機能が提供するものをすべて配信するようになります。

Apps は組み込みのリファレンス拡張機能で、専用のページがあります。MCP Apps を参照してください。

Note

拡張機能は構築時に固定されます。後から呼び出す add_extension はありません。クライアントが接続している間、サーバーのケイパビリティマップは変わるべきではないからです。

ケイパビリティマップは server/discover に載って運ばれます。これは 2026-07-28 の経路です。レガシーの initialize ハンドシェイクにはこれを載せる場所がないため、レガシークライアントには拡張機能がそもそも見えません。それを前提に設計してください。拡張機能はサーバーを「補強する」ものであり、サーバーを使う唯一の手段になってはいけません。

独自の拡張機能を書く

Extension をサブクラス化し、必要なものだけをオーバーライドします。どのメソッドにもデフォルトがあります。

識別子

from mcp.server.extension import Extension


class Stamps(Extension):
    identifier = "com.example/stamps"

識別子は、仕様の _meta キーの文法に従った vendor-prefix/name 形式の文字列です。ドット区切りのラベル(それぞれ英字で始まり、英字または数字で終わる)、スラッシュ、そして名前が続きます。クラスが定義された時点で検証されるため、タイプミスがサーバーの起動まで放置されることはありません。

TypeError: Stamps.identifier must be a `vendor-prefix/name` string
(reverse-DNS prefix required), got 'stamps'

プレフィックスには自分が管理するドメインを使ってください。io.modelcontextprotocol/* は MCP プロジェクト自身が仕様化する拡張機能用です。

ツールの提供

役に立つ最小の拡張機能は、ツール 1 つと設定マップ 1 つです。

server.py
from collections.abc import Sequence
from typing import Any

from mcp import Client
from mcp.server.extension import Extension, ToolBinding
from mcp.server.mcpserver import MCPServer


def stamp(text: str) -> str:
    """Stamp a message with the office seal."""
    return f"[stamped] {text}"


class Stamps(Extension):
    """A purely additive extension: one tool, one capability entry."""

    identifier = "com.example/stamps"

    def settings(self) -> dict[str, Any]:
        return {"sealed": True}

    def tools(self) -> Sequence[ToolBinding]:
        return [ToolBinding(fn=stamp)]


mcp = MCPServer("post-office", extensions=[Stamps()])


async def main() -> None:
    async with Client(mcp) as client:
        print(client.server_capabilities.extensions)
        # {'com.example/stamps': {'sealed': True}}
        result = await client.call_tool("stamp", {"text": "hello"})
        print(result.content)
        # [TextContent(text='[stamped] hello')]
  • tools()ToolBinding を返します。サーバーはそれぞれを、自分で mcp.add_tool(...) を呼んだ場合とまったく同じように登録します。スキーマ生成も、Context の注入も、何もかも同じです。
  • settings()capabilities.extensions["com.example/stamps"] にアドバタイズされる値です。設定なしで拡張機能をアドバタイズするには {}(デフォルト)を返してください。
  • 拡張機能がサーバーを受け取ることはありません。提供するものをデータとして宣言し、MCPServer がそれを消費します。書き換えられる self.server はありません。

そして main() がその証明です。mcp に直接つなぐインメモリのクライアントです。

server.py
from collections.abc import Sequence
from typing import Any

from mcp import Client
from mcp.server.extension import Extension, ToolBinding
from mcp.server.mcpserver import MCPServer


def stamp(text: str) -> str:
    """Stamp a message with the office seal."""
    return f"[stamped] {text}"


class Stamps(Extension):
    """A purely additive extension: one tool, one capability entry."""

    identifier = "com.example/stamps"

    def settings(self) -> dict[str, Any]:
        return {"sealed": True}

    def tools(self) -> Sequence[ToolBinding]:
        return [ToolBinding(fn=stamp)]


mcp = MCPServer("post-office", extensions=[Stamps()])


async def main() -> None:
    async with Client(mcp) as client:
        print(client.server_capabilities.extensions)
        # {'com.example/stamps': {'sealed': True}}
        result = await client.call_tool("stamp", {"text": "hello"})
        print(result.content)
        # [TextContent(text='[stamped] hello')]

独自メソッドの提供

拡張機能は新しいリクエストメソッドを登録できます。仕様のメソッドと並んで配信される、独自の動詞(verb)です。

server.py
from collections.abc import Sequence
from typing import Any, Literal

from pydantic import Field

import mcp.types as types
from mcp import Client
from mcp.client import advertise
from mcp.server.context import ServerRequestContext
from mcp.server.extension import Extension, MethodBinding
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/search"


class SearchParams(types.RequestParams):
    query: str
    limit: int = Field(default=10, ge=1, le=100)


class SearchResult(types.Result):
    items: list[str]


class SearchRequest(types.Request[SearchParams, Literal["com.example/search"]]):
    method: Literal["com.example/search"] = "com.example/search"
    params: SearchParams


async def search(ctx: ServerRequestContext[Any, Any], params: SearchParams) -> SearchResult:
    require_client_extension(ctx, EXTENSION_ID)
    return SearchResult(items=[f"{params.query}-{n}" for n in range(params.limit)])


class Search(Extension):
    """An extension that serves its own request method."""

    identifier = EXTENSION_ID

    def methods(self) -> Sequence[MethodBinding]:
        return [
            MethodBinding(
                "com.example/search",
                SearchParams,
                search,
                protocol_versions=frozenset({"2026-07-28"}),
            )
        ]


mcp = MCPServer("catalog", extensions=[Search()])


async def main() -> None:
    async with Client(mcp, extensions=[advertise(EXTENSION_ID)]) as client:
        request = SearchRequest(params=SearchParams(query="mcp", limit=3))
        result = await client.session.send_request(request, SearchResult)
        print(result.items)
        # ['mcp-0', 'mcp-1', 'mcp-2']
  • SearchParamsRequestParams をサブクラス化しているため、2026 の _meta エンベロープが一様にパースされ、ハンドラーが受け取るのは検証済みのパラメーターであって、生の dict ではありません。クライアントが制御できる値には上限を設けてください。Field(ge=1, le=100) は、コードが何かを割り当てる前に、ばかげた limit を拒否します。
  • require_client_extension(ctx, EXTENSION_ID) がゲートです。拡張機能を宣言しなかったクライアントには -32021(必須のクライアントケイパビリティの欠如)エラーが返り、仕様が求める機械可読な requiredCapabilities ペイロードが付きます。
  • protocol_versions=frozenset({"2026-07-28"}) はメソッドを通信路上の 1 つのバージョンに固定します。他のバージョンではクライアントは METHOD_NOT_FOUND を受け取ります。そのバージョンにメソッドが存在しないのとまったく同じです。そのクライアントにとっては、実際に存在しません。

メソッドは厳密に追加のみです。SDK はこれを実行時ではなく構築時に強制します。

  • 仕様で定義されたメソッド(tools/listcompletion/complete など)に対する MethodBinding は、バインディングの構築時に ValueError を送出します。コアの動詞はサーバーのものです。
  • 2 つの拡張機能が同じメソッドをバインドすると、2 つ目の登録時に送出されます。後勝ちはプラグイン同士が互いを壊す原因です。SDK はそうしません。
  • 空の protocol_versions セットも送出します。決して配信できないメソッドはバグであって、設定ではありません。

クライアント側

同じファイルの main() に、クライアント側の話がすべて、その両半分とも入っています。

server.py
from collections.abc import Sequence
from typing import Any, Literal

from pydantic import Field

import mcp.types as types
from mcp import Client
from mcp.client import advertise
from mcp.server.context import ServerRequestContext
from mcp.server.extension import Extension, MethodBinding
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/search"


class SearchParams(types.RequestParams):
    query: str
    limit: int = Field(default=10, ge=1, le=100)


class SearchResult(types.Result):
    items: list[str]


class SearchRequest(types.Request[SearchParams, Literal["com.example/search"]]):
    method: Literal["com.example/search"] = "com.example/search"
    params: SearchParams


async def search(ctx: ServerRequestContext[Any, Any], params: SearchParams) -> SearchResult:
    require_client_extension(ctx, EXTENSION_ID)
    return SearchResult(items=[f"{params.query}-{n}" for n in range(params.limit)])


class Search(Extension):
    """An extension that serves its own request method."""

    identifier = EXTENSION_ID

    def methods(self) -> Sequence[MethodBinding]:
        return [
            MethodBinding(
                "com.example/search",
                SearchParams,
                search,
                protocol_versions=frozenset({"2026-07-28"}),
            )
        ]


mcp = MCPServer("catalog", extensions=[Search()])


async def main() -> None:
    async with Client(mcp, extensions=[advertise(EXTENSION_ID)]) as client:
        request = SearchRequest(params=SearchParams(query="mcp", limit=3))
        result = await client.session.send_request(request, SearchResult)
        print(result.items)
        # ['mcp-0', 'mcp-1', 'mcp-2']
  • Client(..., extensions=[advertise(EXTENSION_ID)]) が拡張機能を宣言します。宣言は ClientCapabilities.extensions になります。2026-07-28 接続では、このマップはリクエストごとの _meta エンベロープで運ばれるため、サーバーはすべてのリクエストでそれを見ます。レガシー接続では initialize ハンドシェイクに載ります。サーバーのコードはどちらでも気にしません。require_client_extension(ctx, ...)ctx.session.check_client_capability(...) は、どちらの経路でも正しい情報源を読みます。
  • ベンダーメソッドは 1 層下がって client.session.send_request(...) を使います。Client がファーストクラスのメソッドを増やすのは仕様の動詞に対してだけです。send_request はどんな Request サブクラスも受け付けるため、ベンダーリクエストはそのまま渡せます。

tools/call のインターセプト

唯一の介入型フックです。ツール呼び出しを監視、短絡、または拒否するには intercept_tool_call をオーバーライドします。

server.py
import logging
from typing import Any

from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer
from mcp.types import CallToolRequestParams

logger = logging.getLogger(__name__)


class AuditLog(Extension):
    """Observe every tools/call without touching its result."""

    identifier = "com.example/audit"

    async def intercept_tool_call(
        self,
        params: CallToolRequestParams,
        ctx: ServerRequestContext[Any, Any],
        call_next: CallNext,
    ) -> HandlerResult:
        logger.info("tool %r called", params.name)
        return await call_next(ctx)


mcp = MCPServer("audited", extensions=[AuditLog()])


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b
  • params は検証済みの CallToolRequestParams です。生の JSON に触れずに params.nameparams.arguments が手に入ります。どのツール呼び出しが実行されるかを決めるのもこれです。書き換えたコンテキストを call_next に渡して変わるのは、ハンドラーが ctx 上で観測するものであり、ツールの呼び出しではありません。通信路レベルのリクエスト書き換えはミドルウェアの仕事です。
  • call_next(ctx) はチェーンの残りを実行し、ハンドラーの結果を返します。そのまま返す(監視)、別のものを返す(置換)、または MCPError を送出する(拒否)のいずれかです。何を返しても、2026 年世代の serverInfo アイデンティティスタンプを含め、ハンドラーの結果と同様にシリアライズされるため、短絡するインターセプターが匿名またはスキーマ外のレスポンスを生み出すことはありません。
  • 複数の拡張機能がある場合、インターセプターは登録順にネストします。extensions=[...] の最初の拡張機能が最も外側です。
  • デフォルトの実装は素通しで、拡張機能がこのフックを一切オーバーライドしないサーバーでは、素の tools/call ハンドラーがそのまま保たれます。使わないもののコストを払うことはありません。

このフックがラップするのは tools/call だけです。すべてのメッセージに関わる処理にはミドルウェアを使ってください。それがミドルウェアの役目です。

クライアント拡張機能を使う

クライアント拡張機能は、同じ契約を利用する側から見たものです。1 つの識別子の下にまとめられたクライアント側の振る舞い一式です。インスタンスを Client(extensions=[...]) に渡し、通常どおりツールを呼び出します。

client.py
from collections.abc import Sequence
from typing import Any, Literal

import mcp.types as types
from mcp import Client
from mcp.client import ClaimContext, ClientExtension, ResultClaim
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/receipts"


class ReceiptResult(types.Result):
    """The claimed result shape; `result_type` pins the wire tag."""

    result_type: Literal["receipt"] = "receipt"
    receipt_token: str


class ReceiptIssuer(Extension):
    """Server half: answers `buy` with a receipt instead of a final result."""

    identifier = EXTENSION_ID

    async def intercept_tool_call(
        self,
        params: types.CallToolRequestParams,
        ctx: ServerRequestContext[Any, Any],
        call_next: CallNext,
    ) -> HandlerResult:
        if params.name != "buy":
            return await call_next(ctx)
        require_client_extension(ctx, EXTENSION_ID)
        return {"resultType": "receipt", "receiptToken": "r-117"}


class Receipts(ClientExtension):
    """Client half: claims the `receipt` shape and supplies the code that finishes it."""

    identifier = EXTENSION_ID

    def claims(self) -> Sequence[ResultClaim[Any]]:
        return [ResultClaim(result_type="receipt", model=ReceiptResult, resolve=self._redeem)]

    async def _redeem(self, claimed: ReceiptResult, ctx: ClaimContext) -> types.CallToolResult:
        return await ctx.session.call_tool("redeem", {"token": claimed.receipt_token})


mcp = MCPServer("shop", extensions=[ReceiptIssuer()])


@mcp.tool()
def buy(item: str) -> types.CallToolResult:
    """Buy an item."""
    raise NotImplementedError  # ReceiptIssuer answers `buy` before the tool runs


@mcp.tool()
def redeem(token: str) -> str:
    """Exchange a receipt token for the goods."""
    return f"goods for {token}"


async def main() -> None:
    async with Client(mcp, extensions=[Receipts()]) as client:
        result = await client.call_tool("buy", {"item": "lamp"})
        print(result.content)
        # [TextContent(text='goods for r-117')]

call_tool("buy", ...) は、他のすべての呼び出しと同様にプレーンな CallToolResult を返します。拡張機能が変えたのは次の点です。サーバーは buy に対して、最終結果の代わりに receipt という結果の形状で答えられるようになり、call_tool が戻る前に Receipts がそれを完了させます(ここでは後続の呼び出しでレシートを引き換えます)。呼び出し側のコードは何も変わりません。

拡張機能を外せば、このどれも存在しません。サーバーのゲートは宣言しなかったクライアントを拒否し(エラー -32021)、ゲートを省いたサーバーから届いた引き受け対象の形状は検証に失敗します。認識できない resultType に対して仕様が求めるとおりです。通信路の両端で、デフォルトはオフです。

クライアント側の振る舞いを一切持たない識別子をアドバタイズするには(サーバーがケイパビリティでゲートし、クライアントは何もしない、上の検索クライアントのような場合)、advertise() を使います。

from mcp.client import advertise

client = Client(mcp, extensions=[advertise("com.example/search")])

クライアント拡張機能を書く

ClientExtension をサブクラス化し、必要なものだけをオーバーライドします。提供できるものは 3 種類で、それぞれにデフォルトがあります。settings()claims()notifications() です。

client.py
from collections.abc import Sequence
from typing import Any, Literal

import mcp.types as types
from mcp import Client
from mcp.client import ClaimContext, ClientExtension, ResultClaim
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/receipts"


class ReceiptResult(types.Result):
    """The claimed result shape; `result_type` pins the wire tag."""

    result_type: Literal["receipt"] = "receipt"
    receipt_token: str


class ReceiptIssuer(Extension):
    """Server half: answers `buy` with a receipt instead of a final result."""

    identifier = EXTENSION_ID

    async def intercept_tool_call(
        self,
        params: types.CallToolRequestParams,
        ctx: ServerRequestContext[Any, Any],
        call_next: CallNext,
    ) -> HandlerResult:
        if params.name != "buy":
            return await call_next(ctx)
        require_client_extension(ctx, EXTENSION_ID)
        return {"resultType": "receipt", "receiptToken": "r-117"}


class Receipts(ClientExtension):
    """Client half: claims the `receipt` shape and supplies the code that finishes it."""

    identifier = EXTENSION_ID

    def claims(self) -> Sequence[ResultClaim[Any]]:
        return [ResultClaim(result_type="receipt", model=ReceiptResult, resolve=self._redeem)]

    async def _redeem(self, claimed: ReceiptResult, ctx: ClaimContext) -> types.CallToolResult:
        return await ctx.session.call_tool("redeem", {"token": claimed.receipt_token})


mcp = MCPServer("shop", extensions=[ReceiptIssuer()])


@mcp.tool()
def buy(item: str) -> types.CallToolResult:
    """Buy an item."""
    raise NotImplementedError  # ReceiptIssuer answers `buy` before the tool runs


@mcp.tool()
def redeem(token: str) -> str:
    """Exchange a receipt token for the goods."""
    return f"goods for {token}"


async def main() -> None:
    async with Client(mcp, extensions=[Receipts()]) as client:
        result = await client.call_tool("buy", {"item": "lamp"})
        print(result.content)
        # [TextContent(text='goods for r-117')]
  • 識別子はサーバー側と同じ文法に従い、クラスの定義時に検証されます。
  • claims()ResultClaim を返します。通信上のタグ、それをパースするモデル、それを完了させるリゾルバーの組です。モデルは result_type: Literal["receipt"] でタグを固定しなければならず、その動詞のコア結果型をサブクラス化してはいけません。どちらも引き受けの構築時に強制されます。receipt_token のようなベンダーフィールドはそのまま通信路を流れます。差し替えられた形状はそのままの形でクライアントに届きます。
  • リゾルバーはパース済みのモデルと ClaimContext を受け取ります。ctx.sessionclient.session と同じ公開ハンドルなので、後続の処理は通常のセッション呼び出しです。戻り値はその動詞の通常の CallToolResult です。
  • settings()ClientCapabilities.extensions[identifier] にアドバタイズされる値で、Client の構築時に一度だけ読み取られます。

notifications() は、監視するベンダーのサーバー通知を宣言します。

def notifications(self) -> Sequence[NotificationBinding[Any]]:
    return [NotificationBinding(method="notifications/receipts", params_type=ReceiptEvent, handler=self.on_receipt)]

ハンドラーは検証済みのパラメーターをディスパッチ順に 1 つずつ受け取ります。監視するだけで、拒否も返信もできません。

目立たないルールが 2 つあります。引き受けが有効なのは 2026-07-28 接続だけで、ケイパビリティのアドバタイズもそれに従います。レガシー接続では引き受けは消え、識別子も一緒にアドバタイズから外れるため、自分が拒否してしまう形状を持つ拡張機能をクライアントがアドバタイズすることはありません。また、リゾルバーではなく自分で引き受け対象の形状を受け取りたいときは、client.session.call_tool(..., allow_claimed=True) を呼び出してください。このフラグがないと、セッション層の呼び出し側に届いた引き受け対象の形状は UnexpectedClaimedResult を送出します。

拡張機能の動詞

拡張機能独自のリクエストメソッドには、クライアント側の登録は不要です。ベンダーリクエスト型は mcp.types.Request をサブクラス化し、独自メソッドの提供と同様に client.session.send_request を通ります。追加が 1 つあります。パラメーターのキーを Mcp-Name ヘッダーに載せなければならない場合(tasks のような拡張機能の仕様では、その動詞にこれが必要です)、リクエスト型は name_param を宣言します。

client.py
from collections.abc import Sequence
from typing import Any, Literal

import mcp.types as types
from mcp import Client
from mcp.client import advertise
from mcp.server.context import ServerRequestContext
from mcp.server.extension import Extension, MethodBinding
from mcp.server.mcpserver import MCPServer

EXTENSION_ID = "com.example/jobs"


class JobParams(types.RequestParams):
    job_id: str


class JobStatus(types.Result):
    status: str


class JobStatusRequest(types.Request[JobParams, Literal["com.example/jobs.status"]]):
    method: Literal["com.example/jobs.status"] = "com.example/jobs.status"
    params: JobParams
    name_param = "jobId"  # params["jobId"] rides the Mcp-Name header


async def job_status(ctx: ServerRequestContext[Any, Any], params: JobParams) -> JobStatus:
    return JobStatus(status=f"{params.job_id} is running")


class Jobs(Extension):
    """An extension whose verb names its subject, so the header can route on it."""

    identifier = EXTENSION_ID

    def methods(self) -> Sequence[MethodBinding]:
        return [MethodBinding("com.example/jobs.status", JobParams, job_status)]


mcp = MCPServer("worker", extensions=[Jobs()])


async def main() -> None:
    async with Client(mcp, extensions=[advertise(EXTENSION_ID)]) as client:
        request = JobStatusRequest(params=JobParams(job_id="job-7"))
        result = await client.session.send_request(request, JobStatus)
        print(result.status)
        # job-7 is running

セッションはどの送信経路でも params["jobId"]Mcp-Name に反映し、値が欠けている場合は必須ヘッダーを黙って省くのではなく、はっきりとエラーになります。

拡張機能にできないこと

提供できる範囲は意図的に閉じています。サーバー側では、設定、ツール、リソース、メソッド、tools/call のインターセプター 1 つ。クライアント側では、設定、結果の引き受け、通知のバインディング。拡張機能には次のことができません。

  • ホストの内部に手を伸ばすこと。 データを宣言するだけで、サーバーやクライアントへの参照は持ちません。
  • コアの振る舞いを置き換えること。 仕様のメソッドとコアの結果タグは構築時に拒否されます(initialize はランナーが完全に予約しています)。コアの語彙に隠れた通知バインディングは、代わりに警告を出して沈黙します。
  • 後から登録すること。 MCPServer(...)Client(...) が戻った後は、拡張機能の集合はそのまま確定です。

これらの壁と戦っているなら、書いているのは拡張機能ではありません。フォークです。壁こそが機能です。extensions=[Apps(), Stamps()] を読んだユーザーは、この 2 つが触れた可能性のあるものを「すべて」把握できます。