拡張機能
拡張機能とは、1 つの識別子の下にまとめられた、オプトイン式の MCP の振る舞い一式です。
サーバー側では、ツール、リソース、新しいリクエストメソッドを提供でき、tools/call をラップすることもできます。クライアント側では、追加の tools/call の結果形状を引き受け(claim)、ベンダー通知を監視できます。それぞれの側が自分の capabilities.extensions でアドバタイズし、求めなかった人にとっては何も変わりません。これが契約です(SEP-2133)。そして黄金律が 1 つあります。拡張機能はデフォルトでオフです。
拡張機能を使う
構築時にインスタンスを渡します。
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 つです。
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 に直接つなぐインメモリのクライアントです。
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)です。
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']
SearchParamsはRequestParamsをサブクラス化しているため、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/list、completion/completeなど)に対するMethodBindingは、バインディングの構築時にValueErrorを送出します。コアの動詞はサーバーのものです。 - 2 つの拡張機能が同じメソッドをバインドすると、2 つ目の登録時に送出されます。後勝ちはプラグイン同士が互いを壊す原因です。SDK はそうしません。
- 空の
protocol_versionsセットも送出します。決して配信できないメソッドはバグであって、設定ではありません。
クライアント側
同じファイルの main() に、クライアント側の話がすべて、その両半分とも入っています。
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 をオーバーライドします。
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.nameとparams.argumentsが手に入ります。どのツール呼び出しが実行されるかを決めるのもこれです。書き換えたコンテキストをcall_nextに渡して変わるのは、ハンドラーがctx上で観測するものであり、ツールの呼び出しではありません。通信路レベルのリクエスト書き換えはミドルウェアの仕事です。call_next(ctx)はチェーンの残りを実行し、ハンドラーの結果を返します。そのまま返す(監視)、別のものを返す(置換)、またはMCPErrorを送出する(拒否)のいずれかです。何を返しても、2026 年世代のserverInfoアイデンティティスタンプを含め、ハンドラーの結果と同様にシリアライズされるため、短絡するインターセプターが匿名またはスキーマ外のレスポンスを生み出すことはありません。- 複数の拡張機能がある場合、インターセプターは登録順にネストします。
extensions=[...]の最初の拡張機能が最も外側です。 - デフォルトの実装は素通しで、拡張機能がこのフックを一切オーバーライドしないサーバーでは、素の
tools/callハンドラーがそのまま保たれます。使わないもののコストを払うことはありません。
このフックがラップするのは tools/call だけです。すべてのメッセージに関わる処理にはミドルウェアを使ってください。それがミドルウェアの役目です。
クライアント拡張機能を使う
クライアント拡張機能は、同じ契約を利用する側から見たものです。1 つの識別子の下にまとめられたクライアント側の振る舞い一式です。インスタンスを Client(extensions=[...]) に渡し、通常どおりツールを呼び出します。
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() です。
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.sessionはclient.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 を宣言します。
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 つが触れた可能性のあるものを「すべて」把握できます。