CoPaw 微信个人号渠道集成指南

前阵子想把 CoPaw 接到微信个人号上，折腾了几天总算跑通了。这个过程里踩了不少坑，也积累了一些经验，所以想着干脆整理成一篇文章，既给自己留个记录，也方便后来者少走点弯路。

说实话，微信个人号的集成比企业号要复杂不少，主要是因为微信对自动化操作的限制比较严格。好在腾讯官方提供了一个 @tencent-weixin/openclaw-weixin 的 npm 包，给了我不少启发。不过这个包是用 TypeScript 写的，而 CoPaw 的后端是 Python，所以只能参考它的接口设计和消息流程，具体实现还得自己来。

整篇文章按这个逻辑展开：先说整体架构，再讲具体实现，最后聊聊那些容易踩的坑。

技术选型与架构

为什么选 iLink API

微信个人号接入，本质上就是和微信的 iLink API 打交道。这个 API 的域名是 https://ilinkai.weixin.qq.com，另外还有 CDN 端点 https://novac2c.cdn.weixin.qq.com/c2c 用来处理媒体文件。

整个交互流程其实不复杂：扫码登录拿到 token，然后用长轮询的方式拉取消息，处理完后再调接口回复。但里面有些细节需要注意。

几个关键技术点

做之前得先理解几个核心概念，不然写代码的时候容易懵：

消息加密用的是 AES-128-ECB，需要 PKCS7 填充。这个和常见的 AES-256-CBC 不太一样，一开始我直接用默认的 AES 实现，解密总是报错，后来才发现是算法和填充方式的问题。

context_token 是个很重要的东西。每条消息都会带这个 token，你回复的时候必须原样传回去，不然微信会认为这是不合法的请求。所以我在本地做了一个 context token 的持久化，按用户 ID 索引，每次发送消息时查表获取。

bot_type 这个参数，文档里说是 "3"，代表当前渠道的构建版本。这个保持默认就行，但要注意如果微信那边升级了协议，可能需要跟着改。

长轮询机制是微信推送消息的方式。调用 getupdates 接口后，服务端会挂起一段时间（默认 35 秒），有新消息就立即返回，没消息就等超时。所以代码里捕获超时异常是正常的，不代表出错了。

Session 过期的问题比较头疼。错误码 200011 表示 token 失效了，这时候只能引导用户重新扫码。我在代码里做了自动检测，遇到这个错误就暂停轮询，等用户手动触发登录。

消息流转示意

整个消息流转的过程大概是这样的：

微信用户 → iLink API (长轮询) → CoPaw WeixinChannel 
                                ↓
                         ChannelManager (消息入队)
                                ↓
                           Agent 处理
                                ↓
                    WeixinChannel.send() 回复
                                ↓
                         iLink API (sendmessage)

核心实现

代码结构

我把微信相关的代码都放在了 src/copaw/app/channels/weixin/ 这个目录下，按功能拆成了几个模块：

src/copaw/app/channels/weixin/
├── __init__.py          # 模块导出
├── compat.py            # 版本兼容性检查
├── types.py             # 消息类型定义
├── account.py           # 账号管理
├── api.py               # API 客户端封装
├── auth.py              # 扫码登录
├── cdn.py               # 媒体文件上传下载
└── channel.py           # 渠道主入口

这么拆分的原因主要是想让每个模块的职责尽可能单一。比如 account.py 只管账号数据的读写，api.py 只负责 HTTP 请求，channel.py 则是把前面这些模块串起来，处理消息的接收和发送。

注册渠道

写完之后还得告诉 CoPaw 有这么个渠道。在 src/copaw/app/channels/registry.py 里加一行注册就行：

_BUILTIN_SPECS: dict[str, tuple[str, str]] = {
    # ... 其他渠道
    "weixin": (".weixin", "WeixinChannel"),
}

依赖管理

pyproject.toml 里记得加依赖，主要是 qrcode 用来在终端打印二维码：

dependencies = [
    # ... 其他依赖
    "qrcode>=7.4.0",
]

关键模块详解

类型定义（types.py）

消息类型定义这块，我参考了官方的 TypeScript 实现，把消息结构和枚举值都对应上了。核心是 WeixinMessage 这个 dataclass：

@dataclass
class WeixinMessage:
    """微信消息。"""
    from_user_id: Optional[str] = None
    to_user_id: Optional[str] = None
    message_type: Optional[int] = None
    message_state: Optional[int] = None
    context_token: Optional[str] = None
    item_list: Optional[list[MessageItem]] = None
    seq: Optional[int] = None
    message_id: Optional[str] = None
    create_time_ms: Optional[int] = None

这里最容易出错的是 message_type 和 message_state。BOT 类型的消息 message_type 是 3，消息结束状态 message_state 是 5。这些数字都是微信那边定义好的，不能随便改。

API 客户端（api.py）

API 客户端封装了所有 HTTP 请求。有个细节是请求头的构造，尤其是 X-WECHAT-UIN 这个字段，需要生成一个随机值：

