Перейти до змісту

Перебіг виконання

Машинний переклад

Цю сторінку перекладено автоматично з англомовної документації, і основною версією є англомовна сторінка. Якщо щось читається неправильно, на сторінці Переклади пояснено, як про це повідомити.

Інструмент, який працює тридцять секунд і всі тридцять секунд мовчить, здається зламаним.

Сповіщення про перебіг виконання це виправляють. Інструмент повідомляє, скільки вже зроблено, а клієнт вирішує, що з цього намалювати: смужку, спінер чи рядок у лозі.

Надсилання з інструмента

Додайте параметр 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.

Перебіг виконання — це те, що інструмент під час роботи показує користувачеві. Рядки, які він записує в лог для вас, людини, що експлуатує сервер, — це інший канал: Логування.