Собственный MCP-адаптер на Flask: описание инструментов, маршрутизация и отладка

При подключении браузерной среды к AI Agent есть неизбежная проблема: стандартные инструменты дают примитивы, а бизнес-действия приходится собирать самостоятельно.

Рассмотрим конкретный пример. Чтобы выполнить один цикл сбора данных, фактическая последовательность такова: «открыть среду для нужного региона, привязать сетевой выход, один раз зайти для прогрева и убедиться, что выход работает», и только после этого передать работу Agent. Готовый MCP Server обычно предоставляет лишь одношаговые инструменты: открыть среду, перейти на страницу, кликнуть, сделать снимок экрана. Никто не собирает указанную цепочку за вас. Agent каждый раз должен выстраивать ее с нуля, что медленно, а при сбое на любом промежуточном шаге сам решать, как действовать дальше.

Именно для этого нужен слой адаптера: логика композиции, проверки и состояние остаются на вашей стороне, а наружу выставляется одно бизнес-действие.

Минимальная рабочая структура

Для рабочего слоя адаптера достаточно трех частей: объявления инструментов, маршрутизации запросов и выполнения с возвратом результата. В Flask все это помещается в один процесс.

Начнем с объявления инструментов: оно определяет, что может видеть Agent.

# adapter/tools.py
TOOLS = [
    {
        "name": "prepare_environment",
        "description": "Подготовить доступную среду для региона и после завершения вернуть ID среды",
        "inputSchema": {
            "type": "object",
            "properties": {
                "region": {"type": "string", "description": "Регион сетевого выхода, например US-CA"},
                "purpose": {"type": "string", "description": "Метка назначения для повторного использования и статистики квот"},
                "timeout": {"type": "integer", "minimum": 10, "maximum": 120, "default": 60},
            },
            "required": ["region"],
            "additionalProperties": False,
        },
    },
    {
        "name": "open_page",
        "description": "Открыть страницу в указанной среде и дождаться, когда она станет интерактивной",
        "inputSchema": {
            "type": "object",
            "properties": {
                "env_id": {"type": "string"},
                "url": {"type": "string"},
            },
            "required": ["env_id", "url"],
            "additionalProperties": False,
        },
    },
]

Маршрутизатор распределяет методы JSON-RPC. Клиент сначала запрашивает tools/list, а затем вызывает tools/call по имени.

# adapter/app.py
from flask import Flask, request, jsonify
from adapter.tools import TOOLS
from adapter.runner import run_tool

app = Flask(__name__)

@app.post("/mcp")
def mcp():
    req = request.get_json(force=True)
    method, rid = req.get("method"), req.get("id")

    if method == "tools/list":
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": {"tools": TOOLS}})

    if method == "tools/call":
        name = req["params"]["name"]
        args = req["params"].get("arguments", {})
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": run_tool(name, args)})

    return jsonify({"jsonrpc": "2.0", "id": rid,
                    "error": {"code": -32601, "message": "method not found"}})

Слой выполнения — единственное место, которое обращается к API среды и протоколу страницы. Здесь выполняется цепочка шагов и здесь же ошибки приводятся к единому формату.

# adapter/runner.py
def run_tool(name, args):
    try:
        payload = HANDLERS[name](**args)
        return ok(payload)
    except ToolError as e:
        return fail(e.code, e.retryable, e.message)

Формат ответа стоит определить заранее. Не передавайте Agent необработанные низкоуровневые исключения: он попытается разобрать эти строки и может принять странное решение. Лучше договориться о структуре с кодом результата, например {"ok": false, "code": "env_unavailable", "retryable": true, "attempts": 3}. Тогда Agent должен решить только две вещи: можно ли повторить попытку и нужно ли передать случай человеку.

Как проектировать параметры

Гранулярность инструментов должна соответствовать бизнес-действиям, а не API. Если обернуть каждый низкоуровневый endpoint отдельным инструментом, абстракции фактически не получится — Agent по-прежнему придется самому определять порядок шагов.

