第五步：编写一个可用适配器¶

前几章都在写插件。插件只关心 Event、Context 和 Message，不应该知道外部平台怎么鉴权、怎么拉消息、怎么发 API 请求。适配器就是这层协议边界。

这一章用内置 TelegramAdapter 作为完整例子。它是真正可运行的适配器：通过 Telegram Runtime API getUpdates 长轮询接收消息，通过 sendMessage 发送回复。

配置入口¶

先给 Runtime 加一个适配器配置：

[runtime]
adapters = ["telegram"]

[adapter.telegram]
token = "123456:replace-me"
poll_timeout = 30
allowed_updates = ["message"]

对应的配置模型放在 iamai.config。这样做的目的不是形式主义，而是让启动前校验、管理命令和配置参考都能看到同一份结构：

class TelegramConfigModel(BaseModel):
    token: str = ""
    api_base_url: str = "https://api.telegram.org"
    poll_timeout: int = 30
    request_timeout: float = 40.0
    reconnect_interval: float = 3.0
    limit: int = 100
    allowed_updates: list[str] = Field(default_factory=lambda: ["message"])

内置适配器还需要加入 BUILTIN_ADAPTERS，这样用户可以写 adapters = ["telegram"]。第三方适配器也可以不注册内置名，直接在配置里写导入路径，例如 "my_pkg.telegram:TelegramAdapter"。

适配器骨架¶

适配器至少实现 start 和 send_message，通常也会实现 close 和 call_api：

from iamai import Adapter, Event, Message


class TelegramAdapter(Adapter):
    name = "telegram"

    async def start(self) -> None:
        ...

    async def close(self) -> None:
        ...

    async def send_message(
        self,
        message: Message,
        *,
        event: Event | None = None,
        target: object | None = None,
    ) -> object:
        ...

name 必须稳定。它会用于配置表 [adapter.telegram]、运行时指标、审计日志和 event.adapter。

接收循环¶

Telegram 长轮询的核心是 getUpdates。适配器维护 offset，每处理一个 update 就把 offset 推进，避免重启前后重复消费同一批消息。

async def start(self) -> None:
    while not self._closed.is_set():
        try:
            updates = await self._get_updates()
            for update in updates:
                self._offset = update["update_id"] + 1
                event = self._normalize_update(update)
                if event is not None:
                    await self.emit(event)
        except Exception:
            await asyncio.sleep(self.reconnect_interval)

生产适配器必须处理失败：网络错误、平台超时、无效 payload、重复 update、关闭信号。不要让一次失败退出整个 Runtime，除非这是明确的配置错误，例如 token 缺失。

事件归一化¶

平台 payload 要尽早变成 iamai 的 Event。Telegram 的关键字段映射是：

message.text 或 message.caption -> Message。
message.from.id -> user_id。
message.chat.id -> channel_id。
群组和频道的 chat.id -> guild_id。
原始 update -> event.raw。

def _normalize_update(self, update: Mapping[str, Any]) -> Event | None:
    message = update.get("message")
    if not isinstance(message, Mapping):
        return None
    chat = message.get("chat")
    if not isinstance(chat, Mapping):
        return None
    return Event(
        id=str(message.get("message_id") or update.get("update_id")),
        adapter=self.name,
        platform="telegram",
        type="message",
        detail_type=str(chat.get("type", "private")),
        channel_id=str(chat["id"]),
        message=Message(str(message.get("text") or "")),
        raw=dict(update),
    )

这个边界很重要：插件应该看到稳定的 Event，而不是到处读 update["message"]["chat"]["id"]。平台差异留在适配器里。

发送消息和 API¶

send_message 要支持两种调用方式：

ctx.reply("pong")：通过当前 event 推断目标。
ctx.send("pong", target=12345)：显式指定平台目标。

Telegram 的实现最终调用 sendMessage：

async def send_message(self, message: Message, *, event=None, target=None):
    chat_id = self._resolve_chat_id(event=event, target=target)
    return await self.call_api(
        "sendMessage",
        chat_id=chat_id,
        text=Message.ensure(message).plain_text(),
    )

call_api 是适配器给高级插件留下的平台逃生口。它应该处理平台失败响应，并返回平台 result：

async def call_api(self, action: str, **params: Any) -> Any:
    response = await request_json(self._method_url(action), json_body=params)
    if not response["ok"]:
        raise RuntimeError(response.get("description", "request failed"))
    return response.get("result")

测试适配器¶

适配器测试不应该真的访问 Telegram。把 HTTP helper monkeypatch 掉，就能验证参数和错误处理：

async def fake_request_json(url: str, **kwargs):
    return {"ok": True, "result": {"message_id": 1}}

monkeypatch.setattr("iamai.adapters.telegram.request_json", fake_request_json)

再单独测试 _normalize_update，确保字段映射不会回退：

event = adapter._normalize_update({
    "update_id": 10,
    "message": {
        "message_id": 20,
        "from": {"id": 30},
        "chat": {"id": -40, "type": "group"},
        "text": "hello",
    },
})

assert event.channel_id == "-40"
assert event.text == "hello"

Checkpoint¶

写适配器时按这个顺序自查：

配置能被 load_config 校验，敏感字段会被 redaction。
start 能持续接收事件，失败后能重试，close 能停下来。
inbound payload 会变成稳定 Event，原始数据只放在 raw。
send_message 同时支持 event 和显式 target。
call_api 暴露平台通用能力，并清晰处理失败。
不访问真实外部服务也能测试字段映射和出站参数。