Добавление адаптера платформы

Руководство о том, как подключить новый мессенджер к шлюзу Hermes. Адаптер платформы связывает Hermes с внешним сервисом сообщений (Telegram, Discord, WeCom и др.), чтобы пользователи общались с агентом прямо оттуда.

Архитектурная схема

User ↔ Messaging Platform ↔ Platform Adapter ↔ Gateway Runner ↔ AIAgent

Каждый адаптер наследует BasePlatformAdapter из gateway/platforms/base.py и реализует:

• connect() — установить соединение (WebSocket, long-poll, HTTP-сервер и т. п.) (абстрактный)
• disconnect() — корректное завершение работы (абстрактный)
• send() — отправить текстовое сообщение в чат (абстрактный)
• send_typing() — показать индикатор набора текста (опциональное переопределение)
• get_chat_info() — вернуть метаданные чата (опциональное переопределение)

Входящие сообщения адаптер принимает и передаёт через self.handle_message(event) — базовый класс маршрутизирует их в gateway runner.

Путь через плагин (рекомендуется)

Через систему плагинов адаптер платформы добавляется без правок основного кода Hermes. Плагин — это директория с двумя файлами:

~/.hermes/plugins/my-platform/
  plugin.yaml      # Метаданные плагина
  adapter.py       # Класс адаптера + точка входа register()

plugin.yaml

Метаданные плагина. Блоки requires_env и optional_env автоматически наполняют элементы интерфейса hermes config (см. Переменные окружения в hermes config ниже).

name: my-platform
label: My Platform
kind: platform
version: 1.0.0
description: My custom messaging platform adapter
author: Your Name
requires_env:
  - MY_PLATFORM_TOKEN          # bare string works
  - name: MY_PLATFORM_CHANNEL  # or rich dict for better UX
    description: "Channel to join"
    prompt: "Channel"
    password: false
optional_env:
  - name: MY_PLATFORM_HOME_CHANNEL
    description: "Default channel for cron delivery"
    password: false

Исходящие клиентские инструменты: provides_tools

Плагины kind: platform отложенные: модуль адаптера (и его SDK-импорты) загружается только тогда, когда шлюз, cron или путь send_message впервые запрашивает платформу у реестра платформ. Если ваш плагин также поставляет исходящие клиентские инструменты, которые агент должен уметь вызывать из любой сессии (как a2a_call / a2a_discover и другие во встроенном плагине a2a), положите их в отдельный tools.py с функцией register_tools(ctx) и объявите в манифесте:

provides_tools:
  - my_platform_call
  - my_platform_list

С объявленным provides_tools Hermes при обнаружении плагинов импортирует только tools.py и регистрирует клиентские инструменты в каждом процессе — включая CLI и TUI, — при этом адаптер остаётся отложенным. Держите __init__.py пакета лёгким по импортам и подтягивайте адаптер изнутри register(), чтобы жадный импорт оставался дешёвым. Без этого поля ничего не меняется: весь плагин остаётся отложенным.

Пользователи включают набор инструментов по платформе так же, как любой другой, например hermes tools enable my_platform --platform cli, либо перечислив ключ набора инструментов в platform_toolsets в config.yaml. Имена платформ плагинов — тоже допустимые значения для --platform, так что входящая сессия на вашей платформе может получить собственные исходящие инструменты.

adapter.py

from gateway.platforms._shared import extra_or_secret, get_scoped_secret, seed_extra_from_env
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
from gateway.config import Platform, PlatformConfig


class MyPlatformAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform("my_platform"))
        # Явный env (в границах профиля) → config.extra этого профиля (YAML) → умолчание. Под мультиплексором
        # промах в области профиля падает в его собственный YAML, но никогда — в env процесса другого профиля.
        self.token = extra_or_secret(config.extra, "token", "MY_PLATFORM_TOKEN")

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        # Подключиться к API платформы, запустить слушателей
        self._mark_connected()
        return True

    async def disconnect(self) -> None:
        self._mark_disconnected()

    async def send(self, chat_id, content, reply_to=None, metadata=None):
        # Отправить сообщение через API платформы
        return SendResult(success=True, message_id="...")

    async def get_chat_info(self, chat_id):
        return {"name": chat_id, "type": "dm"}


