🏠 Главная › Developer Guide › adding platform adapters

sidebar_position: 9 lang: ru

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

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

:::совет Есть два способа добавить платформу: - Плагин (рекомендуется для сообщества/сторонних разработчиков): перетащите каталог плагина в ~/.hermes/plugins/ — никаких изменений основного кода не требуется. См. [Путь к плагину] (#plugin-path-recommended) ниже. - Встроенный: изменяйте более 20 файлов в коде, конфигурации и документации. Используйте Встроенный контрольный список ниже.

Обзор архитектуры

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

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

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

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

Система плагинов позволяет добавлять адаптер платформы без изменения какого-либо основного кода Hermes. Ваш плагин представляет собой каталог с двумя файлами:

~/.hermes/plugins/my-platform/
  PLUGIN.yaml      # Plugin metadata
  adapter.py       # Adapter class + register() entry point

ПЛАГИН.yaml

name: my-platform
version: 1.0.0
description: My custom messaging platform adapter
requires_env:
  - MY_PLATFORM_TOKEN
  - MY_PLATFORM_CHANNEL

адаптер.py

import os
from gateway.platforms.base import (
    BasePlatformAdapter, SendResult, MessageEvent, MessageType,
)
from gateway.config import Platform, PlatformConfig


class MyPlatformAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform("my_platform"))
        extra = config.extra or {}
        self.token = os.getenv("MY_PLATFORM_TOKEN") or extra.get("token", "")

    async def connect(self) -> bool:
        # Connect to the platform API, start listeners
        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):
        # Send message via platform 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(os.getenv("MY_PLATFORM_TOKEN"))


def validate_config(config) -> bool:
    extra = getattr(config, "extra", {}) or {}
    return bool(os.getenv("MY_PLATFORM_TOKEN") or extra.get("token"))


def register(ctx):
    """Plugin entry point — called by the Hermes plugin system."""
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
        check_fn=check_requirements,
        validate_config=validate_config,
        required_env=["MY_PLATFORM_TOKEN"],
        install_hint="pip install my-platform-sdk",
        # Per-platform user authorization env vars
        allowed_users_env="MY_PLATFORM_ALLOWED_USERS",
        allow_all_env="MY_PLATFORM_ALLOW_ALL_USERS",
        # Message length limit for smart chunking (0 = no limit)
        max_message_length=4000,
        # LLM guidance injected into system prompt
        platform_hint=(
            "You are chatting via My Platform. "
            "It supports markdown formatting."
        ),
        # Display
        emoji="💬",
    )

    # Optional: register platform-specific tools
    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(), за вас обрабатываются следующие точки интеграции — никаких изменений основного кода не требуется:

Точка интеграции Как это работает
Создание адаптера шлюза Реестр проверяется перед встроенной цепочкой if/elif
Разбор конфига Platform._missing_() принимает любое имя платформы
Проверка подключенной платформы Реестр validate_config() вызван
Авторизация пользователя allowed_users_env / allow_all_env проверено
Доставка Крон Platform() разрешает любое зарегистрированное имя
инструмент send_message Маршруты через адаптер живого шлюза
Кроссплатформенная доставка Webhook Реестр проверен на наличие известных платформ
/update командный доступ allow_update_command флаг
Каталог каналов Платформы плагинов, включенные в список
Подсказки системы platform_hint в контексте LLM
Разбиение сообщений max_message_length для умного разделения
Редактирование личных данных pii_safe флаг
hermes status Показывает платформы плагинов с тегом (plugin)
hermes gateway setup Платформы плагинов появляются в меню настроек
hermes tools / hermes skills Платформы плагинов в конфигурации каждой платформы
Токен-блокировка (многопрофильная) Используйте acquire_scoped_lock() в своем connect()
Предупреждение об потерянной конфигурации Описательный журнал при отсутствии плагина

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

Полный рабочий пример см. в plugins/platforms/irc/ в репозитории — полностью асинхронный IRC-адаптер с нулевыми внешними зависимостями.

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

:::примечание Этот контрольный список предназначен для добавления платформы непосредственно в основную кодовую базу Hermes — обычно это делается основными участниками официально поддерживаемых платформ. Платформы сообщества и сторонние платформы должны использовать указанный выше Путь к плагину.

1. Перечисление платформ

Добавьте свою платформу в перечисление Platform в gateway/config.py:

class Platform(str, Enum):
    # ... existing platforms ...
    NEWPLAT = "newplat"

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

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

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
    BasePlatformAdapter, MessageEvent, MessageType, SendResult,
)

def check_newplat_requirements() -> bool:
    """Return True if dependencies are available."""
    return SOME_SDK_AVAILABLE

class NewPlatAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform.NEWPLAT)
        # Read config from config.extra dict
        extra = config.extra or {}
        self._api_key = extra.get("api_key") or os.getenv("NEWPLAT_API_KEY", "")

    async def connect(self) -> bool:
        # Set up connection, start 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):
        # Send message via platform 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",  # or "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() — Добавьте запись карты окружения токена: Platform.NEWPLAT: "NEWPLAT_TOKEN"
  3. _apply_env_overrides() — Сопоставьте все переменные окружения NEWPLAT_* с конфигурацией.

4. Бегун шлюза (gateway/run.py)

Пять точек соприкосновения:

  1. _create_adapter() — Добавьте ветку elif platform == Platform.NEWPLAT:.
  2. _is_user_authorized() карта разрешенных_пользователей — Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
  3. _is_user_authorized() картаallow_all — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
  4. Ранняя проверка кортежа _any_allowlist — Добавьте "NEWPLAT_ALLOWED_USERS"
  5. Ранняя проверка кортежа _allow_all — Добавьте "NEWPLAT_ALLOW_ALL_USERS"
  6. _UPDATE_ALLOWED_PLATFORMS замороженный — Добавьте Platform.NEWPLAT

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

  1. gateway/platforms/webhook.py — добавьте "newplat" в кортеж типа доставки.
  2. cron/scheduler.py — Добавить в _KNOWN_DELIVERY_PLATFORMS замороженный набор и _deliver_result() карту платформы.

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

  1. hermes_cli/config.py — добавьте все переменные NEWPLAT_* в _EXTRA_ENV_KEYS.
  2. hermes_cli/gateway.py — добавьте запись в список _PLATFORMS с ключом, меткой, эмодзи, token_var, setup_instructions и переменными.
  3. hermes_cli/platforms.py — добавьте запись PlatformInfo с меткой и default_toolset (используется skills_config и tools_config TUI)
  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" в словарь обнаружения платформы.

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

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

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

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

9. Необязательно: подсказки по платформе

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

_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, охватывающий:

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

Файл Что добавить
website/docs/user-guide/messaging/newplat.md Полная страница настройки платформы
website/docs/user-guide/messaging/index.md Таблица сравнения платформ, схема архитектуры, таблица наборов инструментов, раздел безопасности, ссылка на следующие шаги
website/docs/reference/environment-variables.md Все переменные среды NEWPLAT_*
website/docs/reference/toolsets-reference.md набор инструментов гермес-newplat
website/docs/integrations/index.md Ссылка на платформу
website/sidebars.ts Запись на боковой панели страницы документации
website/docs/developer-guide/architecture.md Количество адаптеров + список
website/docs/developer-guide/gateway-internals.md Список файлов адаптера

Аудит четности

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

# Find every .py file mentioning the reference platform
search_files "bluebubbles" output_mode="files_only" file_glob="*.py"

# Find every .py file mentioning the new platform
search_files "newplat" output_mode="files_only" file_glob="*.py"

# Any file in the first set but not the second is a potential gap

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

Общие шаблоны

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

Если ваш адаптер использует длинный опрос (например, Telegram или Weixin), используйте задачу цикла опроса:

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))

Адаптеры обратного вызова/веб-перехватчика

Если платформа отправляет сообщения на вашу конечную точку (например, обратный вызов WeCom), запустите HTTP-сервер:

async def connect(self):
    self._app = web.Application()
    self._app.router.add_post("/callback", self._handle_callback)
    # ... start aiohttp server
    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")  # Acknowledge immediately

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

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

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

from gateway.status import acquire_scoped_lock, release_scoped_lock

async def connect(self):
    if not acquire_scoped_lock("newplat", self._token):
        logger.error("Token already in use by another profile")
        return False
    # ... connect

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

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

Адаптер Узор Сложность Хорошая ссылка для
bluebubbles.py ОТДЫХ + вебхук Средний Простая интеграция REST API
weixin.py Длинный опрос + CDN Высокий Обработка мультимедиа, шифрование
wecom_callback.py Обратный вызов/вебхук Средний HTTP-сервер, шифрование AES, несколько приложений
telegram.py Длинный опрос + API бота Высокий Полнофункциональный адаптер с группами, резьбами