進度

機器翻譯

本頁是從英文說明文件自動翻譯而來，以英文頁面為準。如果哪裡讀起來不對勁，翻譯有說明如何回報。

一個要跑三十秒、而這三十秒內毫無動靜的工具，看起來就像壞了。

進度通知就是用來解決這件事。工具回報自己做到哪裡；用戶端決定拿它畫什麼：進度條、轉圈圈的圖示，或一行記錄。

從工具回報

接收一個 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 的輸入 schema 只有一個屬性，urls。Context 那一頁專門講這個物件；進度只是它提供的功能之一。

從用戶端監聽

用戶端是逐次呼叫選擇加入的，做法是把 progress_callback= 傳給 call_tool：

client.py

import anyio
from mcp import Client

from server import mcp


async def show(progress: float, total: float | None, message: str | None) -> None:
    print(f"{message} ({progress}/{total})")


async def main() -> None:
    async with Client(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(mcp) 在記憶體內直接連上伺服器物件，和測試那一頁用的是同一個用戶端。不管 Client 用哪種傳輸方式，progress_callback 都是同一個參數；接下來看到的時序則是記憶體內連線的。它會就地執行回呼，所以每一筆回報都會在 call_tool 回傳之前送達。換成真正的傳輸方式，通知會和結果競速，一個慢的回呼在 call_tool 回傳之後可能還在執行。

試試看

把 client.py 放在 server.py 旁邊，然後執行：

python client.py

Imported https://example.com/a.json (1/2)
Imported https://example.com/b.json (2/2)
{'result': 'Imported 2 records.'}

伺服器上的每一個 await ctx.report_progress(...) 都變成用戶端上對 show 的一次呼叫，依序發生，而且兩行都在 call_tool 回傳之前印出。進度不會打包進結果裡；它在工具還在執行時就持續串流過來。

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 imported so far...」），但沒辦法顯示百分比。不要為了讓進度條好看一點就捏造一個總量。

Tip

progress 不一定要數某個特定的東西。位元組、資料列、頁數：挑使用者認得的單位，而且只承諾做得到的 total。

重點回顧

在任何接收 Context 的工具裡呼叫 await ctx.report_progress(progress, total=None, message=None)。
用戶端把 progress_callback= 傳給 call_tool：逐次呼叫，永遠不是設在 Client 上。
回呼的形式是 async (progress, total, message) -> None，在工具還在執行時就會觸發。
呼叫時沒有回呼，report_progress 就什麼都不做。無條件回報就好。
不知道 total 就省略；回呼會收到 None。

進度是執行中的工具給使用者看的。它為你（操作伺服器的人）記下的那些行，是另一條通道：記錄。