不用公众号、不用企业微信、不用第三方服务,纯 Python 对接微信 AI Bot 平台,实现消息收发。
本文记录了从 npm 包源码逆向协议、踩坑 context_token、到发现"幽灵字段"导致消息静默丢失的完整过程。

0. 故事的起点

我在做一个量化选股系统 (QuantByQlib),跑完 6-Agent 分析后想把结果推送到微信。

需求很简单:程序跑完 → 自动发到我微信。

但现实很骨感:

方案 问题
微信公众号 需要企业认证,个人订阅号不能主动推
企业微信 webhook 要开通企业微信
Server酱 免费 5 条/天
第三方框架 (itchat等) 2024 年全部阵亡,微信封杀 web 协议

然后我发现了 OpenClaw —— 一个开源 AI 助手框架,它有个官方微信插件 @tencent-weixin/openclaw-weixin,声称扫码即可登录,支持消息收发。

关键是,这个插件是腾讯官方发布的,用的是微信内部的 ilink AI Bot 平台接口。

我的想法是:不装 OpenClaw,直接把协议扒出来,用 Python 复刻。

1. 从 npm 包逆向协议

# 先看看这个包里有什么
curl -s https://unpkg.com/@tencent-weixin/openclaw-weixin@1.0.3/?meta | python -m json.tool

惊喜: 源码是 TypeScript 原文发布的,没混淆、没打包。41 个文件,结构清晰:

src/
├── api/
│   ├── api.ts          ← HTTP 请求层 (5个接口)
│   ├── types.ts        ← 完整类型定义
│   └── session-guard.ts
├── auth/
│   ├── login-qr.ts     ← 扫码登录流程
│   └── accounts.ts     ← 账号持久化
├── messaging/
│   ├── inbound.ts      ← 消息接收 + context_token 管理
│   ├── send.ts         ← 消息发送
│   └── process-message.ts  ← 完整处理链路
├── cdn/
│   ├── aes-ecb.ts      ← AES 加密
│   └── cdn-upload.ts   ← 媒体上传
└── channel.ts          ← 插件主入口

我花了一个晚上通读了所有源码,梳理出了完整协议。

2. 协议全貌: 5 个 HTTP 接口搞定一切

所有接口都是 POST JSON,基地址 https://ilinkai.weixin.qq.com。

通用请求头

headers = {
    "Content-Type": "application/json",
    "AuthorizationType": "ilink_bot_token",       # 固定值
    "Authorization": f"Bearer {bot_token}",        # 扫码获取
    "X-WECHAT-UIN": base64(random_uint32),         # 随机生成
    "Content-Length": str(len(body_bytes)),         # 必须精确
}

接口列表

接口 路径 用途
getUpdates /ilink/bot/getupdates 长轮询收消息
sendMessage /ilink/bot/sendmessage 发消息
getUploadUrl /ilink/bot/getuploadurl CDN 上传
getConfig /ilink/bot/getconfig 获取 typing ticket
sendTyping /ilink/bot/sendtyping "正在输入"状态

另外还有两个登录接口 (不在 bot 路径下):

  • GET /ilink/bot/get_bot_qrcode?bot_type=3 → 获取二维码
  • GET /ilink/bot/get_qrcode_status?qrcode=xxx → 轮询扫码状态

3. 扫码登录: 60 行 Python 搞定

import httpx, qrcode, time

BASE = "https://ilinkai.weixin.qq.com"

# Step 1: 获取二维码
resp = httpx.get(f"{BASE}/ilink/bot/get_bot_qrcode?bot_type=3")
data = resp.json()
qrcode_key = data["qrcode"]
qrcode_url = data["qrcode_img_content"]

# Step 2: 终端显示二维码
qr = qrcode.QRCode(border=1)
qr.add_data(qrcode_url)
qr.make(fit=True)
qr.print_ascii(invert=True)

# Step 3: 长轮询等扫码确认
while True:
    status_resp = httpx.get(
        f"{BASE}/ilink/bot/get_qrcode_status?qrcode={qrcode_key}",
        headers={"iLink-App-ClientVersion": "1"},
        timeout=40,
    )
    status = status_resp.json()

    if status["status"] == "scaned":
        print("已扫码,请在手机上确认...")
    elif status["status"] == "confirmed":
        bot_token = status["bot_token"]
        account_id = status["ilink_bot_id"]
        user_id = status["ilink_user_id"]
        print(f"登录成功! token={bot_token[:20]}...")
        break
    elif status["status"] == "expired":
        print("二维码过期,请重新获取")
        break