@staticmethod
def _random_wechat_uin() -> str:
    """生成随机 X-WECHAT-UIN 头。"""
    import base64
    uint32 = secrets.randbits(32)
    return base64.b64encode(str(uint32).encode()).decode()

这个字段是微信用来标识客户端的，虽然是随机的，但格式必须符合他们的要求（32位无符号整数的 base64 编码）。

长轮询的实现也挺有意思：

async def get_updates(
    self,
    get_updates_buf: str = "",
    timeout_ms: Optional[int] = None,
) -> GetUpdatesResp:
    """长轮询获取新消息。"""
    body = json.dumps({
        "get_updates_buf": get_updates_buf,
        "base_info": {"channel_version": "copaw-weixin-1.0.0"},
    })

    timeout = timeout_ms or self.long_poll_timeout_ms

    try:
        raw_text = await self._api_fetch(
            "ilink/bot/getupdates",
            body,
            timeout_ms=timeout,
        )
        data = json.loads(raw_text)
        return parse_get_updates_resp(data)
    except (asyncio.TimeoutError, httpx.ReadTimeout):
        # 长轮询超时是正常的
        return GetUpdatesResp(ret=0, get_updates_buf=get_updates_buf)

注意这里捕获了两种超时异常：asyncio.TimeoutError 和 httpx.ReadTimeout。一开始我只捕获了前者，结果在 httpx 版本升级后就开始报错，后来才发现 httpx 自己也会抛超时异常。

扫码登录（auth.py）

登录流程是最复杂的部分。整个过程分几步：

调 get_bot_qr_code 获取二维码链接
在终端打印二维码（用了 qrcode 库生成 ASCII 艺术）
轮询 get_qrcode_status 等待用户扫码
拿到 bot_token 后保存账号信息

async def login_with_qr(
    base_url: Optional[str] = None,
    account_id: Optional[str] = None,
    timeout_ms: int = DEFAULT_LOGIN_TIMEOUT_MS,
) -> WeixinQrWaitResult:
    base_url = base_url or DEFAULT_BASE_URL

    # 1. 获取二维码
    qr_response = await fetch_qr_code(base_url, bot_type="3")
    qrcode_url = qr_response.get("qrcode_img_content")

    if not qrcode_url:
        return WeixinQrWaitResult(message="获取二维码失败")

    # 2. 显示二维码
    print("\n使用微信扫描以下二维码：\n")
    print_qr_code(qrcode_url)

    # 3. 等待扫码
    qrcode = qr_response.get("qrcode")
    result = await poll_qr_status(base_url, qrcode, timeout_ms)

    # 4. 保存账号
    if result.connected and result.bot_token and result.account_id:
        save_weixin_account(
            result.account_id,
            token=result.bot_token,
            base_url=result.base_url,
            user_id=result.user_id,
        )
        register_weixin_account_id(result.account_id)

        # 更新所有 workspace 的配置文件
        update_workspace_weixin_account_id(result.account_id)

        print("\n✅ 与微信连接成功！")

    return result

登录超时我设的是 8 分钟（480000 毫秒），因为二维码本身有有效期，过期就得重新生成。实际使用中，大部分用户 1-2 分钟就能完成扫码。

还有个实用的小功能是登录后自动更新所有 workspace 的 agent.json 配置文件，这样用户就不用手动改 account_id 了。

渠道主类（channel.py）

WeixinChannel 是核心入口，继承自 BaseChannel。启动时会做几件事：

检查是否有账号配置
加载并恢复 context tokens
创建 API 客户端
启动长轮询监控循环

async def start(self) -> None:
    """启动渠道。"""
    if not self.enabled:
        return

    if not self._enqueue:
        logger.error("❌ _enqueue callback not set!")
        return

    account_ids = list_weixin_account_ids()
    if not account_ids:
        logger.warning("No Weixin accounts. Run: copaw weixin login")
        return

    account_id = self.account_id or account_ids[0]
    self._account = resolve_weixin_account(account_id)

    if not self._account.configured:
        logger.warning(f"Account {account_id} not configured. Run: copaw weixin login")
        return

    restore_context_tokens(account_id)

    self._client = WeixinApiClient(
        base_url=self._account.base_url,
        token=self._account.token,
    )

    self._abort_event.clear()
    self._monitor_task = asyncio.create_task(self._monitor_loop())

    logger.info(f"✅ WeixinChannel 已启动: account={account_id}")

消息处理这块，收到微信消息后会构建一个标准化的 native 格式入队：

native = {
    "channel_id": self.channel,
    "sender_id": from_user_id,
    "session_id": f"weixin:{from_user_id}",
    "content_parts": content_parts,
    "meta": {
        "weixin_from_user_id": from_user_id,
        "weixin_context_token": msg.context_token,
    },
}

这里把 context_token 放在了 meta 里，后续 Agent 处理完需要回复时，渠道就能从 meta 中取出来用。

配置与部署

