添加平台适配器

hermes agent 添加平台适配器

本指南涵盖如何向 Hermes gateway 添加新的 messaging platform。platform adapter 将 Hermes 连接到外部消息服务（Telegram、Discord、WeCom 等），这样用户就可以通过该服务与 agent 交互。

架构概览

用户 ↔ Messaging Platform ↔ Platform Adapter ↔ Gateway Runner ↔ AIAgent

每个 adapter 都扩展自 gateway/platforms/base.py 中的 BasePlatformAdapter，并实现：

connect() —— 建立连接（WebSocket、long-poll、HTTP server 等）（抽象）
disconnect() —— 干净关闭（抽象）
send() —— 向 chat 发送文本消息（抽象）
send_typing() —— 显示正在输入指示器（可选 override）
get_chat_info() —— 返回 chat metadata（可选 override）

入站消息由 adapter 接收，并通过 self.handle_message(event) 转发，base class 会将其路由到 gateway runner。

Plugin Path（推荐）

插件系统允许你添加一个 platform adapter，而不需要修改任何 Hermes 核心代码。你的 plugin 是一个包含两个文件的目录：

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

`PLUGIN.yaml`

Plugin metadata。requires_env 和 optional_env 块会自动填充 Hermes config UI 条目（参见下面的 Surfacing Env Vars）。

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