扫码后你会得到三个关键值:

  • bot_token — 后续所有 API 的认证令牌
  • ilink_bot_id — Bot 的账户 ID
  • ilink_user_id — 扫码人的微信 ID (格式: xxx@im.wechat)

4. 第一个大坑: 消息发送成功但收不到

拿到 token 后,我写了最简单的发送:

# ❌ 错误的写法 — API 返回 200 但消息不投递
resp = httpx.post(f"{BASE}/ilink/bot/sendmessage", json={
    "msg": {
        "to_user_id": user_id,
        "context_token": saved_context_token,
        "item_list": [{"type": 1, "text_item": {"text": "Hello!"}}],
    }
}, headers=headers)

print(resp.status_code)  # 200
print(resp.text)          # {}
# 微信上: 啥也没收到

HTTP 200,空响应体 {}。没有错误码,没有错误信息,就是收不到。

这是最阴险的 bug —— 静默失败。

我排查了两天:

  1. token 过期? 不是,getUpdates 正常
  2. context_token 问题? 换了新的也不行
  3. user_id 错误? 就是扫码返回的那个

5. 幽灵字段: 逆向发现的真相

最终我回到 OpenClaw 源码,逐字对比 send.ts 里的请求构造:

// OpenClaw 的 buildTextMessageReq (src/messaging/send.ts)
function buildTextMessageReq(params) {
    return {
        msg: {
            from_user_id: "",           // ← 空字符串,不是不传
            to_user_id: to,
            client_id: clientId,        // ← 每条消息唯一 ID !!!
            message_type: 2,            // ← MessageType.BOT !!!
            message_state: 2,           // ← MessageState.FINISH !!!
            item_list: [...],
            context_token: contextToken,
        },
    };
}

然后看 api.ts 的发送函数:

// src/api/api.ts
export async function sendMessage(params) {
    await apiFetch({
        baseUrl: params.baseUrl,
        endpoint: "ilink/bot/sendmessage",
        // 注意这里: 每个请求都附带 base_info !!!
        body: JSON.stringify({ ...params.body, base_info: buildBaseInfo() }),
        token: params.token,
        timeoutMs: params.timeoutMs ?? 15_000,
        label: "sendMessage",
    });
}

function buildBaseInfo() {
    return { channel_version: "1.0.3" };  // ← 版本标识
}

我们漏了 4 个字段:

字段 值 作用
from_user_id "" 空字符串,标记发送方
client_id UUID 每条消息唯一ID,服务端用于去重和路由
message_type 2 标记为 BOT 消息 (1=用户, 2=Bot)
message_state 2 标记为完成态 (0=新建, 1=生成中, 2=完成)

以及请求体顶层的:

字段 值 作用
base_info.channel_version "1.0.3" 插件版本标识

这些字段不在官方文档里 (README 只写了 to_user_id, context_token, item_list),但服务端依赖它们做消息路由。缺少任何一个,消息就被静默丢弃。

6. 正确的发送格式

import uuid

def send_message(token, to_user_id, text, context_token):
    """能实际投递的消息发送"""
    body = {
        "msg": {
            "from_user_id": "",
            "to_user_id": to_user_id,
            "client_id": f"mybot-{uuid.uuid4().hex[:12]}",
            "message_type": 2,        # BOT
            "message_state": 2,       # FINISH
            "context_token": context_token,
            "item_list": [
                {"type": 1, "text_item": {"text": text}}
            ],
        },
        "base_info": {"channel_version": "1.0.3"},
    }

    raw = json.dumps(body, ensure_ascii=False)
    headers = {
        "Content-Type": "application/json",
        "AuthorizationType": "ilink_bot_token",
        "Authorization": f"Bearer {token}",
        "X-WECHAT-UIN": base64.b64encode(
            str(random.randint(0, 0xFFFFFFFF)).encode()
        ).decode(),
        "Content-Length": str(len(raw.encode("utf-8"))),
    }

    resp = httpx.post(
        "https://ilinkai.weixin.qq.com/ilink/bot/sendmessage",
        content=raw.encode("utf-8"),
        headers=headers,
        timeout=15,
    )
    return resp.status_code == 200

7. 第二个大坑: context_token 是什么?

context_token 是 ilink 协议的会话上下文令牌。每次用户给 Bot 发消息时,getUpdates 返回的消息体里都带有一个 context_token。

{
    "msgs": [{
        "from_user_id": "xxx@im.wechat",
        "context_token": "AARzJW...(很长的base64)...",
        "item_list": [{"type": 1, "text_item": {"text": "你好"}}]
    }],
    "get_updates_buf": "CgkI..."
}

