Перейти к содержанию

Ход выполнения

Машинный перевод

Эта страница переведена с английской документации автоматически, и основной версией остаётся английская страница. Если что-то читается неправильно, на странице Переводы объясняется, как об этом сообщить.

Инструмент, который работает тридцать секунд и все тридцать секунд молчит, выглядит сломанным.

Уведомления о ходе выполнения решают эту проблему. Инструмент сообщает, насколько он продвинулся, а клиент решает, что из этого нарисовать: полосу, спиннер, строку в логе.

Отчёт о ходе из инструмента

Примите параметр 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; ход выполнения — одна из возможностей, которые он даёт.

Приём на стороне клиента

Клиент подписывается на каждый вызов отдельно, передавая 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) подключается напрямую к объекту сервера, в памяти, — это тот же клиент, на котором построена страница Тестирование. Параметр 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 на клиенте, в том же порядке, и обе строки напечатались до того, как call_tool вернул управление. Ход выполнения не упаковывается в результат: он передаётся потоком, пока инструмент ещё работает.

Warning

progress_callback относится к вызову, а не к Client. Аргумента конструктора для него нет, потому что разным вызовам нужны разные колбэки: один управляет полосой загрузки, следующий — строкой в логе.

Check

Теперь удалите progress_callback=show и запустите снова:

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

Ни ошибки, ни предупреждения, тот же результат. report_progress ничего не делает, если вызывающая сторона не запросила ход выполнения, поэтому сообщайте о нём безусловно и никогда не гадайте, слушает ли кто-нибудь.

Когда общий объём неизвестен

total нужен, когда известен знаменатель. Часто это не так: вы вычитываете ленту, идёте по курсору, скачиваете что-то без заголовка длины.

Просто не указывайте его:

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, только если сможете это обещание сдержать.

Итоги

  • await ctx.report_progress(progress, total=None, message=None) из любого инструмента, принимающего Context.
  • Клиент передаёт progress_callback= в call_tool: на каждый вызов, никогда не в Client.
  • Колбэк имеет вид async (progress, total, message) -> None и срабатывает, пока инструмент ещё работает.
  • Нет колбэка у вызова — report_progress ничего не делает. Сообщайте безусловно.
  • Опускайте total, когда он неизвестен; колбэк получит None.

Ход выполнения — это то, что работающий инструмент показывает пользователю. Строки, которые он пишет в лог для вас, человека, обслуживающего сервер, — это другой канал: Логирование.