添加平臺適配器

本指南介紹如何向 Hermes 網關添加新的消息傳遞平臺。平臺適配器將 Hermes 連接到外部消息服務（Telegram、Discord、企業微信等），以便用戶可以通過該服務與代理進行交互。

提示

添加平臺適配器會涉及代碼、配置和文檔中 20 多個文件。請將本指南用作檢查清單——適配器文件本身通常只佔工作量的 40%。

架構概述

User ↔ Messaging Platform ↔ Platform Adapter ↔ Gateway 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() allow_all 映射 — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
早期環境變量檢查 _any_allowlist 元組 — 添加 "NEWPLAT_ALLOWED_USERS"
早期環境變量檢查 _allow_all 元組 — 添加 "NEWPLAT_ALLOW_ALL_USERS"
_UPDATE_ALLOWED_PLATFORMS 凍結集合 (frozenset) — 添加 Platform.NEWPLAT

5. 跨平臺交付

gateway/platforms/webhook.py — 將 "newplat" 添加到交付類型元組中
cron/scheduler.py — 添加到 _KNOWN_DELIVERY_PLATFORMS 凍結集合和 _deliver_result() 平臺映射中

6. CLI 集成

hermes_cli/config.py — 將所有 NEWPLAT_* 變量添加到 _EXTRA_ENV_KEYS
hermes_cli/gateway.py — 在 _PLATFORMS 列表中添加條目，包含鍵、標籤、emoji、token_var、setup_instructions 和 vars
hermes_cli/platforms.py — 添加帶有 label 和 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，涵蓋：

從配置構建適配器
構建消息事件
發送方法（模擬外部 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 或微信），請使用輪詢循環任務：

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 適配器

如果平臺將消息推送到您的端點（如企業微信回調），請運行 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 秒限制），請務必立即確認，並稍後通過 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 + webhook	中等	簡單的 REST API 集成
`weixin.py`	長輪詢 + CDN	高	媒體處理、加密
`wecom_callback.py`	回調/webhook	中等	HTTP 服務器、AES 加密、多應用
`telegram.py`	長輪詢 + Bot API	高	支持群組和線程的全功能適配器

架構概述​

逐步檢查清單​

1. 平臺枚舉​

2. 適配器文件​

3. 網關配置 (gateway/config.py)​

4. 網關運行器 (gateway/run.py)​

5. 跨平臺交付​

6. CLI 集成​

7. 工具​

8. 工具集​

9. 可選：平臺提示​

10. 測試​

11. 文檔​

一致性審計​

常見模式​

長輪詢適配器​

回調/Webhook 適配器​

令牌鎖​

參考實現​