关键问题: 没有 context_token 能不能发?

答案: API 不报错 (返回 200),但消息不投递。必须有 context_token。

那 context_token 会过期吗?

这是我踩的第二个坑。一开始我以为 context_token 是一次性的,因为:

  • 用 context_token 发第一条消息 → 收到了
  • 同一个 token 发第二条 → 收不到

但真相是: context_token 可以无限复用,收不到是因为第一条发送的格式就不对!

当我补全了 client_id、message_type、message_state 之后,同一个 context_token 连发 10 条都能收到。

OpenClaw 的源码也证实了这一点 —— 在 inbound.ts 里,context_token 是持久化存储的:

// src/messaging/inbound.ts
const contextTokenStore = new Map();  // 内存缓存

export function setContextToken(accountId, userId, token) {
    contextTokenStore.set(`${accountId}:${userId}`, token);
    persistContextTokens(accountId);  // 同时写磁盘
}

export function getContextToken(accountId, userId) {
    return contextTokenStore.get(`${accountId}:${userId}`);
}

每次收到用户消息就更新 token,发送时取最新的那个。token 会随着用户新消息刷新,但旧的也能用。

8. 完整的 Python 客户端 (120 行)

"""
微信 ilink Bot 客户端 — 完整实现
"""
import base64, json, logging, os, random, time, uuid
from pathlib import Path
import httpx

ILINK_BASE = "https://ilinkai.weixin.qq.com"

class WeChatBot:
    def __init__(self, token, to_user_id, context_token="", config_path="wechat.json"):
        self.base = ILINK_BASE
        self.token = token
        self.to_user_id = to_user_id
        self.context_token = context_token
        self.config_path = config_path
        self._cursor = ""

    @classmethod
    def from_config(cls, path="wechat.json"):
        with open(path) as f:
            cfg = json.load(f)
        return cls(
            token=cfg["token"],
            to_user_id=cfg["to_user_id"],
            context_token=cfg.get("context_token", ""),
            config_path=path,
        )

    def _headers(self):
        uin = base64.b64encode(str(random.randint(0, 0xFFFFFFFF)).encode()).decode()
        return {
            "Content-Type": "application/json",
            "AuthorizationType": "ilink_bot_token",
            "Authorization": f"Bearer {self.token}",
            "X-WECHAT-UIN": uin,
        }

    def _post(self, endpoint, body):
        body["base_info"] = {"channel_version": "1.0.3"}
        raw = json.dumps(body, ensure_ascii=False).encode("utf-8")
        headers = self._headers()
        headers["Content-Length"] = str(len(raw))
        resp = httpx.post(
            f"{self.base}/ilink/bot/{endpoint}",
            content=raw, headers=headers, timeout=35,
        )
        text = resp.text.strip()
        return json.loads(text) if text and text != "{}" else {"ret": 0}

    def get_updates(self):
        """长轮询拉取新消息,自动更新 context_token"""
        result = self._post("getupdates", {"get_updates_buf": self._cursor})
        self._cursor = result.get("get_updates_buf", self._cursor)
        for msg in result.get("msgs", []):
            ct = msg.get("context_token", "")
            if ct:
                self.context_token = ct
                self._save_token(ct)
        return result.get("msgs", [])

    def send(self, text, to=None, context_token=None):
        """发送文本消息"""
        return self._post("sendmessage", {
            "msg": {
                "from_user_id": "",
                "to_user_id": to or self.to_user_id,
                "client_id": f"bot-{uuid.uuid4().hex[:12]}",
                "message_type": 2,
                "message_state": 2,
                "context_token": context_token or self.context_token,
                "item_list": [{"type": 1, "text_item": {"text": text}}],
            }
        })

    def refresh_and_send(self, text):
        """先刷新 context_token,再发送 (推荐)"""
        self.get_updates()
        return self.send(text)

    def _save_token(self, ct):
        try:
            p = Path(self.config_path)
            if p.exists():
                cfg = json.loads(p.read_text())
                cfg["context_token"] = ct
                p.write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
        except Exception:
            pass

    def listen(self, handler):
        """持续监听消息 (阻塞)"""
        while True:
            try:
                msgs = self.get_updates()
                for msg in msgs:
                    ct = msg.get("context_token", "")
                    from_user = msg.get("from_user_id", "")
                    text = ""
                    for item in msg.get("item_list", []):
                        if item.get("type") == 1:
                            text = item.get("text_item", {}).get("text", "")
                    if ct and text:
                        reply = handler(text, from_user)
                        if reply:
                            self.send(reply, to=from_user, context_token=ct)
            except Exception as e:
                logging.error(f"listen error: {e}")
                time.sleep(5)

