Реализуем клиент

В уроке про хосты мы говорили, что программная сторона MCP — это когда цикл вызова инструментов пишете вы сами. Пора это сделать. Клиент, который мы напишем, умещается в сотню строк, и после него внутренности Claude Code перестанут быть магией. Понадобится ключ Anthropic API.

Подготовка

Добавьте в проект две зависимости:

uv add anthropic python-dotenv

(pip install anthropic python-dotenv для pip.) Положите ключ в файл .env в корне проекта:

ANTHROPIC_API_KEY=ваш-ключ

И убедитесь, что .env есть в .gitignore. Anthropic SDK сам подхватит переменную ANTHROPIC_API_KEY из окружения; python-dotenv нужен только для того, чтобы загрузить её из файла.

Две библиотеки, две роли

В клиенте встречаются два SDK, и важно не путать, кто за что отвечает.

mcp — общение с сервером. Client запускает сервер, спрашивает список инструментов, вызывает их. Он ничего не знает о моделях.

anthropic — общение с Claude. messages.create отправляет разговор модели вместе с описанием инструментов и получает ответ, в котором может быть просьба вызвать инструмент. Он ничего не знает об MCP.

Клиент — переводчик между ними: берёт инструменты из mcp и описывает их для anthropic, берёт просьбы модели из anthropic и выполняет через mcp.

Подключение к серверу

Создайте client.py:

import asyncio
import sys

from anthropic import Anthropic
from dotenv import load_dotenv
from mcp import Client, StdioServerParameters
from mcp_types import TextContent

load_dotenv()

MODEL = "claude-opus-5"
anthropic = Anthropic()


def text_of(result) -> str:
    """Собрать текстовые блоки ответа инструмента в одну строку."""
    return "\n".join(b.text for b in result.content if isinstance(b, TextContent))

StdioServerParameters описывает, как запустить сервер: команда и аргументы. Сам запуск делает Client, когда мы входим в async with. В main это будет выглядеть так:

async def main() -> None:
    server = StdioServerParameters(command="uv", args=["run", "docs_server.py"])
    async with Client(server) as client:
        names = [t.name for t in (await client.list_tools()).tools]
        print("Подключено. Инструменты:", ", ".join(names))
        while True:
            try:
                query = (await asyncio.to_thread(input, "\n> ")).strip()
            except EOFError:
                break
            if not query or query.lower() in {"quit", "exit"}:
                break
            try:
                print(await run(client, [{"role": "user", "content": query}]))
            except Exception as e:
                print("Ошибка:", e)


if __name__ == "__main__":
    asyncio.run(main())

Вход в async with запускает дочерний процесс и договаривается о версии протокола; выход останавливает процесс. Ничего закрывать руками не нужно. input() блокирует поток, поэтому мы выносим его в отдельный поток через asyncio.to_thread — иначе клиент не сможет обрабатывать сообщения сервера, пока вы печатаете.

Одна деталь про окружение: дочерний процесс не наследует ваши переменные окружения целиком, только базовый набор вроде PATH и HOME. Если серверу нужен, например, токен для API, передайте его явно: StdioServerParameters(command=..., args=..., env={"API_TOKEN": "..."}).

Инструменты для модели

Anthropic API ожидает описание инструмента в виде словаря с полями name, description и input_schema. У инструмента из MCP есть ровно эти три вещи — name, description и input_schema, — так что перевод почти буквальный:

async def run(client: Client, messages: list[dict]) -> str:
    """Прогнать разговор через Claude, выполняя вызовы инструментов через сервер."""
    tool_list = await client.list_tools()
    tools = [
        {
            "name": t.name,
            "description": t.description or "",
            "input_schema": t.input_schema,
        }
        for t in tool_list.tools
    ]

Это именно то, что делает Claude Code при старте: собирает инструменты всех серверов в один список для модели. У нас сервер один, но добавить второй — вопрос ещё одного Client и объединения списков.

Цикл вызова инструментов

Продолжение функции run:

    while True:
        response = anthropic.messages.create(
            model=MODEL,
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return "\n".join(b.text for b in response.content if b.type == "text")

        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            print(f"[вызов {block.name} {block.input}]", file=sys.stderr)
            result = await client.call_tool(block.name, block.input)
            results.append(
                {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": text_of(result),
                    "is_error": result.is_error,
                }
            )
        messages.append({"role": "user", "content": results})

Разберём по шагам, потому что это сердце любого хоста.

1. Отправляем модели разговор и список инструментов. Ответ приходит с stop_reason: end_turn — модель закончила, tool_use — просит вызвать инструмент.
2. Ответ модели целиком добавляем в историю как сообщение ассистента. Целиком — вместе с блоками tool_use, иначе на следующем шаге модель не поймёт, к чему относятся результаты.
3. Если инструменты не нужны — возвращаем текст, и цикл окончен.
4. Иначе для каждого блока tool_use вызываем инструмент через сервер: client.call_tool(имя, аргументы). Аргументы модель уже собрала по схеме, мы передаём их как есть.
5. Результаты складываем в одно сообщение пользователя из блоков tool_result. tool_use_id связывает результат с просьбой. Флаг is_error из ответа сервера передаём модели: если инструмент вернул ToolError, она прочитает сообщение и попробует иначе.
6. Возвращаемся в начало цикла: модель получает результаты и решает, что дальше — ответить или вызвать ещё что-то.

Модель может попросить несколько инструментов за раз; поэтому все результаты идут в одном сообщении, а не по одному.

Запуск

uv run client.py

Попробуйте: «Какие риски у проекта?» — в stderr появится [вызов read_doc {'name': 'risks.md'}], а затем ответ. Потом: «Исправь в оценке срок тестирования на четыре недели и покажи, что получилось» — модель вызовет edit_doc, потом read_doc, и вы увидите оба вызова. Спросите про документ, которого нет, — увидите, как модель получает ToolError и честно об этом сообщает.

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

Чего здесь нет

Наш клиент — учебный минимум. У настоящего хоста есть подтверждение вызовов человеком (мы выполняем всё подряд), несколько серверов, переподключение при обрыве, обработка уведомлений, ограничение на число шагов цикла. Каждая из этих вещей — десяток строк поверх того, что уже есть; самое важное — подтверждение — можно добавить одним input() перед call_tool.

Если вы предпочитаете не писать цикл руками, в Anthropic SDK есть готовые помощники для MCP-инструментов (anthropic.lib.tools.mcp), которые превращают инструменты сервера в объекты для встроенного исполнителя циклов. Мы написали цикл вручную, чтобы было видно, как он устроен.