Dissecting the Tencent WeChat OpenClaw Plugin API and Recreating It in Pure Python

Reverse‑engineering the npm package

Inspect the package metadata with

curl -s https://unpkg.com/@tencent-weixin/[email protected]/?meta | python -m json.tool

. The TypeScript source (41 files) is unminified and clearly structured:

src/
├── api/
│   ├── api.ts        ← HTTP layer (5 interfaces)
│   ├── types.ts      ← full type definitions
│   └── session-guard.ts
├── auth/
│   ├── login-qr.ts   ← QR‑login flow
│   └── accounts.ts   ← account persistence
├── messaging/
│   ├── inbound.ts    ← message receiving + context_token handling
│   ├── send.ts       ← message sending
│   └── process-message.ts ← full processing chain
├── cdn/
│   ├── aes-ecb.ts    ← AES encryption
│   └── cdn-upload.ts ← media upload
└── channel.ts       ← plugin entry point

Full protocol overview

All requests are POST JSON to the base URL https://ilinkai.weixin.qq.com. The five main bot endpoints are: /ilink/bot/getupdates – long‑poll for messages /ilink/bot/sendmessage – send a message /ilink/bot/getuploadurl – obtain CDN upload URL /ilink/bot/getconfig – retrieve typing ticket /ilink/bot/sendtyping – set "typing" status

Two additional login endpoints (outside the /bot path) are used to obtain the QR code and poll its status:

GET /ilink/bot/get_bot_qrcode?bot_type=3

GET /ilink/bot/get_qrcode_status?qrcode=xxx

QR‑code login in Python (≈60 lines)

import httpx, qrcode, time
BASE = "https://ilinkai.weixin.qq.com"
# Step 1: get QR code
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: display QR code in terminal
qr = qrcode.QRCode(border=1)
qr.add_data(qrcode_url)
qr.make(fit=True)
qr.print_ascii(invert=True)
# Step 3: poll for scan confirmation
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

After login three key values are obtained: bot_token – API authentication token ilink_bot_id – bot account ID ilink_user_id – user’s WeChat ID (e.g. [email protected])

First major pitfall – messages not delivered

An initial request that omitted several hidden fields returned HTTP 200 with an empty JSON body ( {}) but the message never appeared in WeChat. The missing fields are: from_user_id – empty string (must be present, not omitted) client_id – UUID, unique per‑message identifier for deduplication and routing message_type – 2 (BOT message; 1 = user) message_state – 2 (FINISH; 0 =new, 1 =generating) base_info.channel_version – "1.0.3" (plugin version identifier)

These fields are not listed in the official README; the server silently discards messages lacking any of them.

Correct message format

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).encode("utf-8")
    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)),
    }
    resp = httpx.post(
        "https://ilinkai.weixin.qq.com/ilink/bot/sendmessage",
        content=raw,
        headers=headers,
        timeout=15,
    )
    return resp.status_code == 200

Second major pitfall – context_token handling

The context_token is the session token returned in every inbound message. Sending a message without it yields HTTP 200 but no delivery. Early experiments suggested the token was single‑use because the first message succeeded and the second failed. The real cause was the missing required fields, not token reuse. The token can be reused indefinitely as long as the request format is correct.

Source code in inbound.ts stores the token in an in‑memory map and persists it to disk, proving that the token is intended for reuse:

const contextTokenStore = new Map(); // memory cache
export function setContextToken(accountId, userId, token) {
    contextTokenStore.set(`${accountId}:${userId}`, token);
    persistContextTokens(accountId); // also write to disk
}
export function getContextToken(accountId, userId) {
    return contextTokenStore.get(`${accountId}:${userId}`);
}

Complete Python client (≈120 lines)

""" WeChat ilink Bot client – full implementation """
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):
        """Long‑poll for new messages and update 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):
        """Send a text message"""
        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):
        """Refresh context_token then send (recommended)"""
        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):
        """Continuously poll and invoke handler for each incoming message"""
        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)

Pitfall checklist

Missing client_id → 200 but no delivery

Missing message_type (must be 2 for BOT)

Missing message_state (must be 2 for FINISH)

Missing base_info.channel_version → 200 but no delivery

Incorrect Content‑Length (must be exact UTF‑8 byte length)

Omitting context_token → 200 but message dropped

Empty response body {} actually means success for sendMessage QR code expiration returns status expired – simply retry

What the solution can and cannot do

Send and receive one‑to‑one text messages from a personal WeChat account

Support text, image, file, video (requires AES‑128‑ECB encrypted CDN upload)

Run a persistent interactive bot or scheduled notifications

Cannot send group messages (ilink only supports direct chat)

Requires an initial QR‑code login and at least one inbound message to obtain the first context_token Token validity period is undocumented; tests show it works for several days

The protocol is internal to Tencent and may change without notice

Conclusion

Key takeaways:

The npm package provides the full TypeScript SDK, making reverse‑engineering of closed‑source services straightforward.

HTTP 200 does not guarantee successful delivery; sendMessage returns {} even when the message is silently dropped.

Fields omitted from the public README ( from_user_id, client_id, message_type, message_state, base_info.channel_version) are required for routing.

Comparing the generated request with the original OpenClaw source avoids days of trial‑and‑error.

Original article (in Chinese): https://medium.com/@gymayong/%E6%88%91%E9%80%86%E5%90%91%E4%BA%86%E8%85%BE%E8%AE%AF%E5%BE%AE%E4%BF%A1-ilink-%E5%8D%8F%E8%AE%AE-%E7%94%A8-python-%E5%AE%9E%E7%8E%B0%E4%BA%86%E4%B8%80%E4%B8%AA%E8%83%BD%E4%B8%BB%E5%8A%A8%E6%8E%A8%E9%80%81%E7%9A%84%E5%BE%AE%E4%BF%A1-bot-48d429106b72