`adapter.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:
        # 连接到 platform API，启动 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):
        # 通过 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 _env_enablement() -> dict | None:
    token = os.getenv("MY_PLATFORM_TOKEN", "").strip()
    channel = os.getenv("MY_PLATFORM_CHANNEL", "").strip()
    if not (token and channel):
        return None
    seed = {"token": token, "channel": channel}
    home = os.getenv("MY_PLATFORM_HOME_CHANNEL")
    if home:
        seed["home_channel"] = {"chat_id": home, "name": "Home"}
    return seed


def register(ctx):
    """Plugin entry point — 由 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",
        # Env-driven auto-configuration — 在 adapter 构造之前，
        # 从 env vars 填充 PlatformConfig.extra。参见下面的
        # “Env-Driven Auto-Configuration” section。
        env_enablement_fn=_env_enablement,
        # Cron home-channel delivery support。允许 deliver=my_platform cron
        # jobs 在不编辑 cron/scheduler.py 的情况下路由。参见下面的
        # “Cron Delivery” section。
        cron_deliver_env_var="MY_PLATFORM_HOME_CHANNEL",
        # 每个平台的用户授权 env vars
        allowed_users_env="MY_PLATFORM_ALLOWED_USERS",
        allow_all_env="MY_PLATFORM_ALLOW_ALL_USERS",
        # 智能分块的消息长度限制（0 = 无限制）
        max_message_length=4000,
        # 注入到 system prompt 中的 LLM 指导
        platform_hint=(
            "You are chatting via My Platform. "
            "It supports markdown formatting."
        ),
        # Display
        emoji="💬",
    )

    # 可选：注册 platform-specific tools
    ctx.register_tool(
        name="my_platform_search",
        toolset="my_platform",
        schema={...},
        handler=my_search_handler,
    )

配置

用户在 config.yaml 中配置 platform：

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

或通过环境变量配置（adapter 会在 __init__ 中读取）。

Plugin System 自动处理的内容

当你调用 ctx.register_platform() 时，以下集成点会自动为你处理 —— 不需要修改核心代码：

集成点	工作方式
Gateway adapter creation	在 built-in `if/elif` chain 之前检查 Registry
Config parsing	`Platform._missing_()` 接受任何 platform name
Connected platform validation	调用 Registry `validate_config()`
User authorization	检查 `allowed_users_env` / `allow_all_env`
Env-only auto-enable	`env_enablement_fn` 填充 `PlatformConfig.extra` + `home_channel`
YAML config bridge	`apply_yaml_config_fn` 将 `config.yaml` keys 转换为 env vars / extras
Cron delivery	`cron_deliver_env_var` 让 `deliver=<name>` 可用
hermes config UI entries	`plugin.yaml` 中的 `requires_env` / `optional_env` 自动填充
send_message tool	通过 live gateway adapter 路由
Webhook cross-platform delivery	Registry 检查 known platforms
/update command access	`allow_update_command` flag
Channel directory	Plugin platforms 包含在枚举中
System prompt hints	`platform_hint` 注入到 LLM context
Message chunking	`max_message_length` 用于智能拆分
PII redaction	`pii_safe` flag
hermes status	显示带有 `(plugin)` 标签的 plugin platforms
hermes gateway setup	Plugin platforms 出现在 setup menu 中
hermes tools / hermes skills	per-platform config 中的 plugin platforms
Token lock（multi-profile）	在你的 `connect()` 中使用 `acquire_scoped_lock()`
Orphaned config warning	当 plugin 缺失时输出描述性日志

Env-Driven Auto-Configuration

大多数用户通过将 env vars 放入 ~/.hermes/.env 来设置 platform，而不是编辑 config.yaml。env_enablement_fn hook 允许你的 plugin 在 adapter 构造之前获取这些 env vars，因此 hermes gateway status、get_connected_platforms() 和 cron delivery 可以在不实例化 platform SDK 的情况下看到正确状态。

def _env_enablement() -> dict | None:
    """从 env vars 填充 PlatformConfig.extra。

    在 load_gateway_config() 期间由 platform registry 调用。
    当 platform 未达到最小配置要求时返回 None — 然后
    caller 会跳过 auto-enabling。返回一个 dict 来填充 extras。

    特殊的 'home_channel' key 会被提取，并在 PlatformConfig 上变成一个
    正式的 HomeChannel dataclass；其他每个 key 都会
    合并到 PlatformConfig.extra 中。
    """
    token = os.getenv("MY_PLATFORM_TOKEN", "").strip()
    channel = os.getenv("MY_PLATFORM_CHANNEL", "").strip()
    if not (token and channel):
        return None
    seed = {"token": token, "channel": channel}
    home = os.getenv("MY_PLATFORM_HOME_CHANNEL")
    if home:
        seed["home_channel"] = {
            "chat_id": home,
            "name": os.getenv("MY_PLATFORM_HOME_CHANNEL_NAME", "Home"),
        }
    return seed


def register(ctx):
    ctx.register_platform(
        name="my_platform",
        label="My Platform",
        adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
        check_fn=check_requirements,
        validate_config=validate_config,
        env_enablement_fn=_env_enablement,
        # ... other fields
    )

YAML→env Config Bridge

一些用户更喜欢设置 config.yaml keys（my_platform.require_mention、my_platform.allowed_channels 等），而不是 env vars。apply_yaml_config_fn hook 允许你的 plugin 拥有这层转换，而不是强迫核心 gateway/config.py 了解你的 platform 的 YAML schema。

import os

def _apply_yaml_config(yaml_cfg: dict, platform_cfg: dict) -> dict | None:
    """将 config.yaml `my_platform:` keys 转换为 env vars / extras。

    yaml_cfg     — 完整的顶层 parsed config.yaml dict
    platform_cfg — platform 自己的 sub-dict（yaml_cfg.get("my_platform", {})）

    可以直接修改 os.environ（使用 `not os.getenv(...)` guards 来
    保留 env > YAML 的优先级），并且/或者返回一个 dict 以合并到
    PlatformConfig.extra。没有 extras 时返回 None 或 {}。
    """
    if "require_mention" in platform_cfg and not os.getenv("MY_PLATFORM_REQUIRE_MENTION"):
        os.environ["MY_PLATFORM_REQUIRE_MENTION"] = str(platform_cfg["require_mention"]).lower()
    allowed = platform_cfg.get("allowed_channels")
    if allowed is not None and not os.getenv("MY_PLATFORM_ALLOWED_CHANNELS"):
        if isinstance(allowed, list):
            allowed = ",".join(str(v) for v in allowed)
        os.environ["MY_PLATFORM_ALLOWED_CHANNELS"] = str(allowed)
    return None  # 没有额外内容需要合并到 PlatformConfig.extra

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

该 hook 会在 load_gateway_config() 期间被调用，调用时机是在 generic shared-key loop 之后（该 loop 处理诸如 unauthorized_dm_behavior、notice_delivery、reply_prefix、require_mention 等 common keys），并且在 _apply_env_overrides() 之前，因此你的 plugin 只需要桥接 platform-specific keys。

hook 抛出的异常会被吞掉，并以 debug level 记录日志 —— 行为不正常的 plugin 永远不会中止 gateway config load。

Cron Delivery

为了让 deliver=my_platform cron jobs 路由到已配置的 home channel，请将 cron_deliver_env_var 设置为保存默认 chat/room/channel ID 的 env var 名称：

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

scheduler 在解析 deliver=my_platform jobs 的 home target 时会读取这个 env var，并且也会在 _KNOWN_DELIVERY_PLATFORMS 风格的检查中将该 platform 视为有效的 cron target。如果你的 env_enablement_fn 填充了一个 home_channel dict（见上文），它会优先 —— cron_deliver_env_var 是为那些在 env seeding 之前运行的 cron jobs 准备的 fallback。

Out-of-process cron delivery

cron_deliver_env_var 会让你的 platform 成为一个被识别的 deliver= target。为了让 cron job 在与 gateway 不同的进程中运行时实际发送成功（即 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,
):
    """打开一个临时连接 / 获取一个新的 token，发送，然后关闭。"""
    # ... 打开连接，发送消息，返回结果 ...
    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,
)

为什么这个 hook 是必要的：内置 platforms（Telegram、Discord、Slack 等）在 tools/send_message_tool.py 中附带 direct REST helpers，因此 cron 可以在不与 gateway 保持同一进程的情况下 delivery。Plugin platforms 过去依赖 _gateway_runner_ref()，它在 gateway 进程之外会返回 None，因此没有 standalone_sender_fn 时，cron-side send 会失败，并显示 No live adapter for platform '<name>'。

该函数接收与 live adapter 相同的 pconfig 和 chat_id，另外还有可选的 thread_id、media_files 和 force_document keyword arguments。返回 {"success": True, "message_id": ...} 会被视为成功 delivery；返回 {"error": "..."} 会将消息显示在 cron 的 delivery_errors 中。函数内部抛出的异常会被 dispatcher 捕获，并报告为 Plugin standalone send failed: <reason>。参考实现位于 plugins/platforms/{irc,teams,google_chat}/adapter.py。

在 `hermes config` 中展示 Env Vars

hermes_cli/config.py 会在 import time 扫描 plugins/platforms/*/plugin.yaml，并根据 requires_env 和（可选的）optional_env 块自动填充 OPTIONAL_ENV_VARS。使用 rich-dict 形式可以提供合适的 descriptions、prompts、password flags 和 URLs —— CLI setup UI 会免费获得它们。

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: "来自 My Platform console 的 Bot API token"
    prompt: "My Platform bot token"
    url: "https://my-platform.example.com/bots"
    password: true
  - name: MY_PLATFORM_CHANNEL
    description: "要加入的 Channel（例如 #hermes）"
    prompt: "Channel"
    password: false
optional_env:
  - name: MY_PLATFORM_HOME_CHANNEL
    description: "cron delivery 的默认 channel（默认为 MY_PLATFORM_CHANNEL）"
    prompt: "Home channel（或留空）"
    password: false
  - name: MY_PLATFORM_ALLOWED_USERS
    description: "允许与 bot 对话的逗号分隔 user IDs"
    prompt: "Allowed users（逗号分隔）"
    password: false

支持的 dict keys：name（必需）、description、prompt、url、password（bool；省略时会根据 *_TOKEN / *_SECRET / *_KEY / *_PASSWORD / *_JSON 后缀自动检测）、category（默认为 "messaging"）。

Bare-string entries（- MY_PLATFORM_TOKEN）仍然可用 —— 它们会获得一个根据 plugin 的 label 自动派生的 generic description。如果 OPTIONAL_ENV_VARS 中已经存在同一个 var 的 hardcoded entry，则它优先（back-compat）；plugin.yaml 形式会作为 fallback。

Platform-Specific Slow-LLM UX

某些平台存在一些约束，会改变慢速 LLM 响应应该如何呈现：

LINE 会发放一个单次使用的 reply token，该 token 在入站事件后大约 60 秒过期。使用该 token 回复是免费的；回退到计费的 Push API 则不是。如果 LLM 在截止时间前还没有完成，那么选择就是“消耗付费 Push 配额”或“在 reply token 过期前用它做一些更聪明的事情”。
WhatsApp 会在 24 小时后将 session 标记为 inactive，之后只接受 template messages。
SMS 没有 typing indicators 或 progressive updates 的概念 —— 长响应看起来就像 bot 离线了一样。

这些是真实约束，基础的 BasePlatformAdapter 无法预判。plugin surface 有意保留空间，让 adapter 可以在基础 typing loop 之上叠加 platform-specific UX，而不需要扩展 kwarg 列表。

模式：子类化 `_keep_typing` 以叠加 mid-flight UX

BasePlatformAdapter._keep_typing 是 typing-indicator heartbeat —— 它会在 LLM 生成期间作为后台任务运行，并在响应 delivered 时被取消。为了在某个阈值处叠加 platform-specific 行为（例如在 45 秒时发送一个“still thinking”气泡），请在你的 adapter 中 override _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
            # Platform-specific work here — 对于 LINE，使用缓存的 reply token
            # 发送一个 Template Buttons "Get answer" 气泡，
            # 这样用户之后可以通过 postback callback 中
            # 新的（免费）reply token 获取缓存响应。
            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(...)。typing heartbeat 本身是有用的 —— 不要替换它，而是在其上叠加。
在 finally 中清理 side task。当 LLM 完成（或 /stop 取消 run）时，gateway 会取消 typing task。你的 side task 也必须观察到该 cancellation，否则它会继续存在，并且可能在响应已经 delivered 后触发。
配合 interrupt_session_activity 使用，以便在用户发出 /stop 时解决任何孤立的 UX 状态。对于 LINE，这意味着将 postback cache entry 从 PENDING 转换为 ERROR，这样持久的 “Get answer” 按钮会 deliver 一条 “Run was interrupted” 消息，而不是循环。

模式：子类化 `send`，通过 cache 路由，而不是立即发送

如果你的 slow-response UX 会缓存响应以供后续获取（LINE 的 postback flow），你的 send override 需要识别三种模式：

该 chat 有 pending postback active → 将响应缓存在 request_id 下，不发送任何可见内容。
System busy-ack（⚡ Interrupting、⏳ Queued、⏩ Steered）→ 绕过 cache 并可见发送，这样用户可以看到 gateway 对其输入的响应。
Normal response → 像平常一样通过 reply-token-or-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 是 gateway 自己的 busy-acknowledgment prefixes（⚡、⏳、⏩、💾）。无论 cached UX state 如何，始终让这些内容可见通过。

何时适合使用此模式

当满足以下条件时，使用 typing-loop override 方法：

platform 的 outbound API 有硬性的 time-window 约束（single-use reply token、expiring sticky session 等）并且
在该 platform 上，可见的 mid-flight bubble 是可接受的 UX。

当满足以下条件时，使用更简单的 slow_response_threshold = 0 always-Push 路径：

platform 没有有意义的免费 vs. 付费区别，或者
用户社区更喜欢 “loading… loading… DONE” 这种 silence-then-response，而不是交互式 intermediate bubble。

LINE 两者都支持：threshold 默认是 45 秒，用于免费 postback fetch，而 LINE_SLOW_RESPONSE_THRESHOLD=0 会恢复为 “always Push fallback”。

参考实现

完整的 LINE postback 实现请参见 plugins/platforms/line/adapter.py —— 一个 RequestCache state machine（PENDING → READY → DELIVERED，以及用于 /stop 的 ERROR）、一个在阈值处触发 Template Buttons bubble 的 _keep_typing override、一个通过 cache 路由的 send override，以及一个解决孤立 PENDING entries 的 interrupt_session_activity override。

参考实现（Plugin Path）

完整可工作的示例请参见仓库中的 plugins/platforms/irc/ —— 一个没有外部依赖的完整 async IRC adapter。plugins/platforms/teams/ 涵盖 Bot Framework / Adaptive Cards，plugins/platforms/google_chat/ 涵盖 OAuth-based REST APIs，plugins/platforms/line/ 涵盖 webhook-driven Messaging APIs 以及 platform-specific slow-LLM UX。

分步检查清单（Built-in Path）

:::[注意] 此检查清单用于将 platform 直接添加到 Hermes 核心代码库中 —— 通常由核心贡献者为官方支持的平台执行。社区 / 第三方平台应使用上面的 Plugin Path。 :::

1. Platform Enum

在 gateway/config.py 中将你的 platform 添加到 Platform enum：

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

2. Adapter File

创建 gateway/platforms/newplat.py：

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

def check_newplat_requirements() -> bool:
    """如果依赖可用，则返回 True。"""
    return SOME_SDK_AVAILABLE

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

    async def connect(self) -> 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):
        # 通过 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",  # 或 "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（`gateway/config.py`）

三个触点：

get_connected_platforms() —— 为你的 platform 的必需 credentials 添加检查
load_gateway_config() —— 添加 token env map 条目：Platform.NEWPLAT: "NEWPLAT_TOKEN"
_apply_env_overrides() —— 将所有 NEWPLAT_* env vars 映射到 config

4. Gateway Runner（`gateway/run.py`）

五个触点：

_create_adapter() —— 添加一个 elif platform == Platform.NEWPLAT: 分支
_is_user_authorized() allowed_users map —— Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
_is_user_authorized() allow_all map —— Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
Early env check _any_allowlist tuple —— 添加 "NEWPLAT_ALLOWED_USERS"
Early env check _allow_all tuple —— 添加 "NEWPLAT_ALLOW_ALL_USERS"
_UPDATE_ALLOWED_PLATFORMS frozenset —— 添加 Platform.NEWPLAT

5. Cross-Platform Delivery

gateway/platforms/webhook.py —— 将 "newplat" 添加到 delivery type tuple
cron/scheduler.py —— 添加到 _KNOWN_DELIVERY_PLATFORMS frozenset 和 _deliver_result() platform map

6. CLI Integration

hermes_cli/config.py —— 将所有 NEWPLAT_* vars 添加到 _EXTRA_ENV_KEYS
hermes_cli/gateway.py —— 向 _PLATFORMS list 添加条目，包括 key、label、emoji、token_var、setup_instructions 和 vars
hermes_cli/platforms.py —— 添加带有 label 和 default_toolset 的 PlatformInfo 条目（由 skills_config 和 tools_config TUIs 使用）
hermes_cli/setup.py —— 添加 _setup_newplat() 函数（可以委托给 gateway.py），并将 tuple 添加到 messaging platforms list
hermes_cli/status.py —— 添加 platform detection 条目："NewPlat": ("NEWPLAT_TOKEN", "NEWPLAT_HOME_CHANNEL")
hermes_cli/dump.py —— 将 "newplat": "NEWPLAT_TOKEN" 添加到 platform detection dict

7. Tools

tools/send_message_tool.py —— 将 "newplat": Platform.NEWPLAT 添加到 platform map
tools/cronjob_tools.py —— 将 newplat 添加到 delivery target description string

8. Toolsets

toolsets.py —— 添加 "hermes-newplat" toolset definition，并包含 _HERMES_CORE_TOOLS
toolsets.py —— 将 "hermes-newplat" 添加到 "hermes-gateway" includes list

9. 可选：Platform Hints

agent/prompt_builder.py —— 如果你的 platform 有特定的渲染限制（不支持 markdown、消息长度限制等），请向 _PLATFORM_HINTS dict 添加一个条目。这会将 platform-specific guidance 注入到 system prompt 中：

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

并非所有 platform 都需要 hints —— 只有当 agent 的行为应该有所不同时才添加。

10. Tests

创建 tests/gateway/test_newplat.py，覆盖：

从 config 构造 Adapter
Message event building
Send method（mock 外部 API）
Platform-specific features（encryption、routing 等）

11. Documentation

文件	要添加的内容
`website/docs/user-guide/messaging/newplat.md`	完整 platform setup page
`website/docs/user-guide/messaging/index.md`	Platform comparison table、architecture diagram、toolsets table、security section、next-steps link
`website/docs/reference/environment-variables.md`	所有 `NEWPLAT_*` env vars
`website/docs/reference/toolsets-reference.md`	`hermes-newplat` toolset
`website/docs/integrations/index.md`	Platform link
`website/sidebars.ts`	docs page 的 Sidebar entry
`website/docs/developer-guide/architecture.md`	Adapter count + listing
`website/docs/developer-guide/gateway-internals.md`	Adapter file listing

Parity Audit

在将新的 platform PR 标记为完成之前，请针对一个已有 platform 运行 parity audit：

# 查找所有提到 reference platform 的 .py 文件
search_files "bluebubbles" output_mode="files_only" file_glob="*.py"

# 查找所有提到 new platform 的 .py 文件
search_files "newplat" output_mode="files_only" file_glob="*.py"

# 第一个集合中有、第二个集合中没有的任何文件都是潜在缺口

对 .md 和 .ts 文件重复执行。调查每个缺口 —— 它是 platform enumeration（需要更新），还是 platform-specific reference（跳过）？

常见模式

Long-Poll Adapters

如果你的 adapter 使用 long-polling（如 Telegram 或 Weixin），请使用 polling loop task：

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 Adapters

如果 platform 将消息推送到你的 endpoint（如 WeCom Callback），请运行一个 HTTP server：

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")  # 立即确认

对于具有严格响应截止时间的平台（例如 WeCom 的 5 秒限制），始终立即确认，并在之后通过 API 主动 deliver agent 的回复。Agent sessions 会运行 3–30 分钟 —— 在 callback response window 内进行 inline replies 是不可行的。

Token Locks

如果 adapter 持有具有唯一 credential 的持久连接，请添加一个 scoped lock，以防止两个 profiles 使用相同 credential：

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)

参考实现

Adapter	Pattern	Complexity	Good reference for
`bluebubbles.py`	REST + webhook	Medium	简单 REST API integration
`weixin.py`	Long-poll + CDN	High	Media handling、encryption
`wecom_callback.py`	Callback/webhook	Medium	HTTP server、AES crypto、multi-app
`telegram.py`	Long-poll + Bot API	High	Full-featured adapter with groups、threads

添加平台适配器

架构概览

Plugin Path（推荐）

PLUGIN.yaml

adapter.py

配置

Plugin System 自动处理的内容

Env-Driven Auto-Configuration

YAML→env Config Bridge

Cron Delivery

Out-of-process cron delivery

在 hermes config 中展示 Env Vars

Platform-Specific Slow-LLM UX

模式：子类化 _keep_typing 以叠加 mid-flight UX

模式：子类化 send，通过 cache 路由，而不是立即发送

何时适合使用此模式

参考实现

参考实现（Plugin Path）

分步检查清单（Built-in Path）

1. Platform Enum

2. Adapter File

3. Gateway Config（gateway/config.py）

4. Gateway Runner（gateway/run.py）

5. Cross-Platform Delivery

6. CLI Integration

7. Tools

8. Toolsets

9. 可选：Platform Hints

10. Tests

11. Documentation

Parity Audit

常见模式

Long-Poll Adapters

Callback/Webhook Adapters

Token Locks

参考实现

快速上手

使用 Hermes

功能

消息平台

集成

指南与教程

开发者指南

`PLUGIN.yaml`

`adapter.py`

在 `hermes config` 中展示 Env Vars

模式：子类化 `_keep_typing` 以叠加 mid-flight UX

模式：子类化 `send`，通过 cache 路由，而不是立即发送

3. Gateway Config（`gateway/config.py`）

4. Gateway Runner（`gateway/run.py`）