如何将(几乎)任意Agent接入你的微信

基于 ilinkai.weixin.qq.com HTTP API 的实现指南

背景

今天微信终于开放了Openclaw的入口,或者说是任意Agent/机器人的接口。这一天终于还是来了。

在2023-2024年ChatGPT最火的时候,笔者曾经研究过如何在微信中与chatbot聊天。早期方案多基于 Web 微信协议(如 itchat),通过模拟 Web 端登录实现消息收发。2017 年后微信逐步收紧 Web 端权限,大量账号被限制登录,相关项目纷纷停止维护。另一条路是 PC 客户端注入(如 WeChatPYAPI、wxauto),通过 Hook 微信进程直接调用内部函数,但存在封号风险,且每次微信更新都需要适配,维护成本极高。更激进的方案如 OpenWx 尝试逆向微信网络协议,最终作者收到律师函,项目删除收场。

笔者之前唯一跑通过的是企业微信的客服渠道。这也是 qclaw 首先实现微信打通的方式——通过企业微信的客服功能作为消息中转。但这种方式终究是"曲线救国",并非微信官方认可的个人号接入渠道。毕竟qclaw团队也不是官方。

今天开放的 ilink API 是基于 ilinkai.weixin.qq.com 的 HTTP long-poll 接口(ilink 协议本身可能更早存在——用于IoT设备接入,只是今天才开放给个人号 Bot 使用)。无需 WebSocket、无需本地客户端、无需注入,只需扫码获取 bot_token 即可。

这标志着微信战略的重大转变。十多年来微信一直拒绝开放个人号 API,如今终于松动。旧世界正在崩塌——正如多年前那句预言:"打败微信的不会是另一个微信"。这次开放,微信能否借助霸主地位一统 Agent 入口、杀死游戏?还是迫于压力的自毁长城,导致机器人泛滥稀释了熟人社交的体验?又或者用户终于发现,新时代的不需要微信这样一个超级App把大家圈在里面?我们拭目以待。


协议概览

┌─────────────┐      HTTP/JSON       ┌─────────────────────┐
│   Agent     │ ◄──────────────────► │ ilinkai.weixin.qq.com│
│  (你的实现)  │   long-poll + send   │    (腾讯官方 API)    │
└─────────────┘                       └─────────────────────┘

核心特点: 纯 HTTP、Long-poll 轮询、bot_token 持久化、支持文本/图片/语音/文件/视频


1. 认证

获取二维码 → 轮询状态 → 保存 token

# 1. 获取登录二维码
resp = await api_post("ilink/bot/get_bot_qrcode", {"bot_type": 3})
qrcode_url = resp["qrcode_url"]  # 生成二维码让用户扫码
bot_id = resp["bot_id"]

# 2. 轮询登录状态
while True:
    resp = await api_post("ilink/bot/get_qrcode_status", {"bot_id": bot_id})
    if resp["status"] == 1:
        bot_token = resp["bot_token"]
        break
    await asyncio.sleep(2)

# 3. 持久化
save_state({"bot_token": bot_token, "bot_id": bot_id})

2. 消息接收

Long-poll 轮询,超时约 35 秒:

while running:
    resp = await api_post("ilink/bot/getupdates", {
        "get_updates_buf": sync_buf,
        "base_info": {"channel_version": "1.0.2"}
    })
    sync_buf = resp.get("get_updates_buf", "")
    for msg in resp.get("msgs", []):
        await handle_message(msg)

消息格式:

{
  "from_user_id": "wxid_xxx",
  "to_user_id": "bot_id@im.bot",
  "context_token": "用于回复",
  "item_list": [
    {"type": 1, "text_item": {"text": "你好"}}
  ]
}

item_list 类型:

type 字段 说明
1 text_item 文本
2 image_item 图片
3 voice_item 语音(含语音转文字)
4 file_item 文件
5 video_item 视频

3. 消息发送

await api_post("ilink/bot/sendmessage", {
    "msg": {
        "from_user_id": "",
        "to_user_id": to_user_id,
        "client_id": f"nanobot-{uuid.uuid4().hex[:12]}",
        "message_type": 2,  # Bot 发出
        "message_state": 2,  # Finish
        "context_token": context_token,  # 必须用收到的 token
        "item_list": [{"type": 1, "text_item": {"text": content}}]
    },
    "base_info": {"channel_version": "1.0.2"}
})

关键点:


4. 媒体接收(图片/语音/文件/视频)

用户发送的媒体需从 CDN 下载并 AES 解密:

CDN_BASE = "https://novac2c.cdn.weixin.qq.com/c2c"

async def download_media(item: dict) -> bytes:
    media = item.get("media", {})
    param = media.get("encrypt_query_param", "")
    
    # 解析 AES 密钥(两种格式)
    if item.get("aeskey"):  # 32位hex
        key = bytes.fromhex(item["aeskey"])
    else:  # base64
        raw = base64.b64decode(media.get("aes_key", ""))
        key = bytes.fromhex(raw.decode()) if len(raw) == 32 else raw
    
    # 下载并解密
    url = f"{CDN_BASE}/download?encrypted_query_param={quote(param)}"
    data = (await httpx.get(url)).content
    
    from Crypto.Cipher import AES
    return AES.new(key, AES.MODE_ECB).decrypt(data)

依赖: pip install pycryptodome


5. 错误处理

errcode 含义 处理
0 成功 -
-14 session 过期 重新扫码登录
-6 context_token 无效 等待新消息获取 token

连续失败 3 次后进入 30 秒退避,避免频繁请求。


6. 实现参考