9. 实战: 量化选股结果推送

我的使用场景是: 6-Agent 选股分析跑完后,自动把结果推到微信。

bot = WeChatBot.from_config("wechat.json")

# 跑完分析后一行代码推送
bot.refresh_and_send("""
📊 智能选股报告 2026-03-24
━━━━━━━━━━━━━━

#1 AVGO $310.51 [分歧]
   趋势↓ RSI:45 止损$300.64
   支撑$307.20 阻力$318.04

#2 NVDA $172.70 [分歧]
   趋势↓ RSI:37 止损$169.91
   R:R=1.4:1 ← 风险回报最优

#3 AAPL $247.99 [看空]
   RSI:24 超卖! 可能反弹

by QuantByQlib 6-Agent
""")

也可以搭建交互式 Bot,用户发指令触发分析:

def handler(text, from_user):
    if text.startswith("分析"):
        symbols = text.split()[1:]
        report = run_analysis(symbols)
        return format_report(report)
    elif text == "帮助":
        return "发送 '分析 NVDA AAPL' 开始分析"
    return None

bot.listen(handler)

10. 踩坑清单 (省你两天)

# 坑 表现 解法
1 缺 client_id 200 但不投递 每条消息生成唯一 UUID
2 缺 message_type 200 但不投递 固定传 2 (BOT)
3 缺 message_state 200 但不投递 固定传 2 (FINISH)
4 缺 base_info 200 但不投递 {"channel_version": "1.0.3"}
5 缺 Content-Length 偶发超时 手动计算 UTF-8 字节长度
6 缺 context_token 200 但不投递 getUpdates 获取,持久化保存
7 响应体为 {} 以为失败 {} 就是成功,sendMessage 无返回值
8 get_qrcode_status 超时 以为登录失败 正常行为,重试即可
9 二维码过期 status=“expired” 重新调 get_bot_qrcode

11. 这个方案的边界

能做的:

  • 个人微信收发消息 (1对1)
  • 文本/图片/文件/视频 (需 AES-128-ECB 加密上传 CDN)
  • 持续运行的交互 Bot
  • 定时推送通知

不能做的 / 注意事项:

  • 不能发群消息 (ilink 只支持 direct chat)
  • 需要先完成扫码登录 (一次即可,token 持久化)
  • 需要用户至少给 bot 发过一条消息 (获取初始 context_token)
  • 不清楚 token 有效期上限 (目前测试数天内正常)
  • 这是腾讯内部平台,协议可能随时变更

12. 与其他方案对比

方案 主动推送 免费 个人可用 稳定性 难度
ilink Bot (本文) ✅ ✅ ✅ ⚠️ 协议可能变 ⭐⭐⭐
公众号模板消息 ❌ 需用户触发 ✅ ❌ 需认证 ⭐⭐⭐⭐⭐ ⭐⭐
企业微信 webhook ✅ ✅ ⚠️ 需企微 ⭐⭐⭐⭐⭐ ⭐
Server酱 ✅ ⚠️ 5条/天 ✅ ⭐⭐⭐⭐ ⭐
itchat/wechaty ✅ ✅ ✅ ❌ 已封杀 ⭐⭐

总结

整个逆向过程的关键收获:

  1. npm 包是个宝藏 —— 很多"闭源"服务的官方 SDK 都以源码形式发布在 npm 上,TypeScript 类型定义就是最好的 API 文档。

  2. HTTP 200 ≠ 成功 —— ilink 的 sendMessage 无论消息是否投递都返回 200 + {}。没有错误码、没有提示。这种设计对调试是灾难性的。

  3. "可选字段"可能是必填的 —— 官方文档只列了 to_user_id、context_token、item_list,但 client_id、message_type、message_state 才是消息路由的关键。

  4. 先读源码再写代码 —— 如果一开始就完整对比 OpenClaw 的请求格式,可以省两天。别猜,看源码。

如果这篇文章帮到了你,请点赞/收藏/转发。有问题欢迎评论交流。

关键词: 微信 Bot、ilink 协议、OpenClaw、逆向工程、Python 微信消息推送、量化交易通知

# 微信 # python # 开发语言
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

2026 GEO 布局完整解决方案（可落地执行）

avatar AtomGit开源社区
cover

AI检测太高论文过不了？这4个降AIGC工具2026年必须用！

avatar AtomGit开源社区
cover

2026年AI产品经理最全学习路线：一篇文章涵盖所有，足够详细

avatar AtomGit开源社区