Saltar a contenido

Progreso

Traducción automática

Esta página se tradujo automáticamente a partir de la documentación en inglés, y la página en inglés es la versión de referencia. Si algo no se lee bien, Traducciones explica cómo avisarnos.

Una herramienta que tarda treinta segundos y no dice nada durante treinta segundos parece rota.

Las notificaciones de progreso lo solucionan. La herramienta informa de cuánto lleva avanzado; el cliente decide qué dibujar con eso: una barra, un indicador giratorio, una línea de log.

Repórtalo desde la herramienta

Acepta un parámetro Context y llama a 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."

Tres argumentos, y tú decides qué significan:

  • progress: cuánto llevas avanzado. La especificación exige que aumente con cada reporte; nunca repitas un valor ni retrocedas.
  • total: cuánto hay en total, si lo sabes. Opcional.
  • message: una línea legible para humanos sobre este paso. Opcional.

ctx se inyecta por su anotación de tipo y el modelo nunca lo ve: el esquema de entrada de import_catalog tiene una sola propiedad, urls. La página El Context trata por completo de ese objeto; el progreso es una de las cosas que te da.

Escúchalo desde el cliente

El cliente lo activa por llamada, pasando progress_callback= a call_tool:

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)

El callback es una función async que recibe exactamente lo que reportó el servidor: progress, total, message.

Info

progress_callback es el mismo parámetro le pases lo que le pases a Client: una URL como aquí, un StdioServerParameters o el objeto servidor en una prueba. Eso sí, ten en cuenta los tiempos con un transporte real. Cada notificación se entrega por su cuenta, al margen de la respuesta, así que un callback lento puede seguir ejecutándose después de que call_tool haya devuelto. Solo la conexión de prueba en el mismo proceso ejecuta el callback de forma directa y garantiza que cada reporte llegue antes.

Pruébalo

Sirve server.py por HTTP y luego ejecuta el cliente desde una segunda 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.'}

Cada await ctx.report_progress(...) en el servidor se convirtió en una llamada a show en el cliente, en orden. El progreso no va empaquetado en el resultado. Se transmite mientras la herramienta sigue trabajando.

Warning

progress_callback pertenece a la llamada, no al Client. No hay un argumento del constructor para él, porque llamadas distintas quieren callbacks distintos: una maneja una barra de descarga, la siguiente una línea de log.

Check

Ahora borra progress_callback=show y ejecútalo de nuevo:

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

Ningún error, ningún aviso, el mismo resultado. report_progress no hace nada cuando quien llama no pidió progreso, así que reportas sin condiciones y nunca tienes que preguntarte si alguien está escuchando.

Cuando no conoces el total

total es para cuando conoces el denominador. A menudo no es así: estás vaciando un feed, recorriendo un cursor, descargando algo sin cabecera de longitud.

Omítelo:

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

El callback recibe total=None. Un cliente todavía puede mostrar actividad ("3 imported so far...") pero no puede mostrar un porcentaje. No te inventes un total para conseguir una barra más bonita.

Tip

progress no tiene por qué contar nada en particular. Bytes, filas, páginas: elige la unidad que el usuario reconocería, y promete solo un total que puedas cumplir.

Resumen

  • await ctx.report_progress(progress, total=None, message=None) desde cualquier herramienta que reciba un Context.
  • El cliente pasa progress_callback= a call_tool: por llamada, nunca en el Client.
  • El callback es async (progress, total, message) -> None y se dispara mientras la herramienta sigue ejecutándose.
  • Si la llamada no lleva callback, report_progress no hace nada. Reporta sin condiciones.
  • Omite total cuando no lo conozcas; el callback recibe None.

El progreso es lo que una herramienta en ejecución le muestra al usuario. Las líneas que registra para ti, la persona que opera el servidor, van por otro canal: Logging.