不用公眾號、不用企業(yè)微信、不用第三方服務,純 Python 對接微信 AI Bot 平臺,實現(xiàn)消息收發(fā)。
本文記錄了從 npm 包源碼逆向協(xié)議、踩坑 context_token、到發(fā)現(xiàn)"幽靈字段"導致消息靜默丟失的完整過程。

0. 故事的起點

我在做一個量化選股系統(tǒng) (QuantByQlib),跑完 6-Agent 分析后想把結(jié)果推送到微信。

需求很簡單:程序跑完 → 自動發(fā)到我微信。

但現(xiàn)實很骨感:

然后我發(fā)現(xiàn)了 OpenClaw —— 一個開源 AI 助手框架,它有個官方微信插件 @tencent-weixin/openclaw-weixin,聲稱掃碼即可登錄,支持消息收發(fā)。

關鍵是,這個插件是騰訊官方發(fā)布的,用的是微信內(nèi)部的 ilink AI Bot 平臺接口。

我的想法是:不裝 OpenClaw,直接把協(xié)議扒出來,用 Python 復刻。

1. 從 npm 包逆向協(xié)議

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

驚喜: 源碼是 TypeScript 原文發(fā)布的,沒混淆、沒打包。41 個文件,結(jié)構清晰:

src/
├── api/
│   ├── api.ts          ← HTTP 請求層 (5個接口)
│   ├── types.ts        ← 完整類型定義
│   └── session-guard.ts
├── auth/
│   ├── login-qr.ts     ← 掃碼登錄流程
│   └── accounts.ts     ← 賬號持久化
├── messaging/
│   ├── inbound.ts      ← 消息接收 + context_token 管理
│   ├── send.ts         ← 消息發(fā)送
│   └── process-message.ts  ← 完整處理鏈路
├── cdn/
│   ├── aes-ecb.ts      ← AES 加密
│   └── cdn-upload.ts   ← 媒體上傳
└── channel.ts          ← 插件主入口

我花了一個晚上通讀了所有源碼,梳理出了完整協(xié)議。

2. 協(xié)議全貌: 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)),         # 必須精確
}

接口列表

另外還有兩個登錄接口 (不在 bot 路徑下):

• GET /ilink/bot/get_bot_qrcode?bot_type=3 → 獲取二維碼
• GET /ilink/bot/get_qrcode_status?qrcode=xxx → 輪詢掃碼狀態(tài)

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 — 后續(xù)所有 API 的認證令牌
• ilink_bot_id — Bot 的賬戶 ID
• ilink_user_id — 掃碼人的微信 ID (格式: xxx@im.wechat)

4. 第一個大坑: 消息發(fā)送成功但收不到

拿到 token 后,我寫了最簡單的發(fā)送:

# ? 錯誤的寫法 — 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. 幽靈字段: 逆向發(fā)現(xiàn)的真相

最終我回到 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 的發(fā)送函數(shù):

// 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 個字段:

以及請求體頂層的:

這些字段不在官方文檔里 (README 只寫了 to_user_id, context_token, item_list),但服務端依賴它們做消息路由。缺少任何一個,消息就被靜默丟棄。

6. 正確的發(fā)送格式

import uuid

def send_message(token, to_user_id, text, context_token):
    """能實際投遞的消息發(fā)送"""
    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 協(xié)議的會話上下文令牌。每次用戶給 Bot 發(fā)消息時,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 能不能發(fā)?

答案: API 不報錯 (返回 200),但消息不投遞。必須有 context_token。

那 context_token 會過期嗎?

這是我踩的第二個坑。一開始我以為 context_token 是一次性的,因為:

• 用 context_token 發(fā)第一條消息 → 收到了
• 同一個 token 發(fā)第二條 → 收不到

但真相是: context_token 可以無限復用,收不到是因為第一條發(fā)送的格式就不對!

當我補全了 client_id、message_type、message_state 之后,同一個 context_token 連發(fā) 10 條都能收到。

OpenClaw 的源碼也證實了這一點 —— 在 inbound.ts 里,context_token 是持久化存儲的:

// src/messaging/inbound.ts
const contextTokenStore = new Map();  // 內(nèi)存緩存

export function setContextToken(accountId, userId, token) {
    contextTokenStore.set(`${accountId}:${userId}`, token);
    persistContextTokens(accountId);  // 同時寫磁盤
}

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

每次收到用戶消息就更新 token,發(fā)送時取最新的那個。token 會隨著用戶新消息刷新,但舊的也能用。

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

"""
微信 ilink Bot 客戶端 — 完整實現(xiàn)
"""
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):
        """發(fā)送文本消息"""
        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,再發(fā)送 (推薦)"""
        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):
        """持續(xù)監(jiān)聽消息 (阻塞)"""
        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. 實戰(zhàn): 量化選股結(jié)果推送

我的使用場景是: 6-Agent 選股分析跑完后,自動把結(jié)果推到微信。

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 ← 風險回報最優(yōu)

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

by QuantByQlib 6-Agent
""")

也可以搭建交互式 Bot,用戶發(fā)指令觸發(fā)分析:

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

bot.listen(handler)

10. 踩坑清單 (省你兩天)

11. 這個方案的邊界

能做的:

• 個人微信收發(fā)消息 (1對1)
• 文本/圖片/文件/視頻 (需 AES-128-ECB 加密上傳 CDN)
• 持續(xù)運行的交互 Bot
• 定時推送通知

不能做的 / 注意事項:

• 不能發(fā)群消息 (ilink 只支持 direct chat)
• 需要先完成掃碼登錄 (一次即可,token 持久化)
• 需要用戶至少給 bot 發(fā)過一條消息 (獲取初始 context_token)
• 不清楚 token 有效期上限 (目前測試數(shù)天內(nèi)正常)
• 這是騰訊內(nèi)部平臺,協(xié)議可能隨時變更

12. 與其他方案對比

總結(jié)

整個逆向過程的關鍵收獲:

1. npm 包是個寶藏 —— 很多"閉源"服務的官方 SDK 都以源碼形式發(fā)布在 npm 上,TypeScript 類型定義就是最好的 API 文檔。

2. HTTP 200 ≠ 成功 —— ilink 的 sendMessage 無論消息是否投遞都返回 200 + {}。沒有錯誤碼、沒有提示。這種設計對調(diào)試是災難性的。

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

4. 先讀源碼再寫代碼 —— 如果一開始就完整對比 OpenClaw 的請求格式,可以省兩天。

如果這篇文章幫到了你,請點贊/收藏/轉(zhuǎn)發(fā)。有問題歡迎評論交流。