Fortschritt

Maschinelle Übersetzung

Diese Seite wurde automatisch aus der englischen Dokumentation übersetzt, und die englische Seite ist die maßgebliche Fassung. Wenn sich etwas falsch liest, erklärt Übersetzungen, wie du es melden kannst.

Ein Tool, das dreißig Sekunden braucht und dreißig Sekunden lang schweigt, wirkt kaputt.

Fortschrittsbenachrichtigungen beheben das. Das Tool meldet, wie weit es ist; der Client entscheidet, was er daraus zeichnet: einen Balken, einen Spinner, eine Log-Zeile.

Aus dem Tool melden

Nimm einen Context-Parameter entgegen und rufe report_progress auf:

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."

Drei Argumente, und du bestimmst, was sie bedeuten:

progress: wie weit du bist. Die Spezifikation verlangt, dass der Wert mit jeder Meldung steigt; wiederhole nie einen Wert und geh nie rückwärts.
total: wie viel es insgesamt ist, falls du es weißt. Optional.
message: eine menschenlesbare Zeile über diesen Schritt. Optional.

ctx wird wegen seines Type Hints injiziert, und das Modell sieht ihn nie: Das Eingabeschema von import_catalog hat eine einzige Property, urls. Die Seite Der Context dreht sich ganz um dieses Objekt; Fortschritt ist eines der Dinge, die es dir bietet.

Im Client darauf lauschen

Der Client meldet sich pro Aufruf an, indem er progress_callback= an call_tool übergibt:

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)

Der Callback ist eine async-Funktion, die genau das entgegennimmt, was der Server gemeldet hat: progress, total, message.

Info

progress_callback ist derselbe Parameter, egal was du Client übergeben hast: eine URL wie hier, ein StdioServerParameters-Objekt oder das Server-Objekt in einem Test. Achte über einen echten Transport allerdings auf das Timing. Jede Benachrichtigung wird für sich zugestellt, neben der Response, sodass ein langsamer Callback noch laufen kann, nachdem call_tool bereits zurückgekehrt ist. Nur die In-Process-Testverbindung führt den Callback inline aus und garantiert, dass jede Meldung vorher eintrifft.

Ausprobieren

Stelle server.py über HTTP bereit und starte dann den Client aus einem zweiten Terminal:

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.'}

Jedes await ctx.report_progress(...) auf dem Server wurde zu einem Aufruf von show auf dem Client, in derselben Reihenfolge. Fortschritt wird nicht ins Ergebnis gepackt. Er streamt, während das Tool noch arbeitet.

Warning

progress_callback gehört zum Aufruf, nicht zum Client. Es gibt kein Konstruktorargument dafür, weil verschiedene Aufrufe verschiedene Callbacks wollen: Einer treibt einen Download-Balken an, der nächste eine Log-Zeile.

Check

Lösche jetzt progress_callback=show und starte es erneut:

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

Kein Fehler, keine Warnung, dasselbe Ergebnis. report_progress ist ein No-op, wenn der Aufrufer keinen Fortschritt angefordert hat. Du meldest also bedingungslos und musst dich nie fragen, ob überhaupt jemand zuhört.

Wenn du die Gesamtmenge nicht kennst

total ist für den Fall, dass du den Nenner kennst. Oft kennst du ihn nicht: Du leerst einen Feed, läufst einen Cursor ab, lädst etwas ohne Längen-Header herunter.

Lass es weg:

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."

Der Callback erhält total=None. Ein Client kann weiterhin Aktivität anzeigen („3 imported so far...“), aber keinen Prozentwert. Erfinde keine Gesamtmenge, nur um einen hübscheren Balken zu bekommen.

Tip

progress muss nichts Bestimmtes zählen. Bytes, Zeilen, Seiten: Wähle die Einheit, die die Person am Host wiedererkennt, und versprich nur ein total, das du halten kannst.

Zusammenfassung

await ctx.report_progress(progress, total=None, message=None) aus jedem Tool, das einen Context entgegennimmt.
Der Client übergibt progress_callback= an call_tool: pro Aufruf, nie am Client.
Der Callback ist async (progress, total, message) -> None und feuert, während das Tool noch läuft.
Kein Callback am Aufruf heißt: report_progress tut nichts. Melde bedingungslos.
Lass total weg, wenn du es nicht kennst; der Callback bekommt None.

Fortschritt ist das, was ein laufendes Tool der Person am Host zeigt. Die Zeilen, die es für dich loggt – für dich, weil du den Server betreibst –, sind ein anderer Kanal: Logging.