def check_requirements() -> bool:
    return bool(get_scoped_secret("MY_PLATFORM_TOKEN"))


def validate_config(config) -> bool:
    return bool(extra_or_secret(getattr(config, "extra", None), "token", "MY_PLATFORM_TOKEN"))


def _env_enablement() -> dict | None:
    if not get_scoped_secret("MY_PLATFORM_TOKEN", "").strip():
        return None
    return seed_extra_from_env(
        (("MY_PLATFORM_TOKEN", "token", None), ("MY_PLATFORM_CHANNEL", "channel", None)),
        home_env="MY_PLATFORM_HOME_CHANNEL",
    )


def register(ctx):
    """Точка входа плагина — вызывается системой плагинов Hermes."""
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=MyPlatformAdapter,
        # ПАССИВНАЯ проверка — «доступны ли зависимости/конфигурация прямо сейчас?».
        # Вызывается из статус-дисплеев и загрузки конфигурации, поэтому НИКОГДА
        # не должна делать pip install.
        check_fn=check_requirements,
        # АКТИВНЫЙ установщик (опционально) — только для платформ с
        # SDK, устанавливаемым лениво. create_adapter() вызывает его, когда
        # check_fn возвращает False, непосредственно перед тем как шлюз
        # подключит платформу. Обычно оборачивает
        # pm.extras.ensure_and_bind(...). Если опустить это поле,
        # False от check_fn становится жёсткой блокировкой.
        # ensure_deps_fn=ensure_requirements,
        validate_config=validate_config,
        required_env=["MY_PLATFORM_TOKEN"],
        install_hint="Declare my-platform-sdk in this plugin's Python dependencies, then retry hermes plugins enable my-platform",
        # Автоконфигурация через переменные окружения — заполняет PlatformConfig.extra
        # из env до создания адаптера. См. раздел «Автоконфигурация через env».
        env_enablement_fn=_env_enablement,
        # Поддержка доставки cron в домашний канал. Позволяет заданиям
        # deliver=my_platform маршрутизироваться без правок cron/scheduler.py.
        # См. раздел «Доставка cron».
        cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
        # Переменные авторизации пользователей для платформы
        allowed_users_env="MY_PLATFORM_ALLOWED_USERS",
        allow_all_env="MY_PLATFORM_ALLOW_ALL_USERS",
        # Ограничение длины сообщения для умного разбиения (0 = без ограничений)
        max_message_length=4000,
        # Подсказка для LLM, добавляемая в системный промпт
        platform_hint=(
            "You are chatting via My Platform. "
            "It supports markdown formatting."
        ),
        # Отображение
        emoji="💬",
    )

    # Опционально: зарегистрировать инструменты, специфичные для платформы
    ctx.register_tool(
        name="my_platform_search",
        toolset="my_platform",
        schema={...},
        handler=my_search_handler,
    )

Конфигурация

Пользователи настраивают платформу в config.yaml:

gateway:
  platforms:
    my_platform:
      enabled: true
      extra:
        token: "..."
        channel: "#general"

Либо через переменные окружения, которые адаптер читает в __init__.

Что система плагинов делает автоматически

При вызове ctx.register_platform() следующие точки интеграции отрабатывают сами — никаких изменений в основном коде:

Сервисные адаптеры: trusted_inbound, display_tier, shared_env_prefixes

У ctx.register_platform(...) / PlatformEntry есть три необязательных поля для адаптеров, которые связывают с Hermes сервис, а не чат-сеть. Образцовый потребитель — плагин Home Assistant.

import dataclasses

def register(ctx):
    kwargs = dict(
        name="my_service",
        label="My Service",
        adapter_factory=MyServiceAdapter,
        check_fn=check_requirements,
    )
    # Feature-detect: older cores reject unknown register_platform kwargs.
    from gateway.platform_registry import PlatformEntry
    known = {f.name for f in dataclasses.fields(PlatformEntry)}
    for key, value in (
        ("trusted_inbound", True),
        ("display_tier", "minimal"),
        ("shared_env_prefixes", ("MY_SERVICE_",)),
    ):
        if key in known:
            kwargs[key] = value
    ctx.register_platform(**kwargs)

Определяйте наличие этих полей через dataclasses.fields(PlatformEntry), как выше: тогда плагин останется загружаемым на старых ядрах Hermes, где этих полей ещё нет.

