Типы аргументов и результата

Подготовка к этой главе

Нужны Linux, Python 3.12 и активное окружение из первого сервера. Продолжайте свой diag_server.py после предыдущей главы. Если начинаете здесь, скачайте исходное состояние — стадия 2, распакуйте в новую папку и создайте данные через python3 make_lab.py. Не перезаписывайте свою работу: сначала сохраните её отдельно.

Ситуация

log_tail(lines=100000) вернёт весь журнал — больше, чем поместится в контекст модели. log_tail(lines=-5) вернёт что-то странное без ошибки. Ответ — одна строка текста: чтобы узнать уровень или номер строки, модели придётся разбирать текст и ошибаться.

Главная мысль

Типы — это часть контракта, которую проверяет код. SDK строит из аннотаций функции схему аргументов (inputSchema) и проверяет каждый вызов до того, как вызовет вашу функцию. Модель результата даёт схему результата (outputSchema) и структурированный ответ (structuredContent). Описание в docstring помогает модели, но ничего не запрещает; аннотации — запрещают.

Аргументы с пределами

Начните с файла после главы 2 (свой или project/stages/02/diag_server.py). Добавьте импорты и замените log_tail:

from typing import Annotated, Literal

from pydantic import BaseModel, Field

MAX_LINE = 500                 # символов строки журнала в ответе
Level = Literal["DEBUG", "INFO", "WARN", "ERROR"]


class LogLine(BaseModel):
    line_no: int
    level: str
    text: str
    cut: bool = Field(description="строка обрезана до MAX_LINE символов")


class TailResult(BaseModel):
    lines: list[LogLine]


def parse_line(no: int, raw: str) -> LogLine:
    parts = raw.split(" ", 2)
    level = parts[1] if len(parts) > 1 and parts[1] in ("DEBUG", "INFO", "WARN", "ERROR") else "?"
    return LogLine(line_no=no, level=level, text=raw[:MAX_LINE], cut=len(raw) > MAX_LINE)


def raw_lines() -> list[str]:
    return LOG.read_text(encoding="utf-8", errors="replace").splitlines()


def read_lines() -> list[LogLine]:
    return [parse_line(no, raw) for no, raw in enumerate(raw_lines(), 1)]


@mcp.tool()
def log_tail(
    lines: Annotated[int, Field(ge=1, le=200, description="сколько последних строк")] = 20,
    level: Level | None = None,
) -> TailResult:
    """Последние строки журнала приложения, при желании только одного уровня."""
    found = [x for x in read_lines() if level is None or x.level == level]
    return TailResult(lines=found[-lines:])

Схема аргументов, которую клиент получит в tools/list:

{"type": "object", "properties": {
  "lines": {"default": 20, "description": "сколько последних строк", "maximum": 200, "minimum": 1, "type": "integer"},
  "level": {"anyOf": [{"enum": ["DEBUG", "INFO", "WARN", "ERROR"], "type": "string"}, {"type": "null"}], "default": null}}}

Вызов log_tail(lines=500) теперь не доходит до функции:

is_error=True
Error executing tool log_tail: 1 validation error for log_tailArguments
lines
  Input should be less than or equal to 200 [type=less_than_equal, input_value=500, input_type=int]

level="error" (строчными) тоже отклоняется: допустимы ровно четыре значения. Это лучше, чем молча вернуть пусто: модель видит, что ошиблась в аргументе, а не что ошибок в журнале нет.

Результат структурой

Тип возврата TailResult даёт схему результата и структурированный ответ:

{"lines": [
  {"line_no": 1833, "level": "ERROR", "text": "2026-10-16T09:03:12 ERROR notes.storage write failed: No space left on device path=/srv/app/data/notes", "cut": false},
  {"line_no": 1837, "level": "ERROR", "text": "2026-10-16T09:03:36 ERROR notes.storage write failed: No space left on device path=/srv/app/data/notes", "cut": false}]}

SDK кладёт тот же JSON и в текстовую часть ответа (content) — для клиентов, которые не читают структуру.