В параметрах есть четкая граница: что должен определять модель, а что — сам слой адаптера. Регион, назначение и целевой URL относятся к первой группе, их заполняет модель. Порт отладки, имена внутренних очередей и выбор пула сред относятся ко второй; не включайте их в schema, иначе рано или поздно модель заполнит их неправильно.

Селекторы — частый источник проблем. Если структура страницы меняется, вызовы с жестко заданными селекторами могут массово сломаться. Лучше позволить модели передавать семантические цели, например относительно устойчивый признак вроде «кнопка входа», а сопоставление селекторов хранить в слое адаптера. Тогда при изменении страницы достаточно поправить одно место.

У числовых параметров обязательно должны быть верхние пределы. Если у timeout, числа повторов или количества страниц в schema нет maximum, модель может передать очень большое значение и превратить один вызов в задачу на десятки минут. Любому инструменту с циклической логикой нужна явная точка завершения. Ограничивайте работу полем вроде max_pages, а не правилом «листать, пока данные не закончатся».

Нужно учитывать и идемпотентность. Пусть вызывающая сторона передает task_id; повторный запрос сможет сразу вернуть предыдущий результат, чтобы при retry Agent не создал среду второй раз.

Возвращаемые данные должны быть компактными. Не отправляйте снимки экрана в base64 — возвращайте ссылку на файл. Для списков указывайте count и признак усечения, а не помещайте всю таблицу в контекст.

При отладке сначала проверьте эти три места

При работе через stdio JSON-RPC занимает stdout, поэтому логам нельзя попадать в stdout. Один случайный print может сразу сломать разбор протокола и заставить долго искать причину. Логи следует направлять в stderr или в файл.

tools/list — первая точка проверки. Если объявления не загрузились, дальнейших вызовов не будет. Сначала проверьте имена инструментов, структуру schema и не блокирует ли additionalProperties какие-либо параметры.

Вторая точка — пошаговый trace. Для каждого вызова записывайте task_id, краткое содержание параметров, длительность и код результата. При проблеме будет видно, где произошла остановка: при создании среды, навигации или проверке. Сохраняйте исходные параметры неудачных случаев, чтобы воспроизводить их без изменений: воспроизводимая ошибка исправляется намного быстрее.

Третья точка — набор фиксированных fixtures: стабильная тестовая страница и несколько неизменных локаторов элементов. Запуск smoke test после каждого изменения адаптера экономит больше времени, чем любая устная проверка.

Границы полномочий

Слой адаптера — место с самой высокой концентрацией полномочий во всей системе: у него есть доступ к учетным данным, средам и действиям на странице. Поэтому границы должны задаваться именно здесь.

Храните учетные данные в конфигурации сервера, а не в параметрах инструментов и не в контексте модели. Разделяйте инструменты по уровню риска: инструменты только для чтения, такие как снимки экрана и извлечение текста, включены по умолчанию; инструменты записи, такие как клик, отправка и удаление, по умолчанию выключены и временно включаются под задачу. Даже если модель ошибется, масштаб последствий останется ограниченным.

Изолируйте среды по назначению. Разные аккаунты и разные задачи должны работать в разных средах; не смешивайте их в одной среде. В сценариях с несколькими аккаунтами создание среды, привязку сетевого выхода и поддержание состояния можно передать специализированному слою управления средами. Инструменты вроде PurpleMark как раз отделяют этот слой, а адаптер отвечает только за композицию бизнес-логики и проверки.

Сохраняйте журналы аудита. Должна быть возможность выгрузить по времени, какая задача использовала какую среду и какие инструменты записи вызывались. Если действительно произойдет проблема, только эта запись позволит восстановить ход событий.

И наконец, жесткая граница: слой адаптера не должен предоставлять способы обхода правил платформы. Массовую регистрацию, обход проверок или подделку личности нельзя упаковывать в инструменты даже под видом внутренних функций. Как только инструмент становится доступен модели, вызовы могут происходить автоматически; до этого нет надежной точки, где их можно остановить, а после нельзя отменить последствия.

Версии протокола и определения полей следует сверять с официальной документацией MCP.