Расширения независимого пути отправки

Платформа-плагин может участвовать в исходящей доставке, инициированной хостом, — через прямой hermes send --to ... и через cron deliver=platform:..., — если объявит поведение отправки на той же PlatformEntry, которую создаёт ctx.register_platform(). send_message намеренно не является инструментом модели, вызываемым агентом; плагины не должны регистрировать эквивалентную поверхность модели, которая позволила бы агенту самостоятельно инициировать исходящие сообщения.

async def _send_request(args, chat_id, platform_name, pconfig):
    # `args` содержит поля запроса на отправку, инициированного хостом.
    message_id = await client.send(
        address=chat_id,
        body=args["message"],
        subject=args.get("subject"),
    )
    return {"success": True, "platform": platform_name,
            "chat_id": chat_id, "message_id": message_id}


def _parse_address(raw):
    normalized = raw.strip().lower()
    if normalized.startswith("@") and "@" in normalized[1:]:
        return normalized, None  # (chat_id, опциональный thread_id)
    return None                 # продолжить через разрешение по каталогу каналов


def _validate_address(address):
    # True — принять; False — отклонить; строка — отклонить с этим диагностическим сообщением.
    return True if address.endswith("@example.com") else "unsupported domain"


def register(ctx):
    ctx.register_platform(
        name="fmsg",
        label="Fixture Message",
        adapter_factory=lambda cfg: FmsgAdapter(cfg),
        check_fn=check_requirements,
        parse_target_ref_fn=_parse_address,
        validate_target_ref_fn=_validate_address,
        # Может быть обычной функцией или async def. Hermes дожидается любого
        # awaitable-результата, включая callable-объекты и обёртки functools.partial.
        send_message_handler=_send_request,
        # Предпочтительный низкоуровневый хук, когда cron должен отправлять из
        # процесса без живого шлюза.
        standalone_sender_fn=_standalone_send,
    )

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

Автоконфигурация через env

Чаще платформу настраивают переменными окружения в ~/.hermes/.env, а не правкой config.yaml. Хук env_enablement_fn даёт плагину подхватить эти переменные до создания адаптера, чтобы hermes gateway status, get_connected_platforms() и доставка cron видели актуальное состояние без инициализации SDK платформы.

Читайте env через gateway.platforms._shared.get_scoped_secret — никогда через os.getenv. Под gateway.multiplex_profiles env процесса содержит значения профиля ПО УМОЛЧАНИЮ; .env вторичного профиля существует только в его области секретов, и сырое чтение включило бы вашу платформу не для того профиля и не с теми учётными данными. seed_extra_from_env собирает seed-словарь по таблице (ENV_VAR, extra_key, conv) через этот же читатель.

from gateway.platforms._shared import get_scoped_secret, seed_extra_from_env


def _env_enablement() -> dict | None:
    """Заполнить PlatformConfig.extra из переменных окружения.

    Вызывается реестром платформ во время load_gateway_config().
    Вернуть None, если платформа не настроена минимально — тогда
    вызывающий код пропускает автовключение. Вернуть dict для заполнения extras.
    """
    token = get_scoped_secret("MY_PLATFORM_TOKEN", "").strip()
    channel = get_scoped_secret("MY_PLATFORM_CHANNEL", "").strip()
    if not (token and channel):
        return None
    # (ENV_VAR, extra_key, conv): пустые значения пропускаются, conv, бросивший
    # ValueError, отбрасывает ключ. home_env заполняет специальный dict 'home_channel'
    # ({chat_id, name} из MY_PLATFORM_HOME_CHANNEL[_NAME], name по умолчанию
    # "Home"), который хук ядра поднимает до HomeChannel.
    return {"token": token, "channel": channel, **seed_extra_from_env(
        (("MY_PLATFORM_PORT", "port", int), ("MY_PLATFORM_NICK", "nick", None)),
        home_env="MY_PLATFORM_HOME_CHANNEL",
    )}


def register(ctx):
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=MyPlatformAdapter,
        check_fn=check_requirements,
        validate_config=validate_config,
        env_enablement_fn=_env_enablement,
        # ... другие поля
    )

Мост конфигурации YAML→env

