Уведомления на практике

Теория из прошлого урока умещается в один метод ctx.report_progress(). Но чтобы прочувствовать, как уведомления обгоняют ответ, нужно увидеть это в терминале. Построим сервер reports с инструментом, который «анализирует» задачи спринта по одной, и клиента, который печатает каждое уведомление в момент прихода. Потом запустим ту же пару через Streamable HTTP и убедимся, что поведение не изменилось.

Шаг 1. Сервер с долгим инструментом

Создайте проект (uv init reports && cd reports && uv add "mcp[cli]") и файл server.py:

import logging

import anyio

from mcp.server import MCPServer
from mcp.server.mcpserver import Context

logger = logging.getLogger(__name__)
mcp = MCPServer("reports")

# Заглушка вместо трекера: задачи спринта и «время анализа» каждой в секундах.
SPRINTS: dict[str, list[tuple[str, float]]] = {
    "2026-09": [
        ("WEB-301", 0.2),
        ("WEB-305", 0.4),
        ("WEB-310", 0.1),
        ("MOB-88", 0.3),
        ("MOB-91", 0.2),
        ("QA-17", 0.1),
    ],
}


async def analyze(key: str, seconds: float) -> str:
    """Имитация долгой работы: в настоящем сервере здесь запрос к трекеру и обработка."""
    await anyio.sleep(seconds)
    return f"{key}: закрыта, оценка совпала с фактом"


@mcp.tool()
async def build_report(sprint: str, ctx: Context) -> str:
    """Собрать отчёт по спринту: пройти по всем задачам и составить сводку."""
    tasks = SPRINTS.get(sprint)
    if tasks is None:
        return f"Спринт {sprint} не найден."

    logger.info("build_report: sprint=%s, tasks=%d", sprint, len(tasks))
    total = len(tasks)
    lines: list[str] = []
    for i, (key, seconds) in enumerate(tasks, start=1):
        lines.append(await analyze(key, seconds))
        await ctx.report_progress(i, total, f"Проанализирована {key}")

    logger.info("build_report: done")
    return f"Отчёт по спринту {sprint}\n" + "\n".join(lines)


if __name__ == "__main__":
    mcp.run()

Три вещи, на которые стоит обратить внимание. Инструмент асинхронный (async def): синхронные функции SDK выполняет в отдельном потоке, и await в них невозможен. ctx: Context — параметр, который SDK подставляет сам; для модели инструмент принимает только sprint. И logger.info вместо print: в stdio-сервере stdout принадлежит протоколу.

Шаг 2. Клиент, который слушает прогресс

Файл check.py:

import time

import anyio

from mcp import Client

from server import mcp

started = time.monotonic()


def stamp() -> str:
    return f"{time.monotonic() - started:5.2f}s"


async def on_progress(progress: float, total: float | None, message: str | None) -> None:
    print(f"[{stamp()}] прогресс {progress:g}/{total:g}: {message}")


async def main() -> None:
    async with Client(mcp) as client:
        print(f"[{stamp()}] вызываю build_report")
        result = await client.call_tool(
            "build_report",
            {"sprint": "2026-09"},
            progress_callback=on_progress,
        )
        print(f"[{stamp()}] результат получен, is_error={result.is_error}")
        print(result.content[0].text)


anyio.run(main)

Запускаем uv run python check.py. Ожидаемый вывод (метки времени будут немного отличаться):

[ 0.00s] вызываю build_report
[ 0.20s] прогресс 1/6: Проанализирована WEB-301
[ 0.60s] прогресс 2/6: Проанализирована WEB-305
[ 0.70s] прогресс 3/6: Проанализирована WEB-310
[ 1.00s] прогресс 4/6: Проанализирована MOB-88
[ 1.20s] прогресс 5/6: Проанализирована MOB-91
[ 1.30s] прогресс 6/6: Проанализирована QA-17
[ 1.31s] результат получен, is_error=False
Отчёт по спринту 2026-09
WEB-301: закрыта, оценка совпала с фактом
...

Метки времени — главное в этом выводе. Уведомления приходят по мере работы, а не пачкой в конце, и результат появляется только после последнего. Если убрать progress_callback из вызова, строки прогресса исчезнут, а результат придёт в ту же секунду: сервер вызывает report_progress, но без токена SDK ничего не отправляет.

В stderr при этом появятся две строки от logger.info — SDK настраивает корневой логгер на уровень INFO при создании MCPServer. Их можно отключить, передав log_level="WARNING" в конструктор.

Шаг 3. То же самое через Streamable HTTP

In-memory клиент вызывает сервер напрямую. Проверим, что через настоящий транспорт прогресс приходит так же. В одном терминале запустите сервер по HTTP:

uv run mcp run server.py --transport streamable-http

Сервер будет слушать http://127.0.0.1:8000/mcp. В check.py замените Client(mcp) на Client("http://127.0.0.1:8000/mcp") и запустите во втором терминале. Вывод должен совпасть с точностью до меток времени.

Что произошло по проводу: клиент отправил POST /mcp с телом tools/call, в _meta которого лежал progressToken. Сервер ответил не одним JSON-объектом, а потоком text/event-stream: шесть событий с notifications/progress, затем событие с ответом, после чего поток закрылся. Как это устроено, разберём в уроке «Транспорт Streamable HTTP».

Шаг 4. Неизвестный total

Не всегда известно, сколько всего работы. Добавьте инструмент, который обходит задачи постранично и не знает общего числа:

@mcp.tool()
async def scan_overdue(ctx: Context) -> str:
    """Найти просроченные задачи, обходя трекер постранично."""
    found = 0
    page = 0
    while True:
        page += 1
        await anyio.sleep(0.2)
        keys = [f"WEB-{300 + page * 3 + i}" for i in range(3)] if page <= 3 else []
        if not keys:
            break
        found += len(keys)
        await ctx.report_progress(found, message=f"Страница {page}, найдено {found}")
    return f"Просроченных задач: {found}"

total не передаём; в колбэк клиента придёт None. Раз total в колбэке может быть None, форматирование {total:g} из шага 2 упадёт — исправьте вывод так, чтобы он это учитывал. Именно ради таких случаев спецификация требует, чтобы progress рос сам по себе: клиент может показать хотя бы «идёт работа, найдено 9».

Шаг 5. Протокольное логирование: узнать в чужом коде

Вы будете встречать серверы, которые вместо logger вызывают ctx.info(...). Чтобы понимать, что при этом происходит, добавьте в build_report одну строку после проверки спринта:

    await ctx.info(f"Начинаю отчёт по спринту {sprint}", logger_name="reports")

При запуске check.py вы увидите предупреждение MCPDeprecationWarning о том, что логирование через протокол устарело с 2026-07-28. Само сообщение при этом никуда не пришло: клиент не запросил логи. Чтобы получить их, нужно и колбэк, и явный уровень:

from mcp.types import LoggingMessageNotificationParams


async def on_log(params: LoggingMessageNotificationParams) -> None:
    print(f"[лог {params.level}] {params.data}")


async with Client(mcp, logging_callback=on_log, log_level="info") as client:
    ...

Теперь строка [лог info] Начинаю отчёт по спринту 2026-09 появится перед первым прогрессом. log_level в конструкторе — это тот самый ключ io.modelcontextprotocol/logLevel, который клиент на современном протоколе ставит в _meta каждого запроса. Без него сервер обязан молчать.

Поэкспериментировав, удалите ctx.info: в новом коде логи должны идти в logger. Оставьте этот шаг в памяти как способ прочитать чужой сервер.