<!-- Source: https://hermesbible.com/docs/developer-guide/adding-platform-adapters -->

新增平台適配器

本指南涵蓋向 Hermes 閘道器新增一個新的訊息平台。平台適配器將 Hermes 連接到外部訊息服務（Telegram、Discord、WeCom 等），讓使用者可以透過該服務與 Agent 互動。

提示

新增平台有兩種方式：

• 外掛（建議用於社群/第三方）：將外掛目錄放入 ~/.hermes/plugins/——無需修改核心程式碼。參見下方外掛路徑。
• 內建：修改程式碼、設定和文件中的 20 多個檔案。使用下方的內建檢查清單。

架構概覽

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) 轉發，基礎類別將其路由到閘道器運行器。

外掛路徑（建議）

外掛系統讓你在不修改任何 Hermes 核心程式碼的情況下新增平台適配器。你的外掛是一個包含兩個檔案的目錄：

~/.hermes/plugins/my-platform/
  plugin.yaml      # 外掛中繼資料
  adapter.py       # 適配器類別 + register() 入口點

plugin.yaml

外掛中繼資料。requires_env 和 optional_env 區塊自動填充 hermes config UI 項目（參見下方在 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          # 純字串即可運作
  - name: MY_PLATFORM_CHANNEL  # 或用豐富字典獲得更好的 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:
        # 連接到平台 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(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):
    """外掛入口點——由 Hermes 外掛系統呼叫。"""
    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",
        # 環境驅動的自動設定——在適配器建構前
        # 從環境變數填充 PlatformConfig.extra。
        # 參見下方「環境驅動的自動設定」區塊。
        env_enablement_fn=_env_enablement,
        # 排程主頻道遞送支援。讓 deliver=my_platform 排程任務
        # 無需編輯 cron/scheduler.py 即可路由。
        # 參見下方「排程遞送」區塊。
        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() 時，以下整合點會為你處理——無需修改核心程式碼：

環境驅動的自動設定

大多數使用者透過將環境變數放入 ~/.hermes/.env 來設定平台，而非編輯 config.yaml。env_enablement_fn hook 讓你的外掛在適配器建構之前拾取這些環境變數，因此 hermes gateway status、get_connected_platforms() 和排程遞送看到正確的狀態，而無需實例化平台 SDK。

def _env_enablement() -> dict | None:
    """從環境變數填充 PlatformConfig.extra。

    由平台註冊表在 load_gateway_config() 期間呼叫。
    當平台未最低限度設定時回傳 None——呼叫者
    然後跳過自動啟用。回傳字典以填充 extras。

    特殊的 'home_channel' 鍵被提取並成為
    PlatformConfig 上的適當 HomeChannel dataclass；
    每個其他鍵被合併到 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,
        # ... 其他欄位
    )

YAML→env 設定橋接

部分使用者偏好設定 config.yaml 鍵（my_platform.require_mention、my_platform.allowed_channels 等）而非環境變數。apply_yaml_config_fn hook 讓你的外掛擁有此轉換，而非強制核心 gateway/config.py 知道你的平台的 YAML schema。

import os

def _apply_yaml_config(yaml_cfg: dict, platform_cfg: dict) -> dict | None:
    """將 config.yaml `my_platform:` 鍵轉換為環境變數 / extras。

    yaml_cfg     — 完整的頂層已解析 config.yaml 字典
    platform_cfg — 平台自己的子字典（yaml_cfg.get("my_platform", {})）

    可能直接修改 os.environ（使用 `not os.getenv(...)` 保護
    以保留 env > YAML 優先順序）和/或回傳字典以合併到
    PlatformConfig.extra。回傳 None 或 {} 表示無 extras。
    """
    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  # 無 extras 需要合併到 PlatformConfig.extra

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

該 hook 在 load_gateway_config() 期間，在通用共享鍵迴圈（處理 unauthorized_dm_behavior、notice_delivery、reply_prefix、require_mention 等常見鍵）之後、_apply_env_overrides() 之前被調用，因此你的外掛只需橋接平台特定的鍵。

hook 拋出的異常被吞沒並以除錯級別記錄——行為不當的外掛永遠不會中斷閘道器設定載入。

排程遞送

要讓 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 任務的主目標時讀取此環境變數，並在 _KNOWN_DELIVERY_PLATFORMS 類型的檢查中也將平台視為有效的排程目標。如果你的 env_enablement_fn 填充了 home_channel 字典（見上方），則優先使用它——cron_deliver_env_var 是環境填充之前執行的排程任務的回退。

跨程序的排程遞送

cron_deliver_env_var 使你的平台成為公認的 deliver= 目標。要讓排程任務在與閘道器不同的程序中運行時（即 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,
)

為什麼需要此 hook：內建平台（Telegram、Discord、Slack 等）在 tools/send_message_tool.py 中發送直接的 REST 輔助函數，因此排程可以在不保持閘道器在同一程序中的情況下遞送。外掛平台歷史上依賴 _gateway_runner_ref()，它在閘道器程序外回傳 None，因此沒有 standalone_sender_fn 時，排程端的發送會因 No live adapter for platform '<name>' 而失敗。

此函數接收與即時適配器相同的 pconfig 和 chat_id，加上選用的 thread_id、media_files 和 force_document 關鍵字參數。回傳 {"success": True, "message_id": ...} 被視為成功的遞送；回傳 {"error": "..."} 在排程的 delivery_errors 中顯示訊息。函數內部拋出的異常被分派器捕捉並報告為 Plugin standalone send failed: <reason>。參考實作位於 plugins/platforms/{irc,teams,google_chat}/adapter.py。

在 hermes config 中暴露環境變數

hermes_cli/config.py 在匯入時掃描 plugins/platforms/*/plugin.yaml，並從 requires_env 和（選用）optional_env 區塊自動填充 OPTIONAL_ENV_VARS。使用豐富字典形式貢獻適當的描述、提示、密碼標誌和 URL——CLI 設定 UI 免費拾取它們。

# 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

支援的字典鍵： name（必要）、description、prompt、url、password（bool；從 *_TOKEN / *_SECRET / *_KEY / *_PASSWORD / *_JSON 後綴自動偵測時省略）、category（預設為 "messaging"）。

純字串項目（- MY_PLATFORM_TOKEN）仍然運作——它們獲得從外掛的 label 自動衍生的通用描述。如果 OPTIONAL_ENV_VARS 中已存在相同變數的硬編碼項目，則其優先（向後相容）；plugin.yaml 形式作為回退。

平台特定的慢速 LLM UX

部分平台有限制，改變了慢速 LLM 回應的呈現方式：

• LINE 發出單次使用的回覆權杖，大約在入站事件後 60 秒過期。使用該權杖回覆是免費的；回退到計量的 Push API 則不是。如果 LLM 在期限前未完成，選擇是「消耗付費 Push 額度」或「在回覆權杖過期前用更聰明的方式處理」。
• WhatsApp 在 24 小時後將會話標記為不活躍，之後只接受模板訊息。
• SMS 沒有輸入指示器或漸進更新的概念——長回應看起來就像機器人離線了。

這些是基礎 BasePlatformAdapter 無法預見的真實限制。外掛表面刻意留出空間，讓適配器在基礎輸入迴圈之上堆疊平台特定的 UX，而不擴展 kwargs 列表。

模式：子類別化 _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「取得回應」泡泡，
            # 讓使用者可以透過 postback 回呼的
            # 新（免費）回覆權杖取得快取的回應。
            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 時解決任何孤立的 UX 狀態。對 LINE，這意味著將 postback 快取條目從 PENDING 轉換為 ERROR，使持久的「取得回應」按鈕傳達「執行已中斷」訊息而非循環。

模式：子類別化 send 以透過快取路由而非立即發送

如果你的慢速回應 UX 快取回應供稍後檢索（LINE 的 postback 流程），你的 send 覆寫需要識別三種模式：

1. 此聊天有作用中的 pending postback → 將回應快取在 request_id 下，不發送任何可見內容。
2. 系統 busy-ack（⚡ Interrupting、⏳ Queued、⏩ Steered）→ 繞過快取並可見地發送，讓使用者看到閘道器對其輸入的回應。
3. 正常回應 → 像往常一樣透過 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 是閘道器自己的 busy-acknowledgment 前綴（⚡、⏳、⏩、💾）。始終讓那些可見地通過，不論快取的 UX 狀態如何。

何時適用此模式

在以下情況使用輸入迴圈覆寫方法：

• 平台的外發 API 有硬性時間窗口限制（單次使用回覆權杖、過期的粘性會話等）且
• 在該平台上，可見的中途泡泡是可接受的 UX。

在以下情況使用較簡單的 slow_response_threshold = 0 的 always-Push 路徑：

• 平台沒有有意義的免費 vs 付費區分，或
• 使用者社群偏好「載入中…載入中…完成」的靜默後回應，而非互動式的中間泡泡。

LINE 兩者都支援：閾值預設為 45 秒用於免費的 postback 取得，LINE_SLOW_RESPONSE_THRESHOLD=0 回退到「總是 Push 回退」。

參考實作

參見 plugins/platforms/line/adapter.py 的完整 LINE postback 實作——一個 RequestCache 狀態機（PENDING → READY → DELIVERED，加上 ERROR 用於 /stop）、在閾值時觸發 Template Buttons 泡泡的 _keep_typing 覆寫、透過快取路由的 send 覆寫，以及解決孤立 PENDING 條目的 interrupt_session_activity 覆寫。

參考實作（外掛路徑）

參見 repo 中的 plugins/platforms/irc/ 了解完整的運作範例——一個零外部依賴的完整非同步 IRC 適配器。plugins/platforms/teams/ 涵蓋 Bot Framework / Adaptive Cards，plugins/platforms/google_chat/ 涵蓋基於 OAuth 的 REST API，plugins/platforms/line/ 涵蓋 webhook 驅動的 Messaging API 搭配平台特定的慢速 LLM UX。

---

逐步檢查清單（內建路徑）

注意

此檢查清單用於直接將平台新增到 Hermes 核心程式碼庫——通常由核心貢獻者為官方支援的平台執行。社群/第三方平台應使用上方的外掛路徑。

1. Platform 列舉

在 gateway/config.py 的 Platform 列舉中加入你的平台：

class Platform(str, Enum):
    # ... 現有平台 ...
    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:
    """若依賴可用則回傳 True。"""
    return SOME_SDK_AVAILABLE

class NewPlatAdapter(BasePlatformAdapter):
    def __init__(self, config: PlatformConfig):
        super().__init__(config, Platform.NEWPLAT)
        # 從 config.extra 字典讀取設定
        extra = config.extra or {}
        self._api_key = extra.get("api_key") or os.getenv("NEWPLAT_API_KEY", "")

    async def connect(self) -> bool:
        # 建立連接，啟動輪詢/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() — 新增權杖 env 映射項目：Platform.NEWPLAT: "NEWPLAT_TOKEN"
3. _apply_env_overrides() — 將所有 NEWPLAT_* 環境變數映射到設定

4. 閘道器運行器（gateway/run.py）

五個接觸點：

1. _create_adapter() — 新增 elif platform == Platform.NEWPLAT: 分支
2. _is_user_authorized() allowed_users 映射 — Platform.NEWPLAT: "NEWPLAT_ALLOWED_USERS"
3. _is_user_authorized() allow_all 映射 — Platform.NEWPLAT: "NEWPLAT_ALLOW_ALL_USERS"
4. 早期 env 檢查 _any_allowlist 元組 — 新增 "NEWPLAT_ALLOWED_USERS"
5. 早期 env 檢查 _allow_all 元組 — 新增 "NEWPLAT_ALLOW_ALL_USERS"
6. _UPDATE_ALLOWED_PLATFORMS frozen set — 新增 Platform.NEWPLAT

5. 跨平台遞送

1. gateway/platforms/webhook.py — 將 "newplat" 加入遞送類型元組
2. cron/scheduler.py — 加入 _KNOWN_DELIVERY_PLATFORMS frozen set 和 _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（被 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_CORE_TOOLS 的 "hermes-newplat" 工具集定義
2. toolsets.py — 將 "hermes-newplat" 加入 "hermes-gateway" 的 includes 列表

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."
    ),
}

並非所有平台都需要提示——只有在 Agent 的行為應該不同時才新增。

10. 測試

建立 tests/gateway/test_newplat.py，涵蓋：

• 從設定建構適配器
• 訊息事件建構
• 發送方法（mock 外部 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 檔案重複此操作。調查每個缺口——它是需要更新的平台列舉，還是平台特定的參考（跳過）？

常見模式

長輪詢適配器

如果你的適配器使用長輪詢（如 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))

Callback/Webhook 適配器

如果平台將訊息推送到你的端點（如 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")  # 立即確認

對於有嚴格回應期限的平台（例如 WeCom 的 5 秒限制），始終立即確認並稍後透過 API 主動遞送 Agent 的回覆。Agent 會話運行 3-30 分鐘——在 callback 回應視窗內的行內回覆不可行。

Token 鎖

如果適配器使用唯一憑證持有持久連接，新增範圍鎖以防止兩個 profile 使用相同的憑證：

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
    # ... 連接

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

參考實作

---

---

建立技能