环境变量

微信渠道支持几个环境变量来控制行为：

变量	说明	默认值
`WEIXIN_CHANNEL_ENABLED`	是否启用渠道	自动检测（有账号配置就启用）
`WEIXIN_BOT_PREFIX`	消息前缀	空
`WEIXIN_BASE_URL`	API 基础 URL	`https://ilinkai.weixin.qq.com`
`WEIXIN_ACCOUNT_ID`	指定账号 ID	使用第一个账号

其实 WEIXIN_CHANNEL_ENABLED 不设置也行，代码会自动检测有没有登录过账号，有的话就默认启用。

CLI 命令

为了方便操作，我加了几个命令行工具：

# 登录（最常用）
copaw weixin login

# 查看状态
copaw weixin status

# 列出所有账号
copaw weixin accounts

也可以通过通用的渠道命令操作：

# 登录
copaw channels login --channel weixin

# 配置
copaw channels config weixin --enabled true

账号数据存储

登录后的账号数据会保存在 ~/.copaw/weixin/ 目录下：

~/.copaw/weixin/
├── accounts.json           # 账号索引，记录有哪些账号
└── accounts/
    └── xxx-im-bot.json     # 具体账号数据（token、user_id 等）

每个账号的 JSON 文件长这样：

{
  "token": "xxx@im.bot:xxx",
  "base_url": "https://ilinkai.weixin.qq.com",
  "user_id": "xxx@im.wechat",
  "saved_at": "2026-03-25T00:00:00"
}

token 是最关键的，有了它就能代表用户调用 API。saved_at 主要用来排查问题，看看是什么时候登录的。

常见问题排查

渠道未启动

如果你看到日志里打印 WeixinChannel is disabled，通常是两种情况：要么确实没启用，要么是没检测到账号配置。

最快的解决办法是先跑一次登录：

copaw weixin login

如果就是想用环境变量控制，可以设置：

export WEIXIN_CHANNEL_ENABLED=1

账号未配置

日志提示 Account xxx not configured，说明配置文件里的 account_id 和你实际登录的不一致。

这一般发生在换了个账号登录之后。新账号登录会生成新的 account_id，但配置文件还是旧的。解决方法就是重新登录，登录流程会自动更新所有 workspace 的配置文件。

Session 过期

微信的登录态不是永久有效的，过段时间就会过期。表现是日志里频繁打印 Session expired 或者错误码 200011。

这个没办法自动恢复，只能重新扫码登录：

copaw weixin login

后续可以考虑加个自动续期的机制，在 session 快过期前提醒用户。

长轮询超时

如果日志里出现 httpx.ReadTimeout 或者 asyncio.TimeoutError，先别紧张。长轮询的机制就是这样，服务端会挂起一段时间（35 秒左右），没新消息就超时返回。

关键是要捕获正确的异常：

except (asyncio.TimeoutError, httpx.ReadTimeout):
    # 长轮询超时是正常的
    pass

一开始我只捕获了 asyncio.TimeoutError，后来 httpx 升级后就开始报错。所以建议两个都捕获，比较稳妥。

消息发了没回复

这种情况排查起来稍微麻烦点，因为涉及整个处理链路。可以按这个顺序检查：

确认消息是否收到了：看日志里有没有 📥 收到微信消息 这行
确认消息是否入队：检查 _enqueue 回调是否被调用
确认 Agent 是否处理了：看 Agent 相关日志有没有报错
确认回复是否发送了：看日志里有没有 📤 发送微信消息

如果消息收到了但没回复，大概率是 Agent 处理环节的问题，比如模型配置不对或者工作流有 bug。

getUpdates 返回 ret=None

有时候日志会打印 getUpdates failed: ret=None errcode=None，看着像失败了但其实不是。

这是因为错误检查逻辑写得不够严谨。正确的判断方式应该是：

is_api_error = (resp.ret is not None and resp.ret != 0) or (
    resp.errcode is not None and resp.errcode != 0
)

注意要先判断 is not None，不然可能会把 0（成功）当成错误。

参考资源

核心文件

文件	职责
`channel.py`	渠道入口，消息接收和发送
`api.py`	HTTP 请求封装
`auth.py`	扫码登录流程
`account.py`	账号数据管理
`types.py`	消息类型定义
`weixin_cmd.py`	命令行工具

后续优化方向

目前这个实现已经能满足基本需求了，但还有一些可以优化的地方：

图片消息支持：现在只处理了文本消息，图片和文件的收发还没做。主要是涉及到 CDN 的上传下载和 AES 解密，复杂度会高一些。

自动续期：现在 session 过期了只能手动重新登录，后续可以考虑在快过期前主动提醒用户，或者尝试自动刷新 token。

多账号支持：目前只用了第一个账号，后续可以扩展成支持同时登录多个微信账号。

更好的错误恢复：网络抖动导致的临时错误可以做自动重试，减少人工干预。

如果你有好的想法或者遇到了什么问题，欢迎交流。