1. 从「跑完任务想推微信」说起:ilink 协议到底能做什么
如果你写过量化脚本、爬虫、定时任务,大概率都遇到过同一个尴尬:程序在服务器上跑得好好的,结果只能自己看日志,想让结果主动弹到微信里,却发现路全被堵死了。公众号要企业认证、个人订阅号不能主动推、企业微信 webhook 要开 Server 酱、第三方框架(itchat、wechaty 之类)这两年基本全凉。我当初做量化选股系统,6 个 Agent 分析跑完,就想把结果推给自己,结果折腾了一圈发现——微信官方其实留了一条口子,藏在@tencent-weixin/openclaw-weixin这个 npm 包里,走的是微信内部的ilink AI Bot 平台接口。
ilink 协议是什么?简单说,它是微信给 AI Bot 场景准备的一套 HTTP 接口,基地址是https://ilinkai.weixin.qq.com,全部 POST JSON,一共 5 个核心接口:getUpdates(长轮询收消息)、sendMessage(发消息)、getUploadUrl(CDN 上传)、getConfig(拿 typing ticket)、sendTyping(正在输入状态)。加上两个扫码登录接口,就能拼出一个完整的收发闭环。
它适合谁?适合想给自己做「私人推送通道」的开发者——量化结果、爬虫告警、CI 构建通知、Agent 任务完成提醒,都能用。不适合谁?不适合想做群发、做营销、做多人群聊 Bot 的人,ilink 只支持 1 对 1 direct chat,群消息发不了。
这篇我会把协议握手、消息上行、主动下发链路完整拆开,给你一份能直接复制的 Python 骨架,最后跑通一条主动推送消息。工具侧统一走 TaoToken 的 Key/API 通道做配置管理,避免 token 散落在各个脚本里。
2. 前置准备:TaoToken 统一 Key 与 ilink 凭证的关系
在动手写代码之前,先把两件事分清楚,不然很容易混。
第一件是 ilink 的 bot_token。这是你扫码登录后微信服务端返回的凭证,格式是一长串 base64,后续所有/ilink/bot/*接口都要带Authorization: Bearer {bot_token}。它跟你的微信号绑定,属于「通道凭证」。
第二件是工具侧的 API Key。如果你像我一样,Bot 背后要调大模型做回复、做分析,那模型调用这一层建议统一走 TaoToken 的 API 通道,把 Key 集中管理,而不是每个脚本里硬编码一份。TaoToken 官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 入口是 https://taotoken.net/api ,两套凭证各管各的,互不干扰。
我踩过的坑是:一开始把 ilink 的 bot_token 和模型 Key 混在一个 config 里,结果换模型的时候不小心把 bot_token 覆盖了,扫码重新登了一次。后来我改成两个文件:wechat.json只放 ilink 凭证,llm.json放模型侧配置,清爽很多。
注意:ilink 的 bot_token 是扫码登录后一次性拿到的,建议持久化到本地文件,别每次启动都重新扫。扫码接口本身有频率限制,扫太勤会被临时拦。
3. 协议握手:扫码登录拿到三个关键值
ilink 的登录流程是「拿二维码 → 轮询扫码状态 → 拿到 token」,两个接口都不在 bot 路径下:
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 | Bot 的认证令牌 | 后续所有 API 的 Bearer |
ilink_bot_id | Bot 账户 ID | 多账号区分 |
ilink_user_id | 扫码人的微信 ID | 格式xxx@im.wechat,主动推送的默认目标 |
get_qrcode_status超时是正常行为,不是登录失败,重试即可。二维码过期(status=expired)就重新调get_bot_qrcode。
4. 可复制配置:Python 客户端骨架与请求头
拿到 token 之后,先别急着发消息。ilink 的请求头有几个固定字段,少一个都可能出问题:
import base64, json, random, uuid, httpx def build_headers(token): uin = base64.b64encode( str(random.randint(0, 0xFFFFFFFF)).encode() ).decode() return { "Content-Type": "application/json", "AuthorizationType": "ilink_bot_token", # 固定值 "Authorization": f"Bearer {token}", "X-WECHAT-UIN": uin, # 随机 uint32 的 base64 }AuthorizationType必须是ilink_bot_token,不是Bearer也不是别的。X-WECHAT-UIN每次请求随机生成一个 uint32 再 base64,服务端用它做请求去重和路由。
然后是完整的客户端骨架,我把收发都封进去了:
import base64, json, logging, 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): 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): 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: passwechat.json长这样:
{ "token": "你的 bot_token", "to_user_id": "xxx@im.wechat", "context_token": "" }5. 消息上行与主动下发:context_token 是命门
ilink 的消息链路分两条:上行是用户给 Bot 发消息,Bot 通过getUpdates长轮询拉取;下发是 Bot 主动调sendMessage推消息。两条链路都绕不开context_token。
context_token是 ilink 的会话上下文令牌。每次用户给 Bot 发消息,getUpdates返回的消息体里都会带一个:
{ "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),但消息不投递。这是 ilink 最阴险的设计——静默失败。
那 context_token 会过期吗?我一开始以为是一次性的,因为用同一个 token 发第一条收到了,发第二条就收不到。后来才发现真相:context_token 可以无限复用,收不到是因为第一条发送的格式就不对。补全client_id、message_type、message_state之后,同一个 token 连发 10 条都能收到。
所以主动推送的正确姿势是:先get_updates刷新一次 context_token,再send。这就是上面refresh_and_send方法存在的原因。
6. 验证请求:跑通第一条主动推送
现在把代码跑起来。先确保你至少给 Bot 发过一条消息(这样才有初始 context_token),然后:
bot = WeChatBot.from_config("wechat.json") # 主动推送一条消息 bot.refresh_and_send(""" 智能选股报告 2026-03-24 ━━━━━━━━━━━━━━ #1 AVGO $310.51 [分歧] 趋势↓ RSI:45 #2 NVDA $172.70 [分歧] 趋势↓ RSI:37 #3 AAPL $247.99 [看空] RSI:24 超卖! by QuantByQlib 6-Agent """)成功的话,微信上会立刻收到这条消息。注意sendMessage的响应体是{},空对象就是成功,别以为没返回就是失败。
如果你想做交互式 Bot,加一个listen循环:
def handler(text, from_user): if text.startswith("分析"): symbols = text.split()[1:] return f"收到,正在分析 {symbols}..." elif text == "帮助": return "发送 '分析 NVDA AAPL' 开始分析" return None 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)7. 本篇常见错排查:HTTP 200 不等于成功
ilink 最坑的地方就是静默失败——返回 200 +{},但消息根本没投递。下面这张表是我踩了两天坑总结出来的:
| 报表现象 | 解法 |
|---|---|
| 200 但不投递 | 缺client_id,每条消息生成唯一 UUID |
| 200 但不投递 | 缺message_type,固定传 2(BOT) |
| 200 但不投递 | 缺message_state,固定传 2(FINISH) |
| 200 但不投递 | 缺base_info,传{"channel_version": "1.0.3"} |
| 偶发超时 | 缺Content-Length,手动算 UTF-8 字节长度 |
| 200 但不投递 | 缺context_token,先getUpdates拿 |
响应体{} | 这是成功,不是失败,sendMessage无返回值 |
get_qrcode_status超时 | 正常行为,重试即可 |
二维码expired | 重新调get_bot_qrcode |
还有一个隐藏坑:from_user_id必须传空字符串"",不是不传,也不是传你的 ID。服务端靠这个字段判断消息方向。
8. 边界与后续:这个方案能做什么、不能做什么
能做的:个人微信 1 对 1 收发文本、图片、文件、视频(媒体要 AES-128-ECB 加密后传 CDN)、持续运行的交互 Bot、定时推送通知。
不能做的:群消息(ilink 只支持 direct chat)、未经扫码登录直接调用、用户没给 Bot 发过消息就主动推(拿不到初始 context_token)。
注意事项:token 有效期目前测试数天内正常,但这是腾讯内部平台,协议可能随时变更。建议把channel_version做成配置项,方便后续跟版本。
如果你 Bot 背后要接大模型做智能回复,模型调用这一层建议走 TaoToken 的 API 通道统一管理,Key 集中配置,别散落在各个脚本里。接入文档在 https://taotoken.net/api ,模型对话调试入口在 https://taotoken.net/api-keys ,长期跑编码类 Agent 任务的话可以看 Coding Plan 页面。
整个逆向过程最大的收获其实不是代码,而是三个认知:npm 包是个宝库,很多「闭源」服务的官方 SDK 都以源码形式发布在 npm 上,TypeScript 类型定义就是最好的 API 文档;HTTP 200 不等于成功,ilink 的sendMessage无论消息是否投递都返回 200 +{};「可选字段」可能是必填的,官方文档只列了to_user_id、context_token、item_list,但client_id、message_type、message_state才是消息路由的关键。先读源码再写代码,别猜。