Python SDK 客户端订阅指南：用 `client.listen(...)` 实时监听 MCP 服务器目录变更

• 人工智能
• MCP 服务
• MCP Clients

服务器目录并非一成不变：工具会在运行时出现，资源 URI 背后的内容也会发生变化。本文聚焦 Python SDK 客户端侧的订阅能力，讲解如何通过一次subscriptions/listen请求打开一条持续推送变更通知的流（stream），如何在主流程之外并行监听、处理流的各种结束方式，以及基于仓库源码理解其去重、限流与协议版本约束等底层机制。读完你将能独立编写一个可靠的客户端订阅观察者，并在断流后正确重连。

一条请求即一条流：订阅机制概览

在 MCP 生态中，"订阅"描述的是客户端感知服务器目录变化的能力。它的核心模型很简单：客户端发送一次subscriptions/listen请求，而这次请求的响应本身就是那条流——它不会像普通 JSON-RPC 请求那样立即返回结果，而是保持打开状态，持续承载客户端所请求的那几类变更通知。

这条流从创建到结束都由一个异步上下文管理器（async context manager）管理。进入async with client.listen(...)块会发出请求（你传入的关键字参数即为订阅过滤器），并等待服务器的确认（acknowledgment）——也就是说，等到代码块真正开始执行时，流已经是"活的"，服务器确认之后才发布的任何变更都不会漏掉。

从源码可以看到整个契约的完整描述，src/mcp/client/subscriptions.py 顶部模块注释写道：

listen()opens the stream as an async context manager: entering waits for the server's acknowledgment, iteration yields typed change events, a graceful server close ends the loop, and an abrupt drop raisesSubscriptionLost. There is no replay and no automatic re-listen.

即：进入即等待确认、迭代产出类型化事件、优雅关闭结束循环、突然断开抛出SubscriptionLost；没有回放、也没有自动重连——需要重新订阅的客户端必须自行重新获取它所依赖的数据。

需要注意的协议版本前提：subscriptions/listen是 2026-07-28 协议版本的能力。若协商出的协议版本更早，调用会在入口处直接抛出ListenNotSupportedError（详见下文"入口处的三类异常"）。2025 时代的resources/subscribe旧路径由ctx.session.send_resource_updated(uri)服务，与本文的notify_*通知流是两条互不相干的通道。发布变更、过滤与在服务器侧服务该方法，属于 docs/handlers/subscriptions.md（"Inside your handler" 部分）讲述的另一半故事；本文示例所对话的，正是该页构建的 sprint 看板服务器。

打开并监听订阅流：一个完整的客户端示例

仓库中的 docs_src/subscriptions/tutorial003.py 演示了订阅客户端最完整的形态——订阅资源变更与工具列表变更，并在收到事件时重新拉取数据。假设你已经运行了该页服务器端示例（暴露在http://localhost:8000/mcp），可以原样运行：

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)

四种类型化事件

对sub的迭代会产出四种类型化事件（type 均在 src/mcp/client/subscriptions.py 中定义并导出）：

示例代码使用match ... case按事件类型分发，这正是推荐做法：事件只告诉你"什么"变了，从不告诉你"怎么"变的。正因如此，follow_board在收到ResourceUpdated后要调用read_resource、收到ToolsListChanged后要调用list_tools——事件只是一个"重新拉取数据"的提示信号（cue），绝不是携带新数据的载荷（payload）。

不要臆测 URI，读取event.uri

处理ResourceUpdated时，请直接读取event.uri，而不是假设哪条资源动了。原因有二：

• 一个过滤器可以同时命名多个 URI；
• 协议允许服务器在订阅 URI 的子资源上报告变更。

从 src/mcp/client/subscriptions.py 的ListenRoute.deliver可以看到，客户端在收包侧就已经接受了这种宽容语义：只要"资源订阅这一大类"被服务器确认过（self._honored_uris非空），任何ResourceUpdated都会被放行，因为携带的 URI 可能是某个已订阅 URI 的子资源。换言之，客户端永远无法假设事件里的 URI 恰好等于你请求的那几个。

重复事件合并

