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点击查看免费下载一个需要三十秒才能完成的工具如果在这三十秒内一言不发看起来就像卡死了。进度通知Progress Notifications正是为此而生的机制工具实时上报自己执行到哪一步客户端决定如何把这些信息画出来——进度条、转圈动画或是一行日志。本指南基于官方 Python SDK 的 handlers/progress 文档结合仓库源码与测试用例完整讲解在服务端工具中上报进度、在客户端按调用接收进度的全流程读完你将掌握ctx.report_progress()与progress_callback的完整用法、参数语义与底层消息流转原理。为什么工具需要进度通知设想一个导入书目的工具它要逐个抓取远程目录 URL每个 URL 的抓取与解析耗时数秒。若工具在整个执行期间不发出任何中间信号用户看到的就是一个毫无反应的黑盒——尽管它其实一直在干活。进度通知把这个黑盒打开服务端在工具执行期间持续报告我在哪里、总共多少、这一步在做什么客户端收到后自主决定渲染形式进度条、spinner、日志行。进度信息不属于工具结果它是一条独立的通知流在工具仍在工作时就实时送达。服务端通过Context上报进度在官方 Python SDK 中任何需要上报进度的工具函数只需增加一个带Context类型注解的参数然后在函数体内调用await ctx.report_progress(...)即可。以文档示例 tutorial001.py 为骨架完整代码如下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.report_progress接受三个参数含义完全由你决定参数类型是否可选语义要求progressfloat必填当前进度值。MCP 规范要求每次上报都必须递增不能重复同一个值也不能回退totalfloat \| None可选总进度值分母不知道总数时直接省略messagestr \| None可选关于当前这一步的一行人类可读描述从源码看Context.report_progress 的实现非常薄它直接把参数转发给底层会话async def report_progress(self, progress: float, total: float | None None, message: str | None None) - None: await self.request_context.session.report_progress(progress, total, message)ctx是类型注解注入的模型永远看不到它ctx之所以可用仅仅是因为它的类型注解——SDK 在调用工具时检测到Context类型的参数并注入上下文对象。它不会出现在工具的输入 schema 中上例中import_catalog的输入 schema 只有唯一属性urls模型LLM在调用工具时只看到urls看不到ctx。这一点在 test_progress.py 的测试用例 中被严格验证tool.input_schema[properties] {urls: ...}且required [urls]。Context是一个功能丰富的对象进度只是它提供的众多能力之一。关于它承载的请求上下文、资源读取、elicitation 等能力详见 Context 专题文档。客户端按调用订阅进度回调客户端按调用per call选择是否接收进度方法是在call_tool时传入progress_callbackimport 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。这个签名的类型定义在 shared/dispatcher.py 的ProgressFnTclass 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: ...在 Client.call_tool 中progress_callback是与name、arguments平级的参数SDK 随后在 client/session.py 中把它映射为底层调度器的on_progress选项if progress_callback is not None: opts[on_progress] progress_callback动手试试先在一个终端里用 HTTP 方式启动服务端uv run mcp run server.py --transport streamable-http然后在第二个终端运行客户端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调用且保持顺序。进度并不打包进工具结果里——它在工具仍在运行时就已经作为独立通知流送达了。底层原理notifications/progress如何从服务端流到客户端从源码结构可以梳理出这条消息的完整链路服务端发起ctx.report_progress(...)经由Context.report_progresssrc/mcp/server/mcpserver/context.py委托给会话层由ServerSession发出 MCP 协议的notifications/progress通知其中携带progressToken与当前请求 id 关联、progress、可选的total与message。客户端接收底层调度器在 jsonrpc_dispatcher.py 中拦截notifications/progress用progressToken在挂起的请求表_pending中反查出对应的on_progress回调然后调用它。请求 id 同时充当 progress token因此token → 回调的映射是直接的。回调的调度与隔离每条进度通知会被独立调度_spawn并通过_shielded_progress包装——从源码看这个包装的作用是防止回调自身抛出的异常影响通知处理主循环。时序上的关键事实在真实网络传输上每条通知都是独立送达的与结果响应并行。因此一个执行较慢的回调可能在call_tool已经返回之后仍在运行。只有进程内测试连接直接把 server 对象传给Client会内联执行回调并保证每条上报都在结果返回前送达。仓库测试 test_progress.py 精确复现了这一行为在modelegacy走 wire dispatcher下回调被一个事件闸门阻塞call_tool返回时回调尚未执行完随后才陆续完成——这正是文档提醒你在真实传输上要注意的时序情形。两个关键设计约束progress_callback属于调用不属于Clientprogress_callback是call_tool的参数Client构造器没有这个参数。原因很直接不同调用想要不同的回调——这一次驱动下载进度条下一次可能只想打印一行日志。若把它挂在Client上所有调用就被迫共享同一个回调失去了按调用定制的能力。相关测试 test_progress.py 用inspect.signature断言了这一点progress_callback存在于Client.call_tool的签名中而不存在于Client.__init__中。没有回调时report_progress是一个 no-op把progress_callbackshow删掉再运行一次{result: Imported 2 records.}没有报错没有警告结果完全一样。这意味着report_progress在调用方没有要求进度时是 no-op——所以你在服务端可以无条件上报永远不必担心有没有人在听。这个行为在 test_progress.py 中同样被验证。当你不知道总数时省略totaltotal是为你知道分母的场景准备的。但很多时候你并不知道总数你正在排空一个 feed、遍历一个游标、下载一个没有长度头的流。此时直接省略total即可见文档示例 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.注意这里report_progress(imported, message...)只传了progress和message。客户端回调收到的将是totalNone对应测试 test_progress.py 中断言的(1, None, ...)、(2, None, ...)、(3, None, ...)序列。客户端此时仍然可以展示活动状态目前已导入 3 条……但无法给出百分比。不要为了一个更好看的进度条而编造total。百分比建立在真实分母之上虚构的总数只会误导用户。最佳实践小结progress不一定要计数某种特定单位。字节、行数、页数都行——选择用户能一眼认出的单位并且只承诺你一定能兑现的total。从任何接收Context的工具函数中用await ctx.report_progress(progress, totalNone, messageNone)上报进度。客户端在call_tool时传入progress_callback按调用生效永远不要指望在Client构造器里配置它。回调签名固定为async (progress, total, message) - None在工具仍在执行时触发注意在真实传输上回调可能与结果赛跑。调用没有回调时report_progress什么都不做——放心无条件上报。不知道总数就省略total回调会收到None。进度与日志是两个不同的通道进度是工具运行时展示给**用户使用者的信号而工具写给你运维服务端的人**看的日志行是另一条独立的通道。需要服务端运行时日志ctx.info(...)、ctx.debug(...)等的用法时请参考 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 本文基于当前仓库 doc人工智能MCP 服务MCP ClientsMCP Python SDK 进度通知实战从工具端 report_progress 到客户端 progress_callbackMCP Python SDK 进度通知实战从工具端 report_progress 到客户端 progress_callback 导读 本文围绕 Model人工智能MCP 服务MCP ClientsMCP Python SDK 进度通知实战从工具端 report_progress 到客户端 progress_callback 完整指南MCP Python SDK 进度通知实战从工具端 report_progress 到客户端 progress_callback 完整指南 本文以官方 Pyt人工智能MCP 服务MCP Clients上一篇fio内存管理机制详解高效处理大文件I/O的内存优化策略下一篇Argo CD CLI 详解argocd account bcrypt 密码哈希生成命令的用法与源码解析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考