Кто-то предпочитает задавать ключи config.yaml (my_platform.require_mention, my_platform.allowed_channels и т. д.) вместо переменных окружения. Хук apply_yaml_config_fn отдаёт эту трансляцию самому плагину, чтобы основной gateway/config.py не разбирался в YAML-схеме вашей платформы.

from gateway.platforms._shared import apply_yaml_bridge

# (ключ yaml, ENV_VAR, kind). "lower" пробрасывает всякий раз, когда ключ есть
# (булевы значения становятся "true"/"false"); "str" пропускает null/пустые; "csv" склеивает
# списки через запятую; "json" сериализует значение.
_YAML_BRIDGE = (
    ("require_mention", "MY_PLATFORM_REQUIRE_MENTION", "lower"),
    ("allowed_channels", "MY_PLATFORM_ALLOWED_CHANNELS", "csv"),
)


def _apply_yaml_config(yaml_cfg: dict, platform_cfg: dict) -> dict | None:
    """Транслировать ключи `my_platform:` из config.yaml в env / extras.

    yaml_cfg     — полный разобранный dict верхнего уровня config.yaml
    platform_cfg — собственный sub-dict платформы (yaml_cfg.get("my_platform", {}))

    apply_yaml_bridge записывает каждую переменную env только если она не задана
    (env > YAML) и НИКОГДА — пока загружается конфигурация вторичного профиля под
    мультиплексором: запись там закрепила бы политику этого профиля на весь процесс.
    Те же значения она возвращает для PlatformConfig.extra, так что в адаптере
    сначала читайте `extra`, а потом откатывайтесь к переменной env
    (`_shared.extra_or_secret`).
    """
    return apply_yaml_bridge(platform_cfg, _YAML_BRIDGE)

def register(ctx):
    ctx.register_platform(
        name="my_platform",
        ...,
        apply_yaml_config_fn=_apply_yaml_config,
    )

Хук вызывается во время load_gateway_config() после общего цикла по общим ключам (он обрабатывает unauthorized_dm_behavior, notice_delivery, reply_prefix, require_mention и т. д.) и до _apply_env_overrides() — так что плагину остаётся транслировать только платформоспецифичные ключи.

Исключения из хука перехватываются и пишутся в лог на уровне debug — некорректный плагин никогда не прервёт загрузку конфигурации шлюза.

Доставка cron

Чтобы задания cron с deliver=my_platform уходили в настроенный домашний канал, укажите в cron_deliver_env_var имя переменной окружения с ID чата/комнаты/канала по умолчанию:

ctx.register_platform(
    name="my_platform",
    ...
    cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
)

Планировщик читает эту переменную, когда определяет домашнюю цель для заданий deliver=my_platform, и считает платформу допустимой целью cron наравне с _KNOWN_DELIVERY_PLATFORMS. Если env_enablement_fn заполняет dict home_channel (см. выше), приоритет у него — cron_deliver_env_var остаётся запасным вариантом для cron-заданий, которые запускаются до заполнения env.

Доставка cron из отдельного процесса

cron_deliver_env_var делает платформу распознаваемой целью deliver=. Чтобы сама отправка прошла, когда cron-задание выполняется в отдельном процессе от шлюза (то есть hermes cron run отдельно от hermes gateway), зарегистрируйте standalone_sender_fn:

async def _standalone_send(
    pconfig,
    chat_id,
    message,
    *,
    thread_id=None,
    media_files=None,
    force_document=False,
):
    """Открыть эфемерное соединение / получить свежий токен, отправить и закрыть."""
    # ... открыть соединение, отправить сообщение, вернуть результат ...
    return {"success": True, "message_id": "..."}
    # или {"error": "..."}

ctx.register_platform(
    name="my_platform",
    ...
    cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
    standalone_sender_fn=_standalone_send,
)

Зачем нужен этот хук: встроенные платформы (Telegram, Discord, Slack и др.) поставляют прямые REST-хелперы в tools/send_message_tool.py, и cron доставляет сообщения, не удерживая шлюз в том же процессе. Платформы-плагины исторически зависели от _gateway_runner_ref(), который вне процесса шлюза возвращает None, — без standalone_sender_fn отправка на стороне cron падает с ошибкой No live adapter for platform '<name>'.

Функция получает те же pconfig и chat_id, что и живой адаптер, плюс опциональные именованные аргументы thread_id, media_files и force_document. Возврат {"success": True, "message_id": ...} считается успешной доставкой; возврат {"error": "..."} попадает в delivery_errors cron-задания. Исключения внутри функции перехватывает диспетчер и выводит как Plugin standalone send failed: <reason>. Эталонные реализации лежат в plugins/platforms/{irc,teams,google_chat}/adapter.py.

Переменные окружения в hermes config

hermes_cli/config.py сканирует plugins/platforms/*/plugin.yaml при импорте и сам наполняет OPTIONAL_ENV_VARS из блоков requires_env и (опционально) optional_env. Используйте расширенный формат dict, чтобы задать описания, подсказки, флаги пароля и URL — интерфейс CLI подхватит их без лишней возни.

# plugins/platforms/my_platform/plugin.yaml
name: my_platform-platform
label: My Platform
kind: platform
version: 1.0.0
description: >
  My Platform gateway adapter for Hermes Agent.
author: Your Name
requires_env:
  - name: MY_PLATFORM_TOKEN
    description: "Bot API token from the My Platform console"
    prompt: "My Platform bot token"
    url: "https://my-platform.example.com/bots"
    password: true
  - name: MY_PLATFORM_CHANNEL
    description: "Channel to join (e.g. #hermes)"
    prompt: "Channel"
    password: false
optional_env:
  - name: MY_PLATFORM_HOME_CHANNEL
    description: "Default channel for cron delivery (defaults to MY_PLATFORM_CHANNEL)"
    prompt: "Home channel (or empty)"
    password: false
  - name: MY_PLATFORM_ALLOWED_USERS
    description: "Comma-separated user IDs allowed to talk to the bot"
    prompt: "Allowed users (comma-separated)"
    password: false

Поддерживаемые ключи dict: name (обязательный), description, prompt, url, password (bool; если не указан, определяется по суффиксам *_TOKEN / *_SECRET / *_KEY / *_PASSWORD / *_JSON), category (по умолчанию "messaging").

Записи в формате строки (- MY_PLATFORM_TOKEN) тоже работают — для них генерируется общее описание на основе label плагина. Если запись для той же переменной уже жёстко прописана в OPTIONAL_ENV_VARS, приоритет у неё (ради обратной совместимости); форма plugin.yaml служит запасным вариантом.

UX при медленном LLM на разных платформах

У некоторых платформ есть ограничения, которые меняют способ подачи медленного ответа LLM:

• LINE выдаёт одноразовый reply token, который истекает примерно через 60 секунд после входящего события. Ответ с этим токеном бесплатен, фолбэк на платный Push API — нет. Если LLM не успел к дедлайну, выбор такой: «сжечь платную квоту Push» или «сделать что-то поумнее с reply token, пока он не истёк».
• WhatsApp закрывает сессию после 24 часов бездействия, дальше принимаются только шаблонные сообщения.
• SMS не знает ни индикаторов набора текста, ни поэтапных обновлений — длинный ответ выглядит так, будто бот ушёл в офлайн.

Это реальные ограничения, которые базовый BasePlatformAdapter предусмотреть не может. Интерфейс плагинов намеренно оставляет место, чтобы надстроить платформоспецифичный UX поверх базового цикла набора текста, не раздувая список аргументов.

Паттерн: переопределение _keep_typing для промежуточного UX

BasePlatformAdapter._keep_typing — это «сердцебиение» индикатора набора текста: фоновая задача, которая работает, пока LLM генерирует ответ, и отменяется при его доставке. Чтобы добавить платформоспецифичное поведение по порогу времени (например, отправить «думаю…» через 45 секунд), переопределите _keep_typing в адаптере, запустите собственную задачу рядом с super()._keep_typing() и завершите её в блоке finally:

class LineAdapter(BasePlatformAdapter):
    async def _keep_typing(self, chat_id: str, *args, **kwargs) -> None:
        if self.slow_response_threshold <= 0:
            await super()._keep_typing(chat_id, *args, **kwargs)
            return

        async def _fire_at_threshold() -> None:
            try:
                await asyncio.sleep(self.slow_response_threshold)
            except asyncio.CancelledError:
                raise
            # Платформоспецифичная работа — для LINE отправляем Template
            # Buttons «Get answer» с закешированным reply token, чтобы
            # пользователь мог получить кешированный ответ позже через
            # свежий (бесплатный) reply token из postback callback.
            await self._send_slow_response_button(chat_id)

        side_task = asyncio.create_task(_fire_at_threshold())
        try:
            await super()._keep_typing(chat_id, *args, **kwargs)
        finally:
            if not side_task.done():
                side_task.cancel()
                try:
                    await side_task
                except (asyncio.CancelledError, Exception):
                    pass

Ключевые моменты:

• Всегда вызывайте await super()._keep_typing(...). «Сердцебиение» набора текста полезно само по себе — не заменяйте его, а надстраивайте поверх.
• Завершайте побочную задачу в finally. Когда LLM заканчивает работу (или /stop отменяет выполнение), шлюз отменяет задачу набора текста. Побочная задача тоже должна заметить отмену — иначе зависнет и может сработать после того, как ответ уже доставлен.
• Используйте совместно с interrupt_session_activity, чтобы по команде /stop разрешить orphan-состояния UX. Для LINE это перевод записи в postback-кеше из PENDING в ERROR, чтобы постоянная кнопка «Get answer» выдала «Run was interrupted» вместо зацикливания.

Паттерн: переопределение send для маршрутизации через кеш

Если ваш UX при медленном ответе кеширует результат для последующего получения (postback-поток LINE), переопределённый send должен распознавать три режима:

1. Активный postback pending для этого чата → закешировать ответ по request_id, ничего видимого не отправлять.
2. Системное подтверждение занятости (⚡ Interrupting, ⏳ Queued, ⏩ Steered) → обойти кеш и отправить видимо, чтобы пользователь увидел реакцию шлюза.
3. Обычный ответ → отправить через reply-token-или-push как обычно.

async def send(self, chat_id: str, content: str, **kw) -> SendResult:
    if _is_system_bypass(content):
        return await self._send_text_chunks(chat_id, content, force_push=False)
    pending_rid = self._pending_buttons.get(chat_id)
    if pending_rid:
        self._cache.set_ready(pending_rid, content)
        return SendResult(success=True, message_id=pending_rid)
    return await self._send_text_chunks(chat_id, content, force_push=False)

_SYSTEM_BYPASS_PREFIXES — это собственные префиксы подтверждения занятости шлюза (⚡, ⏳, ⏩, 💾). Их всегда нужно пропускать видимо, в каком бы состоянии ни был кешированный UX.

Когда применять этот паттерн

Используйте переопределение цикла набора текста, если:

• Исходящий API платформы имеет жёсткое временно́е ограничение (одноразовый reply token, истекающая sticky-сессия и т. д.) И
• Видимый промежуточный пузырь — приемлемый UX для этой платформы.

Используйте более простой путь slow_response_threshold = 0 (всегда Push), если:

• Платформа не различает бесплатные и платные отправки, ИЛИ
• Пользователи предпочитают «загрузка… загрузка… ГОТОВО» без промежуточного интерактивного пузыря.

LINE поддерживает оба варианта: по умолчанию порог 45 с для бесплатного postback-получения, а LINE_SLOW_RESPONSE_THRESHOLD=0 возвращает режим «всегда Push».

Эталонная реализация

Полная реализация postback для LINE — в plugins/platforms/line/adapter.py: конечный автомат RequestCache (PENDING → READY → DELIVERED плюс ERROR для /stop), переопределение _keep_typing с Template Buttons по порогу, переопределение send с маршрутизацией через кеш и переопределение interrupt_session_activity для разрешения orphan-записей PENDING.

Эталонные реализации (путь через плагин)

Полный рабочий пример — plugins/platforms/irc/ в репозитории: полноценный async IRC-адаптер без внешних зависимостей. plugins/platforms/teams/ охватывает Bot Framework / Adaptive Cards, plugins/platforms/google_chat/ — OAuth-based REST API, plugins/platforms/line/ — webhook-driven Messaging API с платформоспецифичным UX для медленного LLM.

---

Пошаговый чеклист (встроенный путь)

1. Enum платформы

Добавьте платформу в enum Platform в gateway/config.py:

class Platform(Enum):
    # ... существующие платформы ...
    NEWPLAT = "newplat"

2. Файл адаптера

Создайте plugins/platforms/newplat/adapter.py:

from gateway.config import Platform, PlatformConfig
from gateway.platforms._shared import extra_or_secret
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType

def check_newplat_requirements() -> bool:
    """Вернуть True, если зависимости доступны."""
    return SOME_SDK_AVAILABLE

class NewPlatAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform.NEWPLAT)
        # Читать конфигурацию из dict config.extra
        extra = config.extra or {}
        self._api_key = extra_or_secret(extra, "api_key", "NEWPLAT_API_KEY")

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        # Установить соединение, запустить polling/webhook
        self._mark_connected()
        return True

    async def disconnect(self) -> None:
        self._running = False
        self._mark_disconnected()

    async def send(self, chat_id, content, reply_to=None, metadata=None):
        # Отправить сообщение через API платформы
        return SendResult(success=True, message_id="...")

    async def get_chat_info(self, chat_id):
        return {"name": chat_id, "type": "dm"}

Для входящих сообщений создайте MessageEvent и вызовите self.handle_message(event):

source = self.build_source(
    chat_id=chat_id,
    chat_name=name,
    chat_type="dm",  # или "group"
    user_id=user_id,
    user_name=user_name,
)
event = MessageEvent(
    text=content,
    message_type=MessageType.TEXT,
    source=source,
    message_id=msg_id,
)
await self.handle_message(event)

3. Конфигурация шлюза (gateway/config.py)

Три точки изменений:

1. get_connected_platforms() — Добавить проверку необходимых учётных данных платформы
2. load_gateway_config() — Добавить запись в map токен/env: Platform.NEWPLAT: "NEWPLAT_TOKEN"
3. _apply_env_overrides() — Сопоставить все переменные NEWPLAT_* с конфигурацией

4. Gateway runner (gateway/run.py и соседние gateway/run_*.py)

Шесть точек изменений:

1. Таблица _BUILTIN_ADAPTERS (gateway/run.py) — Добавить запись Platform.NEWPLAT: (module, class, check_fn, error_msg); функция _instantiate_adapter() (gateway/run_adapters.py) сначала обращается к реестру плагинов, затем к этой таблице — расширять цепочку elif больше не нужно. Обёртка _create_adapter() привязывает каждый успешно созданный адаптер к его gateway runner.
2. _is_user_authorized() map allowed_users — Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
3. _is_user_authorized() map allow_all — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
4. Проверка политики доступа при старте (gateway/run_startup.py) — Добавить "NEWPLAT" в _ALLOWLIST_ENV_PLATFORMS (отсюда выводятся и NEWPLAT_ALLOWED_USERS, и NEWPLAT_ALLOW_ALL_USERS)
5. _BUILTIN_ALLOW_ALL_VARS при старте (gateway/run_startup.py) — выводится из того же кортежа _ALLOWLIST_ENV_PLATFORMS, добавлять ничего не нужно
6. _UPDATE_ALLOWED_PLATFORMS frozenset — Добавить Platform.NEWPLAT

5. Кросс-платформенная доставка

1. gateway/platforms/webhook.py — Добавить "newplat" в кортеж типов доставки
2. cron/scheduler_delivery.py — Добавить в frozenset _KNOWN_DELIVERY_PLATFORMS и в map платформ _deliver_result()

6. Интеграция с CLI

1. hermes_cli/config.py — Добавить все переменные NEWPLAT_* в _EXTRA_ENV_KEYS
2. hermes_cli/gateway.py — Добавить запись в список _PLATFORMS с key, label, emoji, token_var, setup_instructions и vars
3. hermes_cli/platforms.py — Добавить запись PlatformInfo с label и default_toolset (используется TUI skills_config и tools_config)
4. hermes_cli/setup.py — Добавить функцию _setup_newplat() (можно делегировать в gateway.py) и добавить кортеж в список платформ мессенджеров
5. hermes_cli/status.py — Добавить запись обнаружения платформы: "NewPlat": ("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL")
6. hermes_cli/dump.py — Добавить "newplat": "NEWPLAT_TOKEN" в dict обнаружения платформ

7. Инструменты

1. tools/send_message_tool.py — Добавить "newplat": Platform.NEWPLAT в map платформ
2. tools/cronjob_tools.py — Добавить newplat в строку описания целей доставки

8. Наборы инструментов

1. toolsets.py — Добавить определение набора инструментов "hermes-newplat" с _HERMES_CORE_TOOLS
2. toolsets.py — Добавить "hermes-newplat" в список includes "hermes-gateway"

9. Опционально: подсказки платформы

agent/prompt_builder.py — Если у платформы есть особые ограничения рендеринга (нет markdown, лимиты длины сообщений и т. д.), добавьте запись в dict PLATFORM_HINTS. Так в системный промпт попадут платформоспецифичные инструкции:

PLATFORM_HINTS = {
    # ...
    "newplat": (
        "You are chatting via NewPlat. It supports markdown formatting "
        "but has a 4000-character message limit."
    ),
}

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

10. Тесты

Создайте tests/gateway/test_newplat.py, охватывающий:

• Создание адаптера из конфигурации
• Построение события сообщения
• Метод send (с моком внешнего API)
• Платформоспецифичные функции (шифрование, маршрутизация и т. д.)

11. Документация

Аудит паритета

Перед тем как пометить PR с новой платформой завершённым, проведите аудит паритета относительно зрелой платформы:

# Найти все .py-файлы с упоминанием эталонной платформы
search_files "bluebubbles" output_mode="files_only" file_glob="*.py"

# Найти все .py-файлы с упоминанием новой платформы
search_files "newplat" output_mode="files_only" file_glob="*.py"

# Любой файл из первого набора, отсутствующий во втором — потенциальный пробел

Повторите для файлов .md и .ts. Разберите каждый пробел — это перечисление платформы (нужно обновить) или платформоспецифичная ссылка (пропустить)?

Типовые паттерны

Long-poll-адаптеры

Если адаптер работает на long-polling (как Telegram или Weixin), задействуйте задачу polling-цикла:

async def connect(self):
    self._poll_task = asyncio.create_task(self._poll_loop())
    self._mark_connected()

async def _poll_loop(self):
    while self._running:
        messages = await self._fetch_updates()
        for msg in messages:
            await self.handle_message(self._build_event(msg))

Callback/Webhook-адаптеры

Если платформа push-доставляет сообщения на ваш эндпоинт (как WeCom Callback), запустите HTTP-сервер:

async def connect(self):
    self._app = web.Application()
    self._app.router.add_post("/callback", self._handle_callback)
    # ... запустить aiohttp-сервер
    self._mark_connected()

async def _handle_callback(self, request):
    event = self._build_event(await request.text())
    await self._message_queue.put(event)
    return web.Response(text="success")  # Немедленно подтвердить получение

Для платформ с жёсткими дедлайнами ответа (например, 5-секундный лимит WeCom) всегда подтверждайте получение сразу, а ответ агента доставляйте проактивно через API позже. Сессии агента длятся от 3 до 30 минут — вписать ответ в окно callback-запроса нереально.

Дедупликация входящих сообщений

Платформы доставляют сообщения повторно: при возобновлении websocket-соединения проигрываются недавние события, вебхуки ретраятся, неподтверждённая пачка поллинга приходит снова. Отбрасывайте повторы общим хелпером по ID сообщения на платформе:

from gateway.platforms.helpers import MessageDeduplicator

self._dedup = MessageDeduplicator(ttl_seconds=600)  # в __init__

if self._dedup.is_duplicate(msg_id):  # во входящем обработчике
    return

Когда наблюдатель переподключений шлюза заменяет упавший адаптер новым экземпляром, он копирует живые ID из каждого атрибута типа MessageDeduplicator старого экземпляра в новый. Поэтому повтор сразу после переподключения всё равно отбрасывается. Кэш в любой другой структуре (обычный dict или set) на новом экземпляре начинается пустым.

Блокировки токенов

Если адаптер удерживает постоянное соединение с уникальными учётными данными, добавьте scoped-блокировку, чтобы два профиля не использовали один и тот же токен:

from gateway.status import acquire_scoped_lock, release_scoped_lock

async def connect(self, *, is_reconnect: bool = False):
    acquired, _existing = acquire_scoped_lock("newplat", self._token)
    if not acquired:
        logger.error("Token already in use by another profile")
        return False
    # ... подключиться

async def disconnect(self):
    release_scoped_lock("newplat", self._token)

Эталонные реализации