多个尚未被消费的重复事件会合并为一个：因为事件只是"去重新拉取"的信号，重复的信号没有意义——重新拉取一次就能拿到当前状态。注意，只有完全相同的事件才合并：两个针对不同 URI 的ResourceUpdated是两个独立事件。这个语义同样落实在源码中：ListenRoute用self._pending: dict[ServerEvent, None]作为待处理队列，入队前先检查event in self._pending，重复即丢弃；而ServerEvent是四个小 dataclass，其相等性由字段值决定。

句柄（handle）的两个附加属性

上下文管理器产出的sub对象还暴露两个有用属性：

• sub.honored：服务器确认过的过滤器，一个SubscriptionFilter，包含你传入的字段，并可直接作为属性读取（如sub.honored.prompts_list_changed）。MCPServer会满足你请求的每一种事件，所以它会把你的请求原样回显；只支持更少事件类型的服务器会确认得更少，而被确认的类型也可能永远不会触发。此外，服务器可以整体拒绝请求而不是确认它（见服务器页 docs/handlers/subscriptions.md#deciding-who-may-watch 的"Deciding who may watch"一节），这将以请求的错误形式表现出来。
• sub.subscription_id：这次 listen 请求的 JSON-RPC id，它会被盖在这条流的每一帧上（_meta字段中），用于多路复用。多个订阅可以同时打开，各自靠自己的 id 解复用。从源码看，Python 客户端使用进程内递增的字符串 id：_listen_ids = count(1)，生成形如"listen-1"、"listen-2"的 id（模块注释明确说明：字符串 id 永远不会与 dispatcher 铸造的整数 id 冲突）。

不阻塞主流程：把观察者放在业务旁边

follow_board会一直运行到服务器关闭流为止，而服务器可能永远不关——单独运行它等于独占整个程序。真实客户端想要的模式是让观察者并行于主流程：Agent 在调用工具的同时，一个观察者任务在后台维护缓存或界面。

做法是：先打开订阅，再启动观察者任务，然后继续干正事。仓库 docs_src/subscriptions 下提供了 asyncio、trio、anyio 三个等价版本。

asyncio 版本（tutorial004_asyncio.py）

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())

trio 版本（tutorial004_trio.py）

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)

anyio 版本（tutorial004_anyio.py）

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())

关于导入路径的说明：仓库把三个app.py都存储为tutorial004_*.py，它们从第一个示例（仓库中名为tutorial003.py）导入BOARD和read_board。如果你把文中渲染的示例按client.py与app.py并排保存，请把导入写成from client import BOARD, read_board。下面watch.py的例子也同样从tutorial003.py导入read_board。

顺序就是一切

这个模式里，先后顺序是全部要点：进入client.listen(...)会等待服务器确认，因此从那一刻起发生的每个变更都能到达观察者；随后在块内拍摄的"快照"（print(await read_board(client))）不可能漏掉任何一次变更。而如果反过来——先启动观察者再打开订阅——由于没有任何回放，在流存在之前发布的事件就永久丢失了。

请求与流并行

在一条已打开的流旁边，其他请求可以自由执行——无论是来自观察者任务自身，还是来自同一 client 上的任何其他任务。前面说过，重复的未消费事件会合并，所以即使主流程很繁忙，底层也只会产生一次重新拉取而不是三次；而不同的事件不会合并——一个命名了许多 URI 的过滤器，会为每个 URI 各排一个待处理事件。

停止监听：退出块就是退订

停止监听的方式是退出async with块——没有unsubscribe调用。取消持有该块的任务，SDK 会替你取消 listen 请求，并按传输层的预期方式收尾：在 Streamable HTTP 上，表现为关闭该请求对应的流（见 src/mcp/client/subscriptions.py 的listen实现：finally中调用route.settle("local")并取消驱动任务、注销路由）。

一个细节值得注意：运行期与应用同寿命的观察者永远不会自行返回，所以在应用关闭时必须显式取消它（或取消其所属任务组的范围），否则程序无法干净退出。

流的结束：两种收尾与重连策略

一条流只以两种方式结束，而它们都属于普通控制流的范畴：

从源码看，src/mcp/client/subscriptions.py 的Subscription.__anext__正是这样区分的：ListenRoute.next_event返回字符串结局标记，"lost"时抛出SubscriptionLost（并链上原始错误），"graceful"/"local"时抛出StopAsyncIteration结束循环。同时，驱动任务（drive()）里有一条重要注释："A result, whatever its body, is the spec's graceful close"——即协议中，listen 请求的结果帧本身就是服务器有意结束订阅的表示；若在收到确认之前就收到结果，则订阅以"已关闭"状态打开。

两种结束方式的差异只用于诊断，不改变接下来该做什么：流没了、什么都没回放，仍然关心的观察者就应该重新监听、重新拉取。

实战示例：watch.py 与退避重连

仓库中的 docs_src/subscriptions/tutorial005.py 演示了健壮的重连循环：

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)

两个关键点：

• 优雅关闭不代表要停止监听。服务器可能出于自身原因（例如要卸掉一个积压（backlog）过大的订阅者）主动关闭流，所以"干净地结束"不是放弃监听的信号；正确的姿势是先退避（back off）再重新监听，示例中统一anyio.sleep(1)。
• SubscriptionLost也有一个本地成因：客户端最多保留 1024 个未消费事件（源码中的常量_MAX_PENDING_EVENTS = 1024，注释说明：协议允许子资源 URI，因此不同的ResourceUpdated可能无界增长，超过该上限就"宁可让订阅丢失"也不让客户端内存无限膨胀；触发时ListenRoute.deliver会以INTERNAL_ERROR结束流，消息为 "subscription backlog exceeded 1024 unconsumed events; re-listen and refetch"）。一个落后太多的消费者会因此失去订阅而不是无限增长下去。所以请保持async for循环体短小，把慢速工作放到别处去做。

入口处的三类异常

keep_following只捕获SubscriptionLost，但进入listen()本身还可能抛出另外三种异常，需要按需决定观察者是否重试：

• MCPError：连接失败，或服务器不提供该方法（也可能在流确认前连接中断）；
• TimeoutError：在会话读超时（源码中对应session._session_read_timeout_seconds）内没有收到确认；
• ListenNotSupportedError：协商出的协议版本早于 2026（源码 src/mcp/client/subscriptions.py 中ListenNotSupportedError的报错信息明确提示：subscriptions/listen要求 2026-07-28，早期版本请改用subscribe_resource()与经message_handler送达的变更通知）。

策略建议：前两类可能随时间自愈，可以纳入重试；最后一类永远不会自愈，不应盲目重试，而应检查客户端与服务器协商的协议版本。

小结

• 用async with client.listen(...)打开订阅；进入即等待确认，所以其后发布的事件一个都不会漏。
• 用async for event in sub迭代。事件是重新拉取的信号，永远不是数据载荷。
• 先打开订阅，再把观察者作为任务启动，工具调用就能在旁边继续流动。
• 干净结束会停止循环，突然断开会抛出SubscriptionLost。无论哪种：先退避，再重听，再重取。
• 退出块就是退订——没有unsubscribe调用。

延伸阅读

• 在服务器侧发布这些事件（ctx.notify_resource_updated、ctx.notify_tools_changed等）、限制过滤器以及跨进程扩展（实现SubscriptionBus），见 docs/handlers/subscriptions.md（"Inside your handler" 部分；其中的过滤器确认、订阅 id 盖帧等线上细节与本文示例一一对应）。
• 服务器决定"谁能看"的中间件门禁与拒绝语义，见 docs/handlers/subscriptions.md#deciding-who-may-watch。
• 这些同样的事件还会维持客户端缓存的新鲜度——利用client.listen(on_event=...)的屏障钩子在消费者重新拉取前完成缓存驱逐（源码Subscription.__anext__中on_event的语义），这是下一页 docs/client/caching.md 的主题。
• 本文引用的三个官方示例分别位于 docs_src/subscriptions/tutorial003.py、docs_src/subscriptions/tutorial004_asyncio.py（及 trio/anyio 两个姊妹文件）与 docs_src/subscriptions/tutorial005.py，客户端侧完整实现见 src/mcp/client/subscriptions.py。

• 人工智能
• MCP 服务
• MCP Clients