MCP Python SDK 进度通知Progress Notifications机制让长任务工具实时反馈的实现原理与实战指南【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk本文围绕mcpPython SDK 中Context.report_progress与Client.call_tool(progress_callback...)这对工具进度上报与监听 API 展开讲解服务端如何在工具函数中通过类型注入的Context对象发送notifications/progress通知、客户端如何按调用粒度挂载异步回调来接收进度、以及在真实网络传输下回调时序与进程内测试连接的差异。读完本文后你将能独立实现一个带进度反馈的 MCP 工具理解协议层的ProgressNotificationParams字段语义并掌握无条件上报、按需监听的工程范式。为什么需要进度通知一个运行三十秒却在此期间毫无输出的工具在用户眼中往往与卡死无异。MCP 协议为此定义了进度通知progress notifications工具在执行过程中向调用方报告当前完成度由客户端自行决定如何呈现——进度条、旋转指示器或一行日志输出。服务端负责说什么客户端负责怎么显示两者完全解耦。这一机制的协议定义可在ProgressNotificationParams类型中确认# src/mcp-types/mcp_types/_types.py class ProgressNotificationParams(NotificationParams): Parameters for progress notifications. progress_token: ProgressToken # 用于关联通知与触发它的请求 progress: float # 当前进度须单调递增 total: float | None None # 总量若已知 message: str | None None # 人类可读的当前步骤描述其中progress_token是客户端在发起请求时附带的令牌服务端用它在notifications/progress消息中回指原始请求从而让客户端将进度通知与特定调用关联起来。服务端通过 Context 上报进度获取 Context在 MCP Python SDK 中工具函数只要在参数列表中声明Context类型SDK 就会自动将其注入——无需手动获取模型LLM也永远看不到这个参数from mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Bookshop) mcp.tool() async def import_catalog(urls: list[str], ctx: Context) - str: Import book records from a list of catalog URLs. for done, url in enumerate(urls, start1): await ctx.report_progress(done, totallen(urls), messagefImported {url}) return fImported {len(urls)} records.Context由类型注解自动注入import_catalog的输入 schema 中只包含一个属性urls对模型完全透明。report_progress 的三个参数ctx.report_progress接受三个参数语义由工具开发者自行定义参数类型是否必填说明progressfloat是当前进度值。协议要求每次上报必须单调递增不得重复或回退totalfloat \| None否总量。若已知分母则传入客户端可据此计算百分比messagestr \| None否关于当前步骤的人类可读描述如正在处理第 3 条记录progress的单位由工具自行决定——字节、行数、页数、迭代次数均可关键是选择用户能直觉理解的度量。上报的底层链路从源码结构看Context.report_progress的实现是一个简单的代理调用# src/mcp/server/mcpserver/context.py (L113-121) async def report_progress(self, progress: float, total: float | None None, message: str | None None) - None: Report progress for the current operation. Args: progress: Current progress value (e.g., 24) total: Optional total value (e.g., 100) message: Optional message (e.g., Starting render...) await self.request_context.session.report_progress(progress, total, message)它委托给会话层的report_progress# src/mcp/server/session.py (L426-434) async def report_progress(self, progress: float, total: float | None None, message: str | None None) - None: Report progress for the inbound request this session is scoped to. A no-op when the caller did not request progress. Dispatcher-agnostic: on JSON-RPC the held DispatchContext emits notifications/progress against the callers token; on the in-process direct dispatcher it invokes the callers callback directly. await self._request_outbound.progress(progress, total, message)这里的注释明确了两种行为路径JSON-RPC 传输真实网络连接发送一条独立的notifications/progress消息通过调用方提供的 progress token 关联到原始请求。进程内直连in-process test dispatcher直接同步调用调用方的回调函数。两条路径都遵循同一原则若调用方未请求进度即未提供 progress token则report_progress是空操作no-op。这意味着服务端工具可以无条件地调用report_progress无需检查是否有人在监听。客户端按调用挂载 progress_callback基本用法客户端通过Client.call_tool的progress_callback参数按调用粒度per-call启用进度监听import anyio from mcp import Client async def show(progress: float, total: float | None, message: str | None) - None: print(f{message} ({progress}/{total})) async def main() - None: async with Client(http://localhost:8000/mcp) as client: result await client.call_tool( import_catalog, {urls: [https://example.com/a.json, https://example.com/b.json]}, progress_callbackshow, ) print(result.structured_content) anyio.run(main)回调函数必须是一个async函数签名与服务器上报的三元组精确对应(progress: float, total: float | None, message: str | None) - None。progress_callback 归属调用而非客户端progress_callback是call_tool方法的参数不是Client构造器的参数# src/mcp/client/client.py (L751-756) async def call_tool( self, name: str, arguments: dict[str, Any] | None None, read_timeout_seconds: float | None None, progress_callback: ProgressFnT | None None, ...设计意图很明确同一个客户端连接可能在不同调用中需要不同的进度处理策略——一次调用驱动下载进度条下一次调用写入日志行。将回调绑定到调用而非连接提供了最大的灵活性。测试用例直接验证了这一 API 契约# tests/docs_src/test_progress.py def test_progress_callback_is_per_call_not_per_client() - None: The !!! warning: call_tool takes progress_callback; the Client constructor does not. assert progress_callback in inspect.signature(Client.call_tool).parameters assert progress_callback not in inspect.signature(Client.__init__).parameters回调在传输层的挂载路径Client.call_tool将progress_callback向下传递至会话层的send_request# src/mcp/client/session.py (L579-580) if progress_callback is not None: opts[on_progress] progress_callback随后opts字典被传给分发器。在 JSON-RPC 分发器中on_progress被注册为该请求的待处理条目的一部分# src/mcp/shared/jsonrpc_dispatcher.py (L358-367) on_progress opts.get(on_progress) if on_progress is not None: # The request id doubles as the progress token, so _pending[token] finds on_progress directly. ... pending _Pending(sendsend, receivereceive, on_progresson_progress)当notifications/progress消息到达时分发器通过消息中的progress_token即请求 ID查找对应的on_progress回调并执行。真实传输下的时序特性这是一个容易被忽视但生产环境中必须理解的细节在真实网络传输下每条进度通知都作为独立消息随响应旁路beside the response送达慢速回调可能在call_tool已返回之后仍在运行。# src/mcp/shared/jsonrpc_dispatcher.py (L635-640) and pending.on_progress is not None ... _shielded_progress(pending.on_progress),进度回调通过_shielded_progress包装执行与主响应接收路径并行。这意味着进程内直连in-process test connection回调内联同步执行保证每条进度上报都先于结果返回。真实传输HTTP / Stdio回调在独立任务中执行call_tool返回不保证所有回调已完成。测试用例精确复现了这一竞态场景# tests/docs_src/test_progress.py (L45-71) async def test_over_a_wire_dispatcher_callbacks_race_the_result() - None: On a wire dispatcher each progress notification starts its own task, so call_tool can return while a slow callback is still running. release anyio.Event() done anyio.Event() finished: list[float] [] async def gated(progress: float, total: float | None, message: str | None) - None: await release.wait() # 阻塞直到 call_tool 返回后才释放 finished.append(progress) if len(finished) 2: done.set() async with Client(tutorial001.mcp, modelegacy) as client: with anyio.fail_after(5): result await client.call_tool(import_catalog, {urls: URLS}, progress_callbackgated) assert finished [] # call_tool 返回时回调尚未完成 release.set() with anyio.fail_after(5): await done.wait() assert sorted(finished) [1, 2]如果你的进度回调涉及 UI 更新或数据库写入需确保这些操作是幂等且可并发的不依赖于回调一定在结果返回前完成的假设。完整运行示例终端 1—— 启动 HTTP 服务器uv run mcp run server.py --transport streamable-http终端 2—— 运行客户端python client.py预期输出Imported https://example.com/a.json (1.0/2.0) Imported https://example.com/b.json (2.0/2.0) {result: Imported 2 records.}服务端每次await ctx.report_progress(...)对应客户端一次show调用按上报顺序依次执行。进度不在结果result中打包——它在工具仍在执行时就已流式送达。验证移除回调后的行为删除progress_callbackshow后重新运行输出仅为{result: Imported 2 records.}无错误、无警告、结果完全一致。这正是无条件上报设计的价值服务端工具永远调用report_progress无需判断是否有监听者。总量未知时省略 totaltotal适用于已知分母的场景。但在以下情况中总量不可预知遍历无限数据流draining a feed遍历数据库游标walking a cursor下载无长度头的资源此时直接省略total参数from collections.abc import AsyncIterator from mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Bookshop) async def fetch_records(feed_url: str) - AsyncIterator[str]: for title in (Dune, Neuromancer, Hyperion): yield f{feed_url}#{title} mcp.tool() async def import_feed(feed_url: str, ctx: Context) - str: Import every record a catalog feed yields. imported 0 async for record in fetch_records(feed_url): imported 1 await ctx.report_progress(imported, messagefImported {record}) return fImported {imported} records.回调将收到totalNone。客户端仍可显示活动状态已导入 3 条……但无法呈现百分比进度条。不要为了获得更好看的进度条而虚构一个总量值。测试用例验证了totalNone的传递# tests/docs_src/test_progress.py (L88-102) async def test_omitting_total_reaches_the_callback_as_none() - None: tutorial002: a report without total arrives as totalNone: activity, not a percentage. updates: list[tuple[float, float | None, str | None]] [] async def show(progress: float, total: float | None, message: str | None) - None: updates.append((progress, total, message)) async with Client(tutorial002.mcp) as client: result await client.call_tool(import_feed, {feed_url: https://example.com/feed}, progress_callbackshow) assert updates [ (1, None, Imported https://example.com/feed#Dune), (2, None, Imported https://example.com/feed#Neuromancer), (3, None, Imported https://example.com/feed#Hyperion), ] assert result.structured_content {result: Imported 3 records.}设计要点总结要点说明服务端上报await ctx.report_progress(progress, totalNone, messageNone)任何接受Context的工具均可调用客户端监听call_tool(..., progress_callbackfn)按调用粒度不在Client构造器上回调签名async (progress: float, total: float \| None, message: str \| None) - None触发时机工具执行过程中触发非结果返回后无回调行为report_progress静默变为 no-op可无条件上报总量未知省略total回调收到None展示活动而非百分比进度单调性progress必须单调递增不得重复或回退进度通知是工具向用户展示执行状态的通道。工具向服务端运维者输出的日志行是另一条独立通道参见 日志Logging 文档。Context对象的完整能力包括进度上报、日志记录、资源读取、用户交互等参见 Context 对象 文档。源码与测试参考文件内容src/mcp-types/mcp_types/_types.pyProgressNotificationParams与ProgressNotification协议类型定义src/mcp/server/mcpserver/context.py服务端Context.report_progress代理方法src/mcp/server/session.py会话层report_progress区分 JSON-RPC 与进程内直连路径src/mcp/client/client.pyClient.call_tool的progress_callback参数声明src/mcp/client/session.pysend_request将回调挂载至opts[on_progress]src/mcp/shared/jsonrpc_dispatcher.pyJSON-RPC 分发器中进度通知的接收与回调调度docs_src/progress/tutorial001.py完整服务端示例已知总量docs_src/progress/tutorial002.py完整服务端示例未知总量tests/docs_src/test_progress.py页面所有技术声明的自动化验证【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考