进度

机器翻译

本页由英文文档自动翻译而来，以英文页面为准。如果有读起来不对的地方，翻译页面说明了如何反馈。

一个要跑三十秒的工具，如果这三十秒里一声不吭，看起来就像坏了。

进度通知解决的就是这个问题。工具报告自己做到哪了；客户端决定拿它画什么：进度条、旋转指示器，还是一行日志。

从工具里报告

接收一个 Context 参数，然后调用 report_progress：

server.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, start=1):
        await ctx.report_progress(done, total=len(urls), message=f"Imported {url}")
    return f"Imported {len(urls)} records."

三个参数，含义由你决定：

progress：做到哪了。规范要求它每次报告都递增；不要重复同一个值，也不要倒退。
total：总共有多少，如果你知道的话。可选。
message：描述这一步的一行人类可读文字。可选。

ctx 是因为类型注解被注入的，模型永远看不到它：import_catalog 的输入模式只有一个属性 urls。Context 页面专门讲这个对象；进度只是它提供的功能之一。

从客户端监听

客户端按调用选择接收，方法是给 call_tool 传 progress_callback=：

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_callback=show,
        )
    print(result.structured_content)


anyio.run(main)

回调是一个 async 函数，接收的正是服务器报告的内容：progress、total、message。

Info

无论交给 Client 的是什么——像这里这样的 URL、StdioServerParameters，还是测试里的服务器对象——progress_callback 都是同一个参数。不过，走真实传输方式时要留意时序。每条通知都是单独送达的，和响应各走各的，所以 call_tool 已经返回之后，一个慢的回调可能还在运行。只有进程内的测试连接会以内联方式运行回调，并保证每条报告都先送达。

试一试

通过 HTTP 启动 server.py，然后在另一个终端运行客户端：

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 的一次调用，顺序不变。进度不会打包进结果里，而是在工具还在干活的时候就流式送出。

Warning

progress_callback 属于调用，而不是 Client。没有对应的构造函数参数，因为不同的调用想要不同的回调：这一次驱动下载进度条，下一次是一行日志。

Check

现在删掉 progress_callback=show，再运行一次：

{'result': 'Imported 2 records.'}

没有错误，没有警告，结果相同。report_progress 在调用方没有请求进度时是空操作，所以可以无条件地报告，永远不用操心有没有人在听。

不知道总量时

total 用在知道分母的时候。很多时候并不知道：你在消费一个 feed、遍历一个游标、下载一个没有长度头的东西。

那就省略它：

server.py

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, message=f"Imported {record}")
    return f"Imported {imported} records."

回调收到的是 total=None。客户端仍然可以显示有动静（“目前已导入 3 条……”），但显示不了百分比。不要为了进度条好看而编造一个总量。

Tip

progress 不一定非得数某样特定的东西。字节、行、页：选用户认得出的单位，并且只承诺你能兑现的 total。

回顾

在任何接收 Context 的工具里调用 await ctx.report_progress(progress, total=None, message=None)。
客户端给 call_tool 传 progress_callback=：按调用传，永远不在 Client 上设。
回调的形式是 async (progress, total, message) -> None，在工具还在运行时就会触发。
调用上没有回调，report_progress 就什么都不做。无条件地报告即可。
不知道 total 就省略；回调拿到的是 None。

进度是运行中的工具展示给用户看的。它为你——运维这台服务器的人——记录的那些日志行走的是另一条通道：日志。