AI Agent Engineer

AI Agent Engineer

新增 Channel

6 min read

新增 Channel

对应源码

扩展环节源码
Channel 基类契约nanobot/channels/base.py::BaseChannel
发现和插件加载nanobot/channels/registry.py、nanobot/channels/manager.py
入站消息、安全和 Pairingnanobot/channels/manager.py、nanobot/pairing/store.py
出站、stream、reasoning 和 mediananobot/bus/events.py、nanobot/channels/manager.py、具体 Channel
配置接入nanobot/config/schema.py、nanobot/cli/commands.py::onboard
测试tests/channels/、tests/pairing/

学习目标

读完后，你应当能够：

  • 用 BaseChannel 实现最小入站/出站适配；
  • 通过 nanobot.channels entry point 发布外部 plugin；
  • 正确处理 allowlist、pairing、stream、reasoning、media 和关闭；
  • 用 fake platform client 测试，而不连接真实聊天平台。

前置知识

  • 理解 Channel 与 Gateway；
  • 能编写长运行异步任务；
  • 知道平台 user id、chat id 和 message id 不一定相同。

1. Channel 的最小职责

Channel 只做平台适配：

flowchart LR
    A["平台事件"] --> B["解析 sender/chat/content/media"]
    B --> C["BaseChannel._handle_message"]
    C --> D["InboundMessage"]
    D --> E["MessageBus"]
    E --> F["Agent core"]
    F --> G["OutboundMessage"]
    G --> H["Channel.send / send_delta"]
    H --> I["平台 API"]

它不应复制 AgentLoop、Session 或 Tool 调用逻辑。

2. BaseChannel 必需方法

子类必须实现：

  • async start()
  • async stop()
  • async send(msg)

start() 必须保持运行，直到 stop() 或 task 取消。如果建立连接后立即 return，ChannelManager 会认为该 Channel 已结束。

类属性至少定义：

name = "webhook"
display_name = "Webhook"

3. 入站必须经过 _handle_message

平台 callback 中应调用：

await self._handle_message(
    sender_id=sender,
    chat_id=chat,
    content=text,
    media=media,
    metadata=metadata,
    session_key=optional_override,
    is_dm=is_direct_message,
)

BaseChannel._handle_message() 会：

  1. 检查 allowFrom 或 pairing store；
  2. 未授权私聊可生成 pairing code；
  3. 根据 streaming 能力设置 _wants_stream；
  4. 构造 InboundMessage；
  5. 发布到 MessageBus。

直接 bus.publish_inbound() 会绕过这些共用行为，除非你有经过审查的特殊原因。

4. 身份建模

必须明确：

  • sender_id：谁发的；
  • chat_id：回复到哪里；
  • session_key：哪些消息共享会话；
  • message_id：引用/去重/线程；
  • group/thread/reply id：平台自己的层级。

常见错误是把 group chat id 当 user allowlist id，或让不同 thread 意外共享一个 Session。

用字符串保留平台原始稳定 id，不要把可能超出 JavaScript 安全整数的 id 转为 number。

5. allowlist 与 pairing

BaseChannel.is_allowed() 当前顺序：

  1. allowFrom 中有 "*"；
  2. sender exact match；
  3. pairing store 已批准；
  4. 拒绝。

allowFrom 条目是 opaque token，必须精确匹配。空列表会拒绝所有未配对用户；字段省略时，ChannelManager 记录 pairing-only 提示。

新增 Channel 应复用这层逻辑，并用平台真正的用户稳定 ID。

6. 出站与重试

send() 接收：

  • msg.content
  • msg.media
  • msg.chat_id
  • msg.metadata

投递失败应抛异常。ChannelManager._send_with_retry() 统一使用 1、2、4 秒退避，而不是要求每个 Channel 再套一层相同重试。

取消必须向上传播，保证 Gateway 可以正常关闭。

如果平台 API 自身已经有重试，要避免两个层级叠加导致重复消息。

7. Streaming

Channel 只有同时满足以下条件才宣告 streaming：

  1. config 的 streaming 为真；
  2. 子类覆盖 send_delta()。

处理 metadata：

  • _stream_delta
  • _stream_end
  • _stream_id
  • _resuming

有状态 buffer 必须按 _stream_id，不能只按 chat_id。同一 chat 可能并发多个 segment、恢复或多标签页镜像。

平台若不适合频繁编辑，可不开 streaming，直接使用 send()。

8. Progress、Tool hint、Reasoning 和文件编辑

可选表面：

  • send_progress
  • send_tool_hints
  • send_reasoning_delta()
  • send_reasoning_end()
  • send_file_edit_events()

这些是低优先级观察信息。平台不支持时可安全忽略，不应把 reasoning 当普通最终回复发送。

ChannelManager 还会合并连续 text delta，减少平台 API 调用。

9. Media

Channel media 代码需要单独审查：

  • 上传大小和类型；
  • 临时文件清理；
  • 远程 URL 下载的 SSRF；
  • 文件名路径穿越；
  • MIME 与扩展名不一致；
  • 平台下载 token；
  • 出站本地文件是否在允许 media root；
  • Markdown 到平台格式的转义。

参考现有专项测试，而不要只复制 happy path。

10. 外部 plugin 优先

官方推荐独立 package：

[project.entry-points."nanobot.channels"]
webhook = "nanobot_channel_webhook:WebhookChannel"

ChannelsConfig 允许 extra section；ChannelManager 从内置模块名和配置 extra keys 中找 enabled Channel。

发现规则：

  • 内置 Channel 用 pkgutil 扫描；
  • plugin 用 nanobot.channels entry points；
  • 只 import enabled 内置模块，减少第三方 SDK 启动成本；
  • 同名时内置 Channel 优先，外部 plugin 不能 shadow。

default_config() 可让 nanobot onboard 生成初始配置。

11. 最小测试集

  1. registry 能发现 entry point；
  2. disabled config 不实例化；
  3. start() 启动与 stop() 清理；
  4. allowed sender 产生正确 InboundMessage；
  5. denied/pairing 行为；
  6. send() 目标和格式；
  7. delivery failure 由 manager 重试；
  8. streaming delta/end 与 _stream_id；
  9. media 安全；
  10. cancellation 传播。

测试参考：

  • tests/channels/test_base_channel.py
  • tests/channels/test_channel_plugins.py
  • tests/channels/test_channel_manager_delta_coalescing.py
  • tests/channels/test_channel_manager_reasoning.py
  • 一个协议相近的现有 Channel test。

官方完整示例：

12. 最小修改范围

外部 plugin：

  • 独立 package 与 config model；
  • BaseChannel 子类；
  • entry point；
  • tests/readme。

内置 Channel 还可能需要：

  • optional dependency；
  • 用户配置文档；
  • onboarding；
  • WebUI settings；
  • 平台专项安全测试。

不要仅为了实验就修改 ChannelsConfig 的固定字段；当前 extra config 已为 plugin 留出边界。

自检

  1. 为什么平台 callback 应调用 _handle_message()？
  2. sender_id 与 chat_id 混用会造成什么问题？
  3. Channel 的 start() 为什么不能在连接成功后 return？
  4. 外部 plugin 与内置 Channel 同名时谁优先？

下一篇：命令、API、SDK 与 WebUI 扩展边界

Backlinks