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点击查看免费下载服务器端的目录catalog并不是固定不变的工具可能在运行时才出现资源 URI 背后的内容也会随时变化。在 MCPModel Context Protocol的 Python 官方 SDK本仓库 python-sdk中客户端通过client.listen(...)感知这些变化——这本质上是一次subscriptions/listen请求而它的响应本身就是一条流请求发出后流保持打开持续运送客户端所关心的变更通知。本文聚焦客户端一侧的完整链路如何打开订阅流、如何在主流程旁边异步监控它、以及如何处理流的两种结束方式。变更的发布、过滤与该方法在服务端的提供属于另一侧的话题见 服务端订阅指南本文示例所通信的“冲刺看板sprint-board”服务器也由该文构建。打开并监控订阅流client.listen(...)返回一个异步上下文管理器async context manager。进入async with块即发出订阅请求请求的关键字参数就是订阅过滤器subscription filter随后 SDK 会等待服务器的确认acknowledgment——因此代码块真正开始执行时流已经处于活跃状态。以下示例来自 docs_src/subscriptions/tutorial003.pyfrom 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)迭代async for event in sub会产出四种类型化事件定义于 src/mcp/client/subscriptions.py 并重新导出事件类型含义ToolsListChanged工具列表发生变化PromptsListChanged提示prompt列表发生变化ResourcesListChanged资源列表发生变化ResourceUpdated(uri...)指定 URI 的资源内容被更新事件是“重新获取”的信号而非数据载荷事件只告诉你**“什么”变了**从不告诉你**“怎么”变的**。这正是follow_board在收到事件后要调用read_resource与list_tools的原因事件只是重新拉取refetch的提示cue最终以重新读取得到的最新状态为准。不要臆断是哪个资源发生了变动而应读取event.uri过滤器可以同时指定多个 URI且服务器可能报告其中某个 URI 的子资源发生了变更。测试套件 tests/client/test_subscriptions.py 中的test_listen_delivers_all_four_typed_event_kinds即为四种事件的逐项验证。重复事件的合并coalescing等待消费的重复事件会被合并为一条例如连续收到三次相同的ResourceUpdated消费端只会看到一次。由于合并后再重新获取仍能得到最新状态这不会造成信息丢失。只有完全相同的事件才会合并指向不同 URI 的两个ResourceUpdated就是两条独立事件。订阅句柄的两个属性listen返回的Subscription句柄即async with块中的sub还有两个值得关注的属性sub.honored服务器实际确认acknowledge的过滤器类型为SubscriptionFilter包含你传入的那些字段可通过属性直接读取例如sub.honored.prompts_list_changed。本 SDK 的MCPServer会接受你请求的每一种事件类型因此它会原样回显你的请求而支持类型较少的服务器则只确认更少的字段——且即使某种类型被确认接受也不代表它一定会触发。服务器还可以选择整体拒绝该请求而不是确认参见服务端文档中“谁可以观看”一节此时会以请求错误的形式暴露给客户端。sub.subscription_id该 listen 请求的 ID它会被烙印stamp在这条流的所有帧frame上。可以同时打开多条订阅每条流凭借自己的 ID 被多路分解demultiplex。从源码看这个 ID 由进程级计数器生成、形如listen-N见 src/mcp/client/subscriptions.py 中的_listen_ids测试test_listen_surfaces_the_honored_filter_and_subscription_id验证了其字符串前缀与确认过滤器的回显。非阻塞监控让 watcher 与主流程并行follow_board会一直运行到服务器关闭流为止——而服务器可能永远不会关闭因此若单独运行它会独占整个程序。真实客户端想要的是在主流程“旁边”运行的监控者Agent 一边调用工具watcher 一边让缓存或 UI 保持最新。做法是先打开订阅进入块、等到确认再启动 watcher 任务然后继续做正事。三种异步后端各有写法 asynciopython titleapp.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()) triopython titleapp.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) anyiopython titleapp.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从第一个示例中导入BOARD和read_board本仓库将其保存为tutorial003.py即from .tutorial003 import ...。如果你把渲染后的文件并排保存为client.py与app.py请改写成from client import BOARD, read_board下文watch.py示例对read_board的导入同理。这三个文件的仓库路径分别为 docs_src/subscriptions/tutorial004_asyncio.py、docs_src/subscriptions/tutorial004_trio.py 与 docs_src/subscriptions/tutorial004_anyio.py。顺序是关键没有任何重放replay机制在你建立流之前已发布的事件会被错过。而进入client.listen(...)会一直等到服务器确认因此从确认那一刻起的所有变更都会到达你的 watcher——你在块内拍摄的快照snapshot不可能漏掉一次变更。打开的流不阻塞其他请求请求可以在已打开的流旁边自由执行无论来自 watcher 任务还是其他任务都共享同一个客户端连接。由于重复的未消费事件会被合并当主流程忙碌时可能原本需要三次重新获取refetch才能跟上的状态一次就完成了不同的事件不会合并——如果过滤器指定了多个 URI那么每个 URI 各有一条待处理事件在队列中。停止监控停止监控的方式就是退出代码块没有unsubscribe之类的显式调用。取消拥有该代码块的任务即可SDK 会按传输层期望的方式取消 listen 请求——对于 Streamable HTTP即关闭该请求的流。需要说明的是为整个应用生命周期服务的 watcher 永远不会自行返回因此在应用关闭shutdown时请显式取消它或取消其所属任务组task group的作用域。流的两种结束方式流只有两种结束方式且两者都属于正常的控制流服务器正常关闭async for循环自然结束突然断开抛出SubscriptionLost。这个区别仅用于诊断并不改变接下来的动作流已经没了、什么都不会重放仍有关注需求的 watcher 应当重新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)正常关闭也可能是“负载卸载”服务器可能出于自身原因正常关闭流——例如卸载shed积压过大的订阅者。因此“干净地结束”不是“停止监控”的信号在重新listen之前务必先退避back off一段时间上面的示例用anyio.sleep(1)实现这一点。SubscriptionLost 的本地成因1024 条积压上限SubscriptionLost还有一个客户端本地成因客户端最多保留1024 条未消费事件对应源码中的_MAX_PENDING_EVENTS 1024见 src/mcp/client/subscriptions.py。当消费方落后到超过这个上限时SDK 会选择让订阅失联而不是让内存无限膨胀源码注释明确说明由于规范允许子资源 URI去重后的不同ResourceUpdated事件在理论上是无界的因此设置了该积压护栏。所以请保持async for循环体简短把耗时操作放到别处执行。listen() 进入时还可能抛出的异常keep_following只捕获SubscriptionLost。进入listen()时还可能抛出MCPError连接失败或服务器不提供该方法TimeoutError在读取超时时间内没有收到确认ListenNotSupportedError协商的协议版本早于 2026-07-28该功能要求 2026-07-28 版本连接。请自行决定你的 watcher 应对其中哪些进行重试——最后一种永远不会自愈。从 src/mcp/client/client.py 中Client.listen的签名可以看到它还带有一个隐藏联动当启用响应缓存时on_event会在每个事件返回给消费者之前先完成缓存驱逐cache eviction保证消费者重新获取时读到的是新鲜数据这正是“事件也能保持客户端缓存诚实”的底层实现。小结进入async with client.listen(...)进入时会等待确认因此之后发布的内容一个都不会漏。用async for event in sub迭代事件是重新获取的提示永远不是数据载荷。先打开订阅再把 watcher 作为任务运行工具调用就能在它旁边继续流动。干净结束则循环停止突然断开则抛出SubscriptionLost。无论哪种重新 listen、重新获取但先退避。退出代码块就是退订。事件的发布、过滤器的收窄、跨进程的扩展属于服务端的话题详见 服务端订阅指南同样的变更事件还能让客户端缓存保持精确下一篇可继续阅读 客户端缓存。赞分享人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载相关推荐MCP Python SDK 客户端订阅实战用 client.listen() 监听服务器变更流MCP Python SDK 客户端订阅实战用 client.listen 监听服务器变更流 导读 服务器目录工具、提示词、资源并非一成不变工具会在运行人工智能MCP 服务MCP ClientsPython SDK 客户端订阅实战用 client.listen() 感知 MCP 服务器的目录变化Python SDK 客户端订阅实战用 client.listen 感知 MCP 服务器的目录变化 本篇指南讲解官方 Python SDK 的客户端订阅能力人工智能MCP 服务MCP ClientsPython SDK 客户端订阅指南用 client.listen(...) 实时监听 MCP 服务器目录变更Python SDK 客户端订阅指南用 client.listen ... 实时监听 MCP 服务器目录变更 服务器目录并非一成不变工具会在运行时出现资源人工智能MCP 服务MCP Clients上一篇Campus-imaotai SPI机制服务发现与插件化架构下一篇如何利用智能体AI技术打造下一代零售客户行为分析与推荐系统创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考