進捗
30 秒かかるツールが 30 秒間なにも言わなければ、壊れているように見えます。
進捗通知はそれを解決します。ツールはどこまで進んだかを報告し、クライアントはそれを使って何を描くかを決めます。プログレスバー、スピナー、ログの 1 行などです。
ツールから報告する
Context パラメーターを受け取り、report_progress を呼び出してください。
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."
引数は 3 つで、その意味は自分で決めます。
progress：どこまで進んだか。仕様では、報告のたびに増加することが必須です。同じ値を繰り返したり、減らしたりしないでください。total：全体でどれだけあるか（わかっている場合）。省略可能です。message：「この」ステップについての、人が読める 1 行。省略可能です。
ctx は型ヒントによって注入され、モデルからは決して見えません。import_catalog の入力スキーマにあるプロパティは urls の 1 つだけです。Context のページはこのオブジェクトについて詳しく扱っています。進捗はそれが提供するものの 1 つです。
クライアントで受け取る
クライアントは、call_tool に progress_callback= を渡すことで、呼び出しごとにオプトインします。
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) はサーバーオブジェクトにメモリ内で直接接続します。テスト のページの土台になっているのと同じクライアントです。progress_callback は、Client がどのトランスポートを使っていても同じパラメーターです。これから目にする「タイミング」はメモリ内接続のものです。メモリ内接続はコールバックをインラインで実行するため、すべての報告が 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 の 1 回の呼び出しになり、2 行とも call_tool が返る前に出力されました。進捗は結果にまとめられるのではなく、ツールがまだ動いている間にストリーミングされます。
Warning
progress_callback は Client ではなく、呼び出しに属します。そのためのコンストラクター引数はありません。呼び出しごとに必要なコールバックが違うからです。ある呼び出しはダウンロードバーを動かし、次の呼び出しはログの 1 行を出します。
Check
今度は progress_callback=show を削除して、もう一度実行してください。
{'result': 'Imported 2 records.'}
エラーも警告もなく、結果は同じです。report_progress は、呼び出し側が進捗を要求しなかったときは何もしません。ですから無条件に報告すればよく、誰かが聞いているかどうかを気にする必要はありません。
全体量がわからないとき
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, 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)を呼べます。- クライアントは
call_toolにprogress_callback=を渡します。呼び出しごとであり、Clientには渡しません。 - コールバックは
async (progress, total, message) -> Noneで、ツールがまだ実行中の間に呼ばれます。 - 呼び出しにコールバックがなければ、
report_progressは何もしません。無条件に報告してください。 - わからないときは
totalを省略します。コールバックはNoneを受け取ります。
進捗は、実行中のツールが「ユーザー」に見せるものです。サーバーを運用する「自分」のために記録する行は、別のチャネルです。ロギング を参照してください。