В журнале есть строка длиной 20 000 символов. MAX_LINE обрезает её до 500 и ставит cut: true: модель знает, что видит не всё, и не делает вывод из обрезка как из целой строки.

log_levels тоже переведите на структуру: верните dict[str, int] вместо строки. Схема станет объектом с числами, а клиенту не придётся разбирать текст.

Самостоятельно

1. Напишите log_search — поиск подстроки в журнале:
   • аргументы: text — строка от 1 до 100 символов (подстрока, не регулярное выражение); level — как у log_tail; limit — от 1 до 50, по умолчанию 20;
   • результат — модель SearchResult с полями summary: str (короткая сводка для человека: сколько найдено и последняя найденная строка) и matches: list[LogLine];
   • ищите в полной строке журнала, а в ответ отдавайте parse_line(...) — с обрезкой;
   • остановитесь, когда найдено limit совпадений.
2. Почему подстрока, а не регулярное выражение? Запишите одну причину (подсказка: сколько времени может работать выражение вида (a+)+$ на длинной строке).
3. Допишите в свой клиент две проверки и запустите его: log_tail(lines=500) даёт is_error=True; log_tail(lines=50, level="ERROR") возвращает только строки уровня ERROR (проверьте поле level каждой строки в result.structured_content). Пусть клиент печатает «OK» или «FAIL» для каждой.

Проверка

python check.py --chapter 3 diag_server.py

Ожидается Итог: выполнено 13 из 13 проверок, пройдено 13. — 5 проверок главы 2 и 8 новых: схема результата, пределы lines, отказ на 500 и на FATAL, фильтр уровня, предел длины text у log_search, 13 строк с No space (первая — строка 1501), обрезка длинной строки.

Если результат другой

• log_search находит 13 строк не проходит, а у вас 12 или 14 — проверьте limit (по умолчанию 20, проверка передаёт 50) и что ищете в полной строке.
• Схема результата пустая — у функции нет аннотации типа возврата или она возвращает dict без модели.
• log_levels в проверке главы 2 перестал проходить — верните dict[str, int] со значениями-числами.

Решение — answers/03-typed-tools.md; полный файл — project/stages/03/diag_server.py.

Источники

• Архитектура MCP: inputSchema в ответе tools/list.
• Изменения ревизии 2026-07-28: схемы аргументов и результата — JSON Schema 2020-12, structuredContent — любой JSON.
• Пакет mcp 2.3.0: MCPServer.tool(structured_output=...) выбирает структурированный результат по аннотации типа возврата (mcp/server/mcpserver/server.py).

class SearchResult(BaseModel):
    summary: str
    matches: list[LogLine]


@mcp.tool()
def log_levels() -> dict[str, int]:
    """Сколько строк каждого уровня в журнале."""
    counts: dict[str, int] = {}
    for line in read_lines():
        counts[line.level] = counts.get(line.level, 0) + 1
    return counts


@mcp.tool()
def log_search(
    text: Annotated[str, Field(min_length=1, max_length=100, description="искомая подстрока, не регулярное выражение")],
    level: Level | None = None,
    limit: Annotated[int, Field(ge=1, le=50)] = 20,
) -> SearchResult:
    """Ищет подстроку в журнале и возвращает не больше limit совпадений."""
    matches = []
    for no, raw in enumerate(raw_lines(), 1):
        line = parse_line(no, raw)
        if text in raw and (level is None or line.level == level):
            matches.append(line)
            if len(matches) == limit:
                break
    summary = f"Найдено строк: {len(matches)}."
    if matches:
        summary += f" Последняя: {matches[-1].text}"
    return SearchResult(summary=summary, matches=matches)

        r = await client.call_tool("log_tail", {"lines": 500})
        print("OK" if r.is_error else "FAIL", "lines=500 отклонён")
        r = await client.call_tool("log_tail", {"lines": 50, "level": "ERROR"})
        levels = {x["level"] for x in r.structured_content["lines"]}
        print("OK" if levels == {"ERROR"} else "FAIL", "только ERROR")