プラットフォームアダプタ

源码版本v2026.7.20

職務

各プラットフォーム(Telegram、Discord、Slack、Signal、WhatsApp、微信、QQ…)にアダプタ (adapter) があり、プラットフォームネイティブイベントを統一 MessageEvent に翻訳し、統一送信アクションをプラットフォーム API に翻訳し戻す。アダプタは BasePlatformAdapter を継承し、少数の抽象メソッドを実装するだけでよい。アダプタは内蔵 gateway/platforms/ またはプラグイン (plugin) plugins/platforms/ から提供され、どちらも PlatformRegistry に登録される。

設計動機

プラットフォームのネイティブ API は差が大きい:Telegram Bot API は sendMessage + MarkdownV2 エスケープ、Discord は webhook + スノーフレーク ID、Signal はローカルの signal-cli daemon と話す、微信はリッチテキストすらサポートしない。これらの差を agent 主ループに押し込むと、agent は永遠に書き終わらない。アダプタ層がこれらを吸収する——上には統一 MessageEvent を吐き、下には統一 send(chat_id, content) 抽象を受ける。PlatformRegistry の deferred ロードと組み合わせれば、hermes chat のようなプラットフォームを触らないコマンドが二十以上のプラットフォーム SDK に引きずられない。

主要ファイル

• class BasePlatformAdapter(ABC):2317 — アダプタ抽象基底クラス
• 抽象メソッド send / send_draft:2983-3015 — サブクラス必須の送信インターフェース
• MessageType / ProcessingOutcome / MessageEvent:1737-1759 — 統一イベントモデル
• CachedMedia:1638-1737 — メディアキャッシュ(重複アップロード回避)
• SendResult / EphemeralReply:1898-2068 — 送信結果と一時返信
• 汎用送信補助:3173-3326 — send_slash_confirm / send_clarify / send_typing / send_multiple_images
• gateway/platforms/ ディレクトリ — signal / whatsapp_cloud / weixin / bluebubbles / yuanbao / qqbot…
• plugins/platforms/ ディレクトリ — Telegram / Discord / Slack(プラグイン形式)
• ADDING_A_PLATFORM.md — 新規プラットフォーム追加ガイド
• PlatformRegistry:162 — レジストリ (registry)(register / register_deferred / unregister)

データフロー

1. 起動期 GatewayRunner(gateway/run.py:3029) が _start_one_profile_adapter:9389 経由でアダプタを起動する。
2. アダプタは PlatformRegistry(gateway/platform_registry.py:162) から名前で探し、deferred ロードをサポート(安価にプレースホルダを置き、実際に使う時にインポート)。
3. アダプタはプラットフォームイベントを監視し、MessageEvent:1759 に翻訳して _handle_message:9947 に渡す。
4. agent がストリーミング出力を出すと、アダプタは統一インターフェース(send:3008 / send_draft)で内容をプラットフォームに配信する。メディアは CachedMedia:1638 でキャッシュする。
5. プラットフォーム差異(メッセージ長、リッチテキスト、画像上限)はすべてアダプタ内部に吸収され、ゲートウェイ (gateway) コアは関与しない。

PlatformRegistry は二つの登記簿を維持する——_entries は実装済み、_deferred は安価なプレースホルダ。照会された時に初めて後者を前者に変換する。主要フィールドは次のとおり。

class PlatformRegistry:
    """Central registry of platform adapters.

    Thread-safe for reads (dict lookups are atomic under GIL).
    Writes happen at startup during sequential discovery.
    """
    def __init__(self) -> None:
        self._entries: dict[str, PlatformEntry] = {}
        # Deferred platform loaders: name -> zero-arg callable that imports the
        # owning plugin module (which calls register() and populates _entries).
        self._deferred: dict[str, Callable[[], None]] = {}

    def get(self, name: str) -> Optional[PlatformEntry]:
        """Look up a platform entry by name."""
        if name not in self._entries:
            self._resolve(name)
        return self._entries.get(name)

deferred の動機はコメントに書かれている——プラットフォームアダプタのモジュールトップは(lark_oapi、discord.py、slack_bolt のような)重型 SDK を import する。全量ロードすると hermes chat が数秒遅くなる。プレースホルダ loader は初回の get(name) で初めて本当に import する。

BasePlatformAdapter は ABC でサブクラスに三点セット——connect、disconnect、send の実装を強制する。これがアダプタの最小契約だ。

@abstractmethod
async def connect(self, *, is_reconnect: bool = False) -> bool:
    """Connect to the platform and start receiving messages.

    Args:
        is_reconnect: False on a cold first boot; True when the reconnect
            watcher is re-establishing a platform that was previously running
            and dropped after an outage. Adapters that buffer a server-side
            update queue (e.g. Telegram's Bot API) should preserve that queue
            when ``is_reconnect`` is True so messages sent during the outage
            are delivered rather than silently discarded.
    """
    pass

@abstractmethod
async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:

is_reconnect で再接続時にサーバー側の update キュー(Telegram のようなもの)を保留する。さもないと掉線中に送られたメッセージが黙って捨てられる。

具体的なアダプタ(Signal)は次の形だ。send は統一 content を signal-cli の RPC パラメータに翻訳する。

class SignalAdapter(BasePlatformAdapter):
    """Signal messenger adapter using signal-cli HTTP daemon."""

    platform = Platform.SIGNAL
    SUPPORTS_MESSAGE_EDITING = False  # Signal は送信済メッセージに edit API を持たない

    async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
        """Send a text message with native Signal formatting."""
        await self._stop_typing_indicator(chat_id)
        plain_text, text_styles = self._markdown_to_signal(content)
        params: Dict[str, Any] = {"account": self.account, "message": plain_text}
        if text_styles:
            params["textStyles"] = text_styles
        if chat_id.startswith("group:"):
            params["groupId"] = chat_id[6:]
        else:
            params["recipient"] = [await self._resolve_recipient(chat_id)]
        result = await self._rpc("send", params)

SUPPORTED_MESSAGE_EDITING = False のような能力ビットは、ストリーミング消費器に「送信済みメッセージを edit するな」と伝え、Signal クライアントに「編集失敗」のプレースホルダを残さないようにする。プラットフォーム特性の差はすべてアダプタのフィールドで表現し、if-else 分岐にはしない。

境界と失敗

• プラットフォームのオフライン/掉線:connect は is_reconnect でコールド起動と復旧を区別する。掉線中のユーザーメッセージはサーバー側に溜まることがある。アダプタが update キュー(Telegram)を保留しない、あるいは SSE 再接続(Signal)を持たない場合、メッセージは失われる。
• 重型 SDK の誤ロード:deferred の名前を登録し間違えるか、_resolve_all を呼ぶべきでないパスで呼ぶと、CLI chat が discord.py を引き込む。register を deferred より優先するのが安全網。
• 能力ビットの不一致:SUPPORTED_MESSAGE_EDITING=False なのにストリーミング消費器が edit パスを走らせると、クライアントに「編集失敗」のプレースホルダが残る。新規能力ビットはすべてのアダプタがデフォルト値を明示的に宣言しなければならない。
• 複数 profile での listener 重複:同じ Signal 番号を複数 profile で登録すると、SSE が二つ起動する。Signal は _acquire_platform_lock('signal-phone', self.account, ...) でプロセスレベルのロックを取り、重複を防ぐ。

まとめ

アダプタ = 「プラットフォーム方言 ↔ 統一イベント」の翻訳器。新規プラットフォームは BasePlatformAdapter を継承していくつか抽象メソッドを実装し登録するだけでよく、gateway/platforms/ADDING_A_PLATFORM.md を参照。Telegram/Discord/Slack はプラグインパスを通り、内蔵アダプタと同等に扱われる。