2026/9/20 17:17:29
FEATURED · 精选文章 ` 实时监听 MCP 服务器目录变更)
人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载服务器目录并非一成不变工具会在运行时出现资源 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 servers 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.mdInside 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_changedTrue, resource_subscriptions[BOARD]) as sub: async for event in sub: match event: case ResourceUpdated(uriuri): 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 中定义并导出事件类型含义ToolsListChanged工具列表发生变化PromptsListChanged提示词prompt列表发生变化ResourcesListChanged资源列表发生变化ResourceUpdated(uri...)某个资源 URI 背后的内容发生变化uri字段指明是哪个示例代码使用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.pyimport 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.pyimport 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.pyimport 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)并取消驱动任务、注销路由。一个细节值得注意运行期与应用同寿命的观察者永远不会自行返回所以在应用关闭时必须显式取消它或取消其所属任务组的范围否则程序无法干净退出。流的结束两种收尾与重连策略一条流只以两种方式结束而它们都属于普通控制流的范畴结束方式表现服务器优雅关闭graceful closeasync for循环自然结束StopAsyncIteration突然断开abrupt drop抛出SubscriptionLost从源码看src/mcp/client/subscriptions.py 的Subscription.__anext__正是这样区分的ListenRoute.next_event返回字符串结局标记lost时抛出SubscriptionLost并链上原始错误graceful/local时抛出StopAsyncIteration结束循环。同时驱动任务drive()里有一条重要注释A result, whatever its body, is the specs 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.mdInside 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【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载相关推荐使用 awesome-copilot 的 appinsights-instrumentation 技能为 Web 应用接入 Azure Application Insights 遥测使用 awesome copilot 的 appinsights instrumentation 技能为 Web 应用接入 Azure Application人工智能MCP 服务MCP ClientsTwenty 应用怎么管理本地 Docker 服务器、版本固定与元数据同步恢复Twenty 应用怎么管理本地 Docker 服务器、版本固定与元数据同步恢复 开发 Twenty 应用时本地循环依赖三件事一个可控的本地 Docker人工智能MCP 服务MCP ClientsPouchDB 变更订阅指南实时监听数据库变化PouchDB 变更订阅指南实时监听数据库变化 什么是变更订阅Changes Feed PouchDB 作为一款优秀的客户端数据库提供了强大的变更订阅功数据库数据同步上一篇glTFast解决方案Unity中高效加载与导出3D模型的深度实践指南下一篇突破浏览器限制WebLLM滑动窗口实现长文本处理的优化策略创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Python SDK 客户端订阅指南：用 `client.listen(...)` 实时监听 MCP 服务器目录变更
人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载服务器目录并非一成不变工具会在运行时出现资源 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 servers 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.mdInside 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_changedTrue, resource_subscriptions[BOARD]) as sub: async for event in sub: match event: case ResourceUpdated(uriuri): 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 中定义并导出事件类型含义ToolsListChanged工具列表发生变化PromptsListChanged提示词prompt列表发生变化ResourcesListChanged资源列表发生变化ResourceUpdated(uri...)某个资源 URI 背后的内容发生变化uri字段指明是哪个示例代码使用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.pyimport 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.pyimport 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.pyimport 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)并取消驱动任务、注销路由。一个细节值得注意运行期与应用同寿命的观察者永远不会自行返回所以在应用关闭时必须显式取消它或取消其所属任务组的范围否则程序无法干净退出。流的结束两种收尾与重连策略一条流只以两种方式结束而它们都属于普通控制流的范畴结束方式表现服务器优雅关闭graceful closeasync for循环自然结束StopAsyncIteration突然断开abrupt drop抛出SubscriptionLost从源码看src/mcp/client/subscriptions.py 的Subscription.__anext__正是这样区分的ListenRoute.next_event返回字符串结局标记lost时抛出SubscriptionLost并链上原始错误graceful/local时抛出StopAsyncIteration结束循环。同时驱动任务drive()里有一条重要注释A result, whatever its body, is the specs 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.mdInside 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【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载相关推荐使用 awesome-copilot 的 appinsights-instrumentation 技能为 Web 应用接入 Azure Application Insights 遥测使用 awesome copilot 的 appinsights instrumentation 技能为 Web 应用接入 Azure Application人工智能MCP 服务MCP ClientsTwenty 应用怎么管理本地 Docker 服务器、版本固定与元数据同步恢复Twenty 应用怎么管理本地 Docker 服务器、版本固定与元数据同步恢复 开发 Twenty 应用时本地循环依赖三件事一个可控的本地 Docker人工智能MCP 服务MCP ClientsPouchDB 变更订阅指南实时监听数据库变化PouchDB 变更订阅指南实时监听数据库变化 什么是变更订阅Changes Feed PouchDB 作为一款优秀的客户端数据库提供了强大的变更订阅功数据库数据同步上一篇glTFast解决方案Unity中高效加载与导出3D模型的深度实践指南下一篇突破浏览器限制WebLLM滑动窗口实现长文本处理的优化策略创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
