添加平台适配器

本指南介绍了向 Hermes 网关添加新的消息传递平台。平台适配器将 Hermes 连接到外部消息服务（Telegram、Discord、WeCom 等），以便用户可以通过该服务与代理进行交互。

:::提示添加平台适配器涉及代码、配置和文档中的 20 多个文件。使用本指南作为清单 - 适配器文件本身通常只占工作的 40%。 :::

架构概述

User ↔ Messaging Platform ↔ Platform Adapter ↔ 网关 Runner ↔ AIAgent

每个适配器都从 gateway/platforms/base.py 扩展 BasePlatformAdapter 并实现：

connect() — 建立连接（WebSocket、长轮询、HTTP 服务器等）
disconnect() — 干净关闭
send() — 发送短信到聊天室
send_typing() — 显示键入指示符（可选）
get_chat_info() — 返回聊天元数据

入站消息由适配器接收并通过 self.handle_message(event) 转发，基类将其路由到网关运行程序。

分步清单

1. 平台枚举

将您的平台添加到 gateway/config.py 中的 Platform 枚举中：

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

三个接触点：

get_connected_platforms() — 添加对平台所需凭据的检查
load_gateway_config() — 添加令牌环境映射条目：Platform.NEWPLAT: "NEWPLAT_TOKEN"
_apply_env_overrides() — 将所有 NEWPLAT_* 环境变量映射到配置

4. 网关运行器 (`gateway/run.py`)

五个接触点：

_create_adapter() — 添加 elif platform == Platform.NEWPLAT: 分支
_is_user_authorized() allowed_users 映射 — Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
_is_user_authorized() allowed_all 映射 — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
早期环境检查 _any_allowlist 元组 — 添加 "NEWPLAT_ALLOWED_USERS"
早期环境检查 _allow_all 元组 — 添加 "NEWPLAT_ALLOW_ALL_USERS"
_UPDATE_ALLOWED_PLATFORMS freezeset — 添加 Platform.NEWPLAT

5. 跨平台交付

gateway/platforms/webhook.py — 将 "newplat" 添加到传递类型元组
cron/scheduler.py — 添加到 _KNOWN_DELIVERY_PLATFORMS freezeset 和 _deliver_result() 平台地图

6. CLI 集成

hermes_cli/config.py — 将所有 NEWPLAT_* 变量添加到 _EXTRA_ENV_KEYS
hermes_cli/gateway.py — 使用键、标签、表情符号、token_var、setup_instructions 和 var 将条目添加到 _PLATFORMS 列表
hermes_cli/platforms.py — 添加带标签和 default_toolset 的 PlatformInfo 条目（由 skills_config 和 tools_config TUI 使用）
hermes_cli/setup.py — 添加 _setup_newplat() 函数（可以委托给 gateway.py）并将元组添加到消息传递平台列表
hermes_cli/status.py — 添加平台检测条目："NewPlat": ("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL")
hermes_cli/dump.py — 将 "newplat": "NEWPLAT_TOKEN" 添加到平台检测字典中

7. 工具

tools/send_message_tool.py — 将 "newplat": Platform.NEWPLAT 添加到平台地图
tools/cronjob_tools.py — 将 newplat 添加到传递目标描述字符串

8. 工具集

toolsets.py — 使用 _HERMES_CORE_TOOLS 添加 "hermes-newplat" 工具集定义
toolsets.py — 将 "hermes-newplat" 添加到 "hermes-gateway" 包含列表

9. 可选：平台提示

agent/prompt_builder.py — 如果您的平台有特定的渲染限制（无 Markdown、消息长度限制等），请向 _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. 文档

文件	添加什么
`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`	`hermes-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))

回调/Webhook 适配器

如果平台将消息推送到您的端点（例如 WeCom Callback），请运行 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

对于响应期限紧迫的平台（例如，WeCom 的 5 秒限制），请务必立即确认并稍后通过 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`	`bluebubbles.py`休息 + webhook	中等	简单的 REST API 集成
`weixin.py`	`weixin.py`长轮询 + CDN	高	媒体处理、加密
`wecom_callback.py`	`wecom_callback.py`回调/webhook	中等	HTTP 服务器、AES 加密、多应用
`telegram.py`	`telegram.py`长轮询 + 机器人 API	高	带组、线程的全功能适配器

架构概述​

分步清单​

1. 平台枚举​

2.适配器文件​

3.网关配置(gateway/config.py)​

4. 网关运行器 (gateway/run.py)​

5. 跨平台交付​

6. CLI 集成​

7. 工具​

8. 工具集​

9. 可选：平台提示​

10. 测试​

11. 文档​

奇偶校验​

常见模式​

长轮询适配器​

回调/Webhook 适配器​

令牌锁​

参考实现​