MCP Python SDK 进度通知实战：从工具端 `report_progress` 到客户端 `progress_callback` 完整指南

人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载本文以官方 Python SDK 的进度progress通知机制为核心讲解如何让长时间运行的 MCP 工具主动上报进度以及客户端如何在每次工具调用中按需订阅这些进度更新。读完本文你将掌握服务端Context.report_progress与客户端call_tool(progress_callback...)的完整用法、两者的底层调用链与时序约束并学会在总量未知时如何优雅地处理进度展示。为什么需要进度通知一个需要 30 秒才能完成的工具如果在整整 30 秒内不发一言看起来就像是卡死了。进度通知progress notifications正是为解决这一问题而设计的工具端主动报告现在进行到哪一步客户端根据收到的信息自行决定如何绘制——可以是进度条、转圈动画spinner也可以只是一行日志。进度是工具还在运行时这一事实的对外表达它独立于工具调用的最终结果因此在 MCP 协议中是一条独立的notifications/progress通知而不是被塞进tools/call的响应里。服务端从工具中上报进度接收Context参数并调用report_progress在 SDK 中任何需要上报进度的工具函数只需额外声明一个ctx: Context参数然后在函数体内调用await ctx.report_progress(...)即可。完整示例见 docs_src/progress/tutorial001.pyfrom 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.report_progress接受三个参数每个参数的含义完全由你自行定义参数含义约束progress已经完成了多少必须递增。MCP 规范要求每次上报时该值只能增加不能重复同一个值也不能回退total总量是多少如果你知道可选默认Nonemessage描述当前这一步的一行人类可读文本可选默认Nonectx之所以能被注入纯粹是因为它的类型提示type hintSDK 会在调用时自动传入而模型LLM完全看不到这个参数。证据是import_catalog的输入 schema 中只有urls这一个属性——这一点在 tests/docs_src/test_progress.py 的test_context_parameter_is_invisible_to_the_model用例中被直接断言验证。关于Context对象的完整能力参见 Context 文档进度上报只是它提供的功能之一。底层调用链与无人监听即空操作原理从源码结构看ctx.report_progress的调用链是Context.report_progresssrc/mcp/server/mcpserver/context.py→session.report_progresssrc/mcp/server/session.py→ 请求对应的 outbound 通道。值得注意的是 src/mcp/server/session.py 中的 docstring 明确指出当调用方没有请求进度时report_progress是一个空操作no-op且该行为与分发器dispatcher无关——在 JSON-RPC 传输上它通过持有的DispatchContext以调用方的 token 发出notifications/progress在进程内直连分发器上它直接调用调用方的回调。这意味着服务端工具函数可以无条件地上报进度完全不必关心是否有人在听。是否有监听者、监听者是谁都由客户端那一侧决定。客户端按调用接收进度每次调用传入progress_callback客户端以每次调用为单位选择是否接收进度方式是在call_tool上传递progress_callback参数。完整示例见 docs_src/progress/tutorial001_client.py对应文档中的 client.py 代码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、total、message。其签名在 SDK 中以协议类ProgressFnT定义src/mcp/shared/dispatcher.pyclass ProgressFnT(Protocol): Callback invoked when a progress notification arrives for a pending request. async def __call__(self, progress: float, total: float | None, message: str | None) - None: ...从客户端源码看progress_callback的传递链为Client.call_toolsrc/mcp/client/client.py→session.call_tool→send_request把回调写入opts[on_progress]src/mcp/client/session.py→ 分发器根据该选项订阅进度通知。时序约束通知与响应各自独立送达有一个关键的时序事实需要牢记每条进度通知都是单独送达的与最终的响应并不同步。因此慢回调在call_tool已经返回之后仍可能还在运行只有进程内测试连接in-process test connection才会内联inline运行回调从而保证所有上报先于结果到达。这一行为在源码中有清晰体现。在 JSON-RPC 分发器中notifications/progress被单独拦截并通过self._spawn(...)为每条通知启动独立任务来调用回调src/mcp/shared/jsonrpc_dispatcher.py而在进程内直连分发器中回调是直接await内联执行的src/mcp/shared/direct_dispatcher.py。仓库中的测试 test_over_a_wire_dispatcher_callbacks_race_the_result 专门验证了这一行为在 wire 分发器legacy 模式上回调被事件门控call_tool返回时finished列表仍为空直到事件被释放后回调才陆续完成——这正是文档警告不要排除慢回调晚于结果到达的原因。直接上手试一试将server.py以 HTTP 方式启动然后在第二个终端运行客户端uv run mcp run server.py --transport streamable-httppython 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调用且顺序完全一致。进度并不捆绑在结果里而是在工具仍在工作时以流式方式不断送达。progress_callback属于调用而非Client必须特别强调progress_callback属于某次调用Client构造函数中不存在对应参数。因为不同调用往往想要不同的回调——某次调用可能驱动一个下载进度条下一次调用可能只想在日志里留一行。仓库测试 test_progress_callback_is_per_call_not_per_client 用inspect.signature同时断言了两件事progress_callback存在于Client.call_tool的参数列表中而不存在于Client.__init__的参数列表中。没有回调时report_progress是空操作现在把progress_callbackshow删掉再运行一次{result: Imported 2 records.}没有报错、没有警告结果完全相同。这是因为服务端在调用方未请求进度时会把report_progress当作空操作处理。因此正确的姿势是无条件上报不用操心有没有人在听。对应测试为 test_without_a_callback_report_progress_is_a_no_op。不知道总量时省略totaltotal是你知道分母时才使用的值。但在很多场景下你根本不知道总量清空一个 feed、沿着游标翻页、下载一个没有长度头的资源——这时就把total省略掉。完整示例见 docs_src/progress/tutorial002.pyfrom 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.此时客户端回调收到的total是None。客户端仍然可以展示活动状态例如目前已经导入了 3 条……但无法展示百分比。仓库测试 test_omitting_total_reaches_the_callback_as_none 验证了(1, None, ...)、(2, None, ...)、(3, None, ...)这样的回调序列。两个实用建议progress不一定要数某个特定的东西。字节、行数、页数都行——选择用户能一眼认出的单位只承诺你能兑现的total。不要为了画一个更好看的进度条而编造总量。小结任何接收Context的工具都可以调用await ctx.report_progress(progress, totalNone, messageNone)客户端在call_tool上传递progress_callback参数——按调用指定永远不放在Client上回调形如async (progress, total, message) - None在工具仍在运行时触发调用上没有回调时report_progress什么都不做所以请无条件上报不知道total时就省略它回调会收到None。进度与日志是两个不同的通道最后要区分两个容易混淆的概念进度是正在运行的工具展示给用户看的东西而工具为**运营方你**留下的日志行属于另一条独立的通道相关内容请参阅 日志Logging文档。两者虽然都源自一次工具调用但面向的对象、传递的通道和消费方式完全不同在设计服务端行为时应当分开考虑。赞分享人工智能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 进度通知实战从工具端 report_progress 到客户端 progress_callbackMCP Python SDK 进度通知实战从工具端 report_progress 到客户端 progress_callback 导读 本文围绕 Model人工智能MCP 服务MCP ClientsMCP Python SDK 进度通知完整指南从服务端 report_progress 到客户端 progress_callbackMCP Python SDK 进度通知完整指南从服务端 report_progress 到客户端 progress_callback 本文基于当前仓库 doc人工智能MCP 服务MCP ClientsPython MCP SDK 进度通知Progress Notifications实战指南从 report_progress 到 progress_callback 的完整链路Python MCP SDK 进度通知Progress Notifications实战指南从 report_progress 到 progress_cal人工智能MCP 服务MCP Clients上一篇Ant Design Vue Watermark 水印组件完全指南API 配置、源码原理与防篡改机制下一篇F´ Hub 模式Hub Pattern深度解析用 GenericHub 实现跨部署、跨边界组件通信创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考