サブスクリプション
サーバーのカタログは固定ではありません。ツールは実行時に現れ、リソース URI の背後にある内容は変化します。クライアントはそれを client.listen(...) を通じて知ります。これは 1 つの subscriptions/listen リクエストで、そのレスポンスが「そのまま」ストリームになります。ストリームは開いたままになり、クライアントが求めた変更通知を運びます。
このページはクライアント側の話です。ストリームを開き、メインの処理の横で監視し、その終わり方を扱います。変更の公開、フィルタリング、このメソッドの提供はサーバー側の話で、「ハンドラーの中で」にある サブスクリプション で説明しています。ここでの例は、そこで作ったスプリントボードサーバーと通信します。
ストリームを監視する
サブスクリプションは 1 つのコンテキストマネージャーです。中に入るとリクエストが送られ、キーワード引数がサブスクリプションのフィルターになります。そのうえでサーバーの確認応答を待つので、ブロックが始まる時点でストリームはすでに稼働しています。
from mcp import Client
from mcp.client.subscriptions import ResourceUpdated, ToolsListChanged
from mcp.types import TextResourceContents
BOARD = "board://sprint"
async def read_board(client: Client, uri: str = BOARD) -> str:
[contents] = (await client.read_resource(uri)).contents
assert isinstance(contents, TextResourceContents)
return contents.text
async def follow_board(client: Client) -> None:
async with client.listen(tools_list_changed=True, resource_subscriptions=[BOARD]) as sub:
async for event in sub:
match event:
case ResourceUpdated(uri=uri):
print(await read_board(client, uri))
case ToolsListChanged():
tools = await client.list_tools()
print("tools:", [tool.name for tool in tools.tools])
case _:
pass # kinds the filter did not ask for never arrive
async def main() -> None:
async with Client("http://localhost:8000/mcp") as client:
await follow_board(client)
反復処理では 4 種類の型付きイベントが得られます。ToolsListChanged、PromptsListChanged、ResourcesListChanged、ResourceUpdated(uri=...) です。
イベントが伝えるのは「何が」変わったかであり、「どう」変わったかではありません。follow_board が read_resource と list_tools を呼ぶのはそのためで、イベントは再取得の合図です。どのリソースが動いたかを決めつけず、event.uri を読んでください。フィルターは複数の URI を指定でき、サーバーはそのうちの 1 つのサブリソースに対する変更を報告することもあります。
消費待ちの重複したイベントは 1 つにまとめられますが、再取得すれば現在の状態は得られます。まとめられるのは同一のイベントだけです。異なる URI に対する 2 つの ResourceUpdated は 2 つのイベントです。
ハンドルにはさらに 2 つのプロパティがあります。
sub.honoredはサーバーが確認応答したフィルターです。渡したフィールドを持つSubscriptionFilterで、属性として読めます（sub.honored.prompts_list_changed）。MCPServerは求めた種類をすべて受け入れるので、リクエストをそのまま返します。対応する種類が少ないサーバーは確認応答する内容も少なくなり、受け入れられた種類でも一度も発火しないことがあります。サーバーは確認応答する代わりにリクエスト全体を拒否することもあり（サーバー側のページの誰が監視できるかを決めるを参照）、その場合はリクエストのエラーとして表面化します。sub.subscription_idは listen リクエストの ID で、このストリームのすべてのフレームに刻まれます。複数のサブスクリプションを同時に開くことができ、それぞれが自分の ID で多重分離されます。
ブロックせずに監視する
follow_board はサーバーがストリームを閉じるまで動き続けます。それは永遠に来ないかもしれないので、単独で動かすとプログラムを占有してしまいます。実際のクライアントが欲しいのは、メインの処理の「横で」動くウォッチャーです。エージェントがツールを呼ぶ一方で、ウォッチャーがキャッシュや UI を最新に保ちます。
まずサブスクリプションを開き、それからウォッチャーを起動して本来の作業に取りかかってください。
import asyncio
from mcp import Client
from mcp.client.subscriptions import Subscription
from .tutorial003 import BOARD, read_board
async def watch(client: Client, sub: Subscription) -> None:
async for _event in sub:
board = await read_board(client)
print(board)
if "[ ]" not in board:
return # sprint finished: the stream closes when run_sprint leaves the block
async def run_sprint(client: Client) -> None:
async with client.listen(resource_subscriptions=[BOARD]) as sub:
print(await read_board(client)) # snapshot: acknowledged, so nothing after this is missed
watcher = asyncio.create_task(watch(client, sub))
for task in ("design", "build", "ship"):
await client.call_tool("complete_task", {"board": "sprint", "task": task})
await watcher # returns once the watcher has seen the finished board
async def main() -> None:
async with Client("http://localhost:8000/mcp") as client:
await run_sprint(client)
if __name__ == "__main__":
asyncio.run(main())
import trio
from mcp import Client
from mcp.client.subscriptions import Subscription
from .tutorial003 import BOARD, read_board
async def watch(client: Client, sub: Subscription) -> None:
async for _event in sub:
board = await read_board(client)
print(board)
if "[ ]" not in board:
return # sprint finished: the stream closes when run_sprint leaves the block
async def run_sprint(client: Client) -> None:
async with client.listen(resource_subscriptions=[BOARD]) as sub:
print(await read_board(client)) # snapshot: acknowledged, so nothing after this is missed
async with trio.open_nursery() as nursery:
nursery.start_soon(watch, client, sub)
for task in ("design", "build", "ship"):
await client.call_tool("complete_task", {"board": "sprint", "task": task})
async def main() -> None:
async with Client("http://localhost:8000/mcp") as client:
await run_sprint(client)
if __name__ == "__main__":
trio.run(main)
import anyio
from mcp import Client
from mcp.client.subscriptions import Subscription
from .tutorial003 import BOARD, read_board
async def watch(client: Client, sub: Subscription) -> None:
async for _event in sub:
board = await read_board(client)
print(board)
if "[ ]" not in board:
return # sprint finished: the stream closes when run_sprint leaves the block
async def run_sprint(client: Client) -> None:
async with client.listen(resource_subscriptions=[BOARD]) as sub:
print(await read_board(client)) # snapshot: acknowledged, so nothing after this is missed
async with anyio.create_task_group() as tg:
tg.start_soon(watch, client, sub)
for task in ("design", "build", "ship"):
await client.call_tool("complete_task", {"board": "sprint", "task": task})
async def main() -> None:
async with Client("http://localhost:8000/mcp") as client:
await run_sprint(client)
if __name__ == "__main__":
anyio.run(main)
Note
app.py は最初の例から BOARD と read_board をインポートします。このリポジトリではその例を tutorial003.py として保存しています。レンダリングされたファイルを client.py と app.py として並べて保存する場合は、代わりに from client import BOARD, read_board と書いてください。さらに下の watch.py の例も同じように read_board をインポートします。
重要なのは順序です。何も再送されないので、ストリームができる前に公開されたイベントは取りこぼします。client.listen(...) に入ると確認応答を待つので、その瞬間以降のすべての変更はウォッチャーに届きます。ブロックの中で取るスナップショットが変更を取りこぼすことはありません。
リクエストは開いたストリームの横で自由に実行できます。ウォッチャーのタスクからでも他のタスクからでも、同じクライアント上で構いません。未消費の「重複した」イベントはまとめられるので、メインの処理が忙しいときは再取得が 3 回ではなく 1 回で済むこともあります。異なるイベントはまとめられません。多くの URI を指定したフィルターでは、URI ごとに保留中のイベントが 1 つずつキューに入ります。
監視をやめるにはブロックを抜けます。unsubscribe の呼び出しはありません。ブロックを所有するタスクをキャンセルすればそうなり、SDK はトランスポートが期待する方法で listen リクエストをキャンセルします。Streamable HTTP では、そのリクエストのストリームを閉じます。アプリの寿命のあいだ動き続けるウォッチャーは自分からは返らないので、シャットダウン時にそのタスク、またはタスクグループのスコープをキャンセルしてください。
ストリームの終わり
ストリームの終わり方は 2 通りあり、どちらも通常の制御フローです。サーバーが正常に閉じると async for が終わり、突然切れると SubscriptionLost が送出されます。
この違いは診断上のもので、次に何をすべきかの違いではありません。ストリームはなくなり、何も再送されず、まだ関心のあるウォッチャーは改めて listen して再取得します。
import anyio
from mcp import Client
from mcp.client.subscriptions import SubscriptionLost
from .tutorial003 import read_board
async def keep_following(client: Client) -> None:
while True:
try:
async with client.listen(resource_subscriptions=["board://sprint"]) as sub:
print(await read_board(client)) # refetch: no replay across streams
async for _event in sub:
print(await read_board(client))
except SubscriptionLost:
pass
# Either ending means the stream is gone. Back off before re-listening:
# a graceful close may be the server shedding load.
await anyio.sleep(1)
サーバーは自分の都合でストリームを正常に閉じます。バックログが大きくなりすぎた購読者を切り離す場合もそうです。そのため、きれいに終わったことは監視をやめる合図ではありません。改めて listen する前にバックオフしてください。
SubscriptionLost にはローカルな原因も 1 つあります。クライアントが保持する未消費イベントは最大 1024 件で、そこまで遅れた消費側は際限なく膨らむ代わりにサブスクリプションを失います。async for の本体は短く保ち、時間のかかる作業は別の場所で行ってください。
keep_following が捕捉するのは SubscriptionLost だけです。listen() に入るときには、MCPError（接続に失敗した、またはサーバーがこのメソッドを提供していない）、TimeoutError（確認応答が届かなかった）、ListenNotSupportedError（2026 年より前の接続）が送出されることもあります。ウォッチャーがそのうちどれを再試行すべきかを決めてください。最後のものは決して回復しません。
まとめ
async with client.listen(...)に入ります。入ると確認応答を待つので、その後に公開されたものを取りこぼすことはありません。async for event in subで反復します。イベントは再取得の合図であって、ペイロードではありません。- サブスクリプションを開いてからウォッチャーをタスクとして実行すれば、その横でツール呼び出しが流れ続けます。
- きれいに終わればループが止まり、切れれば
SubscriptionLostが送出されます。どちらの場合も、改めて listen して再取得します。その前にまずバックオフしてください。 - ブロックを抜けることがサブスクリプションの解除です。
これらのイベントの公開、フィルターの絞り込み、1 プロセスを超えたスケーリングはサーバー側の話です。詳しくは サブスクリプション を参照してください。同じイベントはクライアント側のキャッシュを正確に保つのにも役立ちます。次のページは キャッシュ です。