Свой MCP-сервер

Пошагово: что нужно сделать, чтобы собственный сервис заработал в EVA как источник инструментов.

Основное правило

Все поставляемые сервисы, кроме searcher, написаны по одному внутреннему регламенту. Он формулируется одной фразой:

Каждому инструменту своя папка со всей его внутренней разбивкой внутри. Сам инструмент это строка-делегат. Всё общее выносится в отдельный слой.

Ниже разобрано, что именно требуется.

Шаг 1. Структура проекта

app/
  main.py                 сборка сервера, выбор транспорта, запуск
  config.py               настройки, никаких секретов в коде
  logging.py              настройка логов и скрытие ключей
  tools/                  внешняя поверхность, по папке на инструмент
    __init__.py           создание сервера и импорт инструментов
    _common.py            общие вспомогательные функции слоя инструментов
    init_eva_session/
      tool.py             привязка сессии, только для Runtime
    <ваш инструмент>/
      __init__.py         импортирует tool
      tool.py             одна строка: вернуть ok(await service.run(...))
      service.py          порядок работы этого инструмента
      <engine>.py         внутренности, принадлежащие этому инструменту
  shared/                 формат ответа, ошибки, общие модели
  session/session.py      привязка сессии, доступы и их срок
  storage/                base.py, keys.py, http_client.py, chat_s3.py

Жёсткие правила:

• файл tool.py содержит одну строку;
• папка одного инструмента никогда не импортирует папку другого;
• если что-то нужно двум инструментам, это переносится в общий слой;
• общий слой не место для внутренностей одного механизма.

Шаг 2. Единый формат ответа

Все инструменты возвращают один и тот же конверт:

class Envelope(BaseModel):
    ok: bool
    result: Any = {}
    warnings: list[Any] = []
    errors: list[Any] = []
    hints: list[str] = []
    code: str | None = None

Правило: инструменты возвращают ok(...). Они никогда не возвращают простой словарь и никогда не выбрасывают исключение наружу.

Общая обёртка превращает ожидаемую ошибку в конверт с кодом, а неожидаемую записывает в лог и возвращает с кодом INTERNAL.

Пример ответа с ошибкой:

{
  "ok": false,
  "result": {},
  "errors": ["Chat file not found: report.docx"],
  "hints": [],
  "code": "CHAT_FILE_NOT_FOUND"
}

Шаг 3. Привязка сессии

Реализуйте инструмент init_eva_session по описанию из раздела Контракт init_eva_session. Ключевое: проверка личности до всего остального, обязательная проверка совпадения префикса с идентификатором чата, сохранение только простых данных.

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

Шаг 4. Слой доступа к файлам

Ровно три части, не больше и не меньше.

Оркестратор (chat_s3.py, класс ChatS3Storage)
   ├── Keyspace (keys.py)          только правила имён, без сети. Граница безопасности
   ├── HttpClient (http_client.py) транспорт: один общий клиент на процесс и повторы
   └── Credentials                 доступы и проверка их срока

Оркестратор называйте по названию хранилища, а не отвлечённо. То есть chat_s3.py и ChatS3Storage, а не blob.py и BlobStorage.

Правила имён и проверки границы описаны в разделе Контракт chat S3.

Что тестировать в первую очередь

Проверку границы и разбор имени файла нужно покрыть тестами исчерпывающе. Именно здесь ошибка приводит к утечке файлов одного чата в другой.

Шаг 5. Возвращайте файл, а не текст

Это правило определяет вид ответа у всех сервисов, работающих с файлами:

return ok({"artifact": {"file_name": name, "artifact_key": key,
                        "bytes": n, "sha256": h}})

Причина прямая: если вернуть полный текст, модель напечатает весь документ в переписку вместо того, чтобы отдать файл. Когда у модели есть только ссылка, печатать нечего.

Неудачная запись результата это жёсткая ошибка. Возвращать текст вместо файла как резервный вариант нельзя.

Запись байтов это не то же самое, что появление файла в чате

Регистрация карточки файла в чате это работа Runtime, а не ваша.

Шаг 6. Скрытие ключей в логах

Нужны обе защиты одновременно.

_TOKEN_RE = re.compile(r"(token=)[^&\s\"'>]+", re.I)

class RedactingFormatter(logging.Formatter):
    def format(self, r):
        return _TOKEN_RE.sub(r"\1[REDACTED]", super().format(r))

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

Шаг 7. Настройки и транспорт

Все несекретные настройки в одном файле. Порядок применения: значения по умолчанию, затем файл настроек, затем переменные окружения.

Сервер создаётся один раз в tools/__init__.py из настроек, а импорты папок инструментов идут ниже его создания, чтобы декораторы успели сработать.

Транспорт	Когда выбирать
stdio	Runtime сам запускает процесс сервиса
streamable-http	Сервис работает по сети, адрес вида http://<хост>:<порт>/mcp

Транспорт выбирается настройкой, а не ветвлением в коде.

Только поверхность MCP

Никаких дополнительных веб-интерфейсов, брокеров и вторых протоколов, пока нет отдельного настоящего потребителя. Проверку состояния делайте инструментом MCP или отдельным маленьким процессом рядом. Если убираете брокер, убирайте целиком.

Шаг 8. Порядок запуска

Строго такой:

config.validate()             # 1. упасть сразу, если настроек нет
logging_setup.configure()     # 2. логи до того, как что-то начнёт писать
start_background_workers()    # 3. обработчики до приёма вызовов
from app.tools import public_mcp   # 4. импорт последним, декораторы срабатывают тут
try:
    public_mcp.run("streamable-http")   # 5. блокирующий запуск
finally:
    cleanup()                 # 6. всегда освобождать тяжёлые ресурсы

Шаг 9. Тяжёлая работа

Случай	Как делать
Блокирующая работа	Выносить в отдельный поток
Единственный неповторнововходимый ресурс	Один рабочий поток или общая блокировка
Тяжёлые модели	Один владелец, ленивая загрузка и выгрузка после работы
Тяжёлые и закрытые зависимости	Импортировать внутри функции, которая их использует

Проверка: импорт сервера должен проходить, когда установлены только описание моделей данных и библиотека MCP.

Шаг 10. Тесты

Что	Когда запускается
Формат ответа, коды ошибок, правила имён, схемы параметров, разбор настроек	Всегда
Работа с настоящими моделями	Только когда закрытый пакет установлен
Поверхность инструментов через клиент MCP в том же процессе	Всегда

Шаг 11. Подключение к платформе

Готовый сервис подключается администратором в разделе MCP. Нужны адрес и транспорт.

После подключения помните две вещи.

Платформа переименует ваши инструменты для модели по правилу mcp_<имя источника>_<имя инструмента>. В своих указаниях модели пишите про инструменты так, чтобы это работало и с новым именем.

Платформа проверит указания сервиса и описания всех инструментов на попытки подменить поведение модели. Прямые повеления вида «ты обязан» и «никогда не делай» именно такую проверку и настораживают. Если источник получит состояние «заблокирован», он не попадёт ни в одно обращение модели.

Чего делать не нужно

Не нужно	Почему
Принимать chatId, token, endpoint или bucket в параметрах рабочего инструмента	Их формирует модель. Все доступы приходят только через привязку сессии
Хранить секреты в файле настроек	Прямое нарушение регламента
Возвращать текст документа в переписку	Модель напечатает его целиком
Держать открытое соединение в состоянии сессии	Состояние сессии сериализуется, живой объект вызовет ошибку
Делать отдельный инструмент обновления доступов	Повторная привязка это и есть обновление