WeChat iLink Bot Protocol & Media Processing

Quick Reference

API Base

• iLink:

  https://ilinkai.weixin.qq.com/ilink/bot

• CDN:

  https://novac2c.cdn.weixin.qq.com/c2c/download

• Auth header:

  X-WECHAT-UIN: {bot_token}

• Source reference:

  @tencent-weixin/openclaw-weixin@1.0.2

  (npm)
• Must use direct connection (

  proxy=None

  ) — Chinese domestic server, no proxy

Core Endpoints

get_bot_qrcode

qrcode_img_content

qr_id

get_qrcode_status

qr_id

getupdates

get_updates_buf

sendmessage

getconfig

ilink_user_id

sendtyping

ilink_user_id

Message Type Constants

text_item.text

image_item

voice_item

Dependencies

httpx>=0.27.0
qrcode>=7.4
pycryptodome>=3.20.0
silk-python>=0.2.0       # PyPI name → import pysilk

sendmessage Protocol

Critical fields that cause silent message loss if incorrect:

base_info

{"channel_version": "1.0.2"}

msg.from_user_id

""

msg.client_id

openclaw-weixin:{ts_ms}-{hex8}

ilink_user_id

getconfig

sendtyping

Response

{}

{
  "msg": {
    "from_user_id": "",
    "to_user_id": "xxx@im.wechat",
    "client_id": "openclaw-weixin:1774427815868-44e6cc41",
    "message_type": 2, "message_state": 2,
    "context_token": "<from inbound msg>",
    "item_list": [{"type": 1, "text_item": {"text": "reply"}}]
  },
  "base_info": {"channel_version": "1.0.2"}
}

Receiving Media — CDN Download

Received images/voice do NOT have

cdn_url

GET https://novac2c.cdn.weixin.qq.com/c2c/download?encrypted_query_param={url_encoded_value}

Three critical details

1. Parameter name:

   encrypted_query_param

   (with d), not

   encrypt_query_param

2. Value must be URL-encoded:

   urllib.parse.quote(value, safe="")

3. Direct GET request to CDN — there is no

   /ilink/bot/getmedia

   endpoint

Received message media structure

image_item:
  url         → WeChat media ID (NOT a CDN URL, do not use for download)
  aeskey      → hex-encoded AES key (32 chars)
  media:
    encrypt_query_param → base64 value for CDN download
    aes_key             → base64(hex string) AES key
    hd_size/mid_size/thumb_size → plain integers (not objects)

voice_item:
  (same structure: media.encrypt_query_param + aes_key)

Download implementation

from urllib.parse import quote
import httpx

async def download_received_media(
    encrypt_query_param: str, aes_key_bytes: bytes, http: httpx.AsyncClient
) -> bytes:
    cdn_url = (
        "https://novac2c.cdn.weixin.qq.com/c2c/download"
        "?encrypted_query_param=" + quote(encrypt_query_param, safe="")
    )
    resp = await http.get(cdn_url)
    resp.raise_for_status()
    return decrypt_aes_ecb(resp.content, aes_key_bytes)

AES Key Decoding

Three possible formats — decode with this priority:

def _decode_aes_key(raw: str) -> bytes:
    # Format 1: Direct hex (32 chars → 16 bytes)
    if len(raw) == 32 and all(c in '0123456789abcdef' for c in raw.lower()):
        return bytes.fromhex(raw)
    # Format 2: base64 → raw 16 bytes
    decoded = _b64_decode_flexible(raw)
    if len(decoded) == 16:
        return decoded
    # Format 3: base64 → hex string → 16 bytes
    if len(decoded) == 32:
        return bytes.fromhex(decoded.decode("ascii"))
    raise ValueError(f"Unknown AES key format: len={len(raw)}")

def _b64_decode_flexible(s: str) -> bytes:
    """Handle standard/URL-safe/padded/unpadded base64."""
    import base64
    s_padded = s + "=" * (-len(s) % 4)
    try:
        return base64.b64decode(s_padded)
    except Exception:
        return base64.urlsafe_b64decode(s_padded)

Key resolution priority:

media.aes_key

aeskey

AES-128-ECB Decryption

from Crypto.Cipher import AES

def decrypt_aes_ecb(ciphertext: bytes, key_16: bytes) -> bytes:
    cipher = AES.new(key_16, AES.MODE_ECB)
    plaintext = cipher.decrypt(ciphertext)
    pad_len = plaintext[-1]
    if 0 < pad_len <= 16 and plaintext[-pad_len:] == bytes([pad_len] * pad_len):
        return plaintext[:-pad_len]
    return plaintext  # invalid padding, return raw

SILK Voice Processing

WeChat voice = SILK codec (Skype variant). Most speech-to-text APIs (Whisper etc.) do not support SILK directly.

WeChat SILK vs Standard SILK

#!SILK_V3

0x02

#!SILK_V3

0xFF 0xFF

Conversion pipeline

CDN download → AES decrypt → strip 0x02 prefix → pysilk.decode(24000)
→ raw PCM (16-bit mono) → WAV header → ready for STT

import io, struct, pysilk

def silk_to_wav(silk_bytes: bytes) -> bytes:
    inp = io.BytesIO(silk_bytes[1:] if silk_bytes[:1] == b'\x02' else silk_bytes)
    pcm = io.BytesIO()
    pysilk.decode(inp, pcm, 24000)
    pcm_data = pcm.getvalue()

    wav = io.BytesIO()
    wav.write(b'RIFF')
    wav.write(struct.pack('<I', 36 + len(pcm_data)))
    wav.write(b'WAVEfmt ')
    wav.write(struct.pack('<IHHIIHH', 16, 1, 1, 24000, 48000, 2, 16))
    wav.write(b'data')
    wav.write(struct.pack('<I', len(pcm_data)))
    wav.write(pcm_data)
    return wav.getvalue()

Package name gotcha: PyPI =

silk-python

pysilk

pysilk

Audio → SILK (Sending voice to WeChat)

WeChat voice messages must be SILK format. To send TTS/MP3/WAV as voice:

MP3/WAV → ffmpeg → PCM (s16le, 24kHz, mono) → pysilk.encode → prepend 0x02 → send as voice_item

import io, os, subprocess, tempfile, pysilk

def audio_to_silk(audio_bytes: bytes, input_ext: str = ".mp3") -> bytes:
    with tempfile.NamedTemporaryFile(suffix=input_ext, delete=False) as tmp_in:
        tmp_in.write(audio_bytes)
        tmp_in_path = tmp_in.name

    tmp_pcm_path = tmp_in_path + ".pcm"
    try:
        subprocess.run(
            ["ffmpeg", "-y", "-i", tmp_in_path,
             "-f", "s16le", "-ar", "24000", "-ac", "1", tmp_pcm_path],
            capture_output=True, timeout=30
        )
        with open(tmp_pcm_path, "rb") as f:
            pcm_data = f.read()
        pcm_input = io.BytesIO(pcm_data)
        silk_output = io.BytesIO()
        pysilk.encode(pcm_input, silk_output, 24000, 24000)  # ← 4 args required!
        return b'\x02' + silk_output.getvalue()
    finally:
        for p in (tmp_in_path, tmp_pcm_path):
            try: os.unlink(p)
            except OSError: pass

pysilk.encode() critical pitfall: requires 4 positional arguments:

encode(input, output, sample_rate, bit_rate)

bit_rate

TypeError: encode() takes at least 4 positional arguments (3 given)

Double 0x02 prefix pitfall:

pysilk.encode(..., tencent=True)

\x02

b'\x02' + silk_bytes

\x02\x02#!SILK_V3...

voice_item must include

playtime

playtime

int(pcm_bytes / (sample_rate * 2) * 1000)

voice_item = {
    "media": cdn_media_obj,
    "encode_type": 6,       # SILK
    "sample_rate": 24000,
    "playtime": duration_ms, # REQUIRED for correct playback length
}

Sending Media — CDN Upload

To send images/video/voice/files to WeChat, you must:

1. Generate AES key, encrypt file
2. Call

   getuploadurl

   to get upload params
3. POST encrypted data to CDN
4. Extract

   x-encrypted-param

   from CDN response header
5. Build

   sendmessage

   with

   media.encrypt_query_param

   and

   media.aes_key

Upload media types

image_item

video_item

file_item

voice_item

getuploadurl request

import hashlib, urllib.parse

aes_key = os.urandom(16)
ciphertext = encrypt_aes_ecb(raw_bytes, aes_key)
filekey = hashlib.md5(ciphertext).hexdigest()

payload = {
    "msg": {
        "from_user_id": "",
        "to_user_id": to_user,
        "filekey": filekey,
        "media_type": media_type,   # 1=image, 2=video, 3=file, 4=voice
        "rawsize": len(raw_bytes),
        "rawfilemd5": hashlib.md5(raw_bytes).hexdigest(),
        "filesize": len(ciphertext),
        "aeskey": aes_key.hex(),    # ← hex string, NOT base64
    },
    "base_info": {"channel_version": "1.0.2"},
}
resp = await http.post("/getuploadurl", json=payload)
upload_param = resp.json()["upload_param"]

CDN upload POST

upload_url = upload_param.get("upload_full_url")
if not upload_url:
    eqp = upload_param["encrypted_query_param"]
    upload_url = (
        "https://novac2c.cdn.weixin.qq.com/c2c/upload"
        f"?encrypted_query_param={urllib.parse.quote(eqp, safe='')}"
        f"&filekey={filekey}"
    )

cdn_resp = await cdn_http.post(
    upload_url,
    content=ciphertext,
    headers={"Content-Type": "application/octet-stream"},
)
download_param = cdn_resp.headers.get("x-encrypted-param", "")
# ↑ This is the download encrypted_query_param for the recipient

sendmessage with uploaded media

media_obj = {
    "encrypt_query_param": download_param,
    "aes_key": base64.b64encode(aes_key.hex().encode()).decode(),
    # ↑ base64 of hex string, matching the format iLink uses for received media
}

# For image:
item = {
    "type": 2,
    "image_item": {
        "media": media_obj,
        "mid_size": len(ciphertext),
    }
}

Upload pitfalls

getuploadurl

x-encrypted-param

media.aes_key

base64(hex_string)

filekey

md5(ciphertext).hexdigest()

Image Format Detection

CDN-decrypted images have no extension info. Detect via magic bytes:

def detect_image_format(data: bytes) -> str:
    if data[:3] == b'\xff\xd8\xff': return ".jpg"
    if data[:8] == b'\x89PNG\r\n\x1a\n': return ".png"
    if data[:4] == b'GIF8': return ".gif"
    if data[:4] == b'RIFF' and data[8:12] == b'WEBP': return ".webp"
    if data[:2] == b'BM': return ".bmp"
    return ".jpg"  # fallback

QR Code Handling

• qrcode_img_content

  from

  get_bot_qrcode

  is a URL string, not base64 image data
• Use

  qrcode

  library to render the URL into a QR image

• get_qrcode_status

  expects

  qr_id

  , not the full QR URL
• After scan confirmation, iLink keeps returning "confirmed" — cache processed

  qr_id

  to prevent duplicate session creation

Network Configuration

iLink API and WeChat CDN are domestic Chinese servers — must use direct connection:

ilink_http = httpx.AsyncClient(
    base_url="https://ilinkai.weixin.qq.com/ilink/bot",
    proxy=None,
    headers={"X-WECHAT-UIN": bot_token}
)

cdn_http = httpx.AsyncClient(proxy=None)

If your application also calls overseas APIs (OpenAI, etc.), configure proxy only for those.

Session Management Pitfalls

QR scan creates duplicate sessions

iLink

get_qrcode_status

_confirmed_qrcodes: set[str] = set()

async def handle_qr_status(qr_id, status):
    if status == "confirmed":
        if qr_id in _confirmed_qrcodes:
            return
        _confirmed_qrcodes.add(qr_id)
        await create_session(...)

Phantom sessions after restart (polling storm)

Sessions are persisted to JSON. On restore, sessions that were created but never received a message (no

from_user_id

getupdates

Fix: Only restore sessions that have a

from_user_id

(from_user_id, service_id)

def restore_sessions(saved: list[dict]):
    groups: dict[tuple, list] = {}
    for s in saved:
        if not s.get("from_user_id"):
            continue  # discard sessions that never received a message
        key = (s["from_user_id"], s["service_id"])
        groups.setdefault(key, []).append(s)
    for key, sessions in groups.items():
        latest = max(sessions, key=lambda x: x["created_at"])
        yield latest

Dead session detection

Sessions where the WeChat user disconnected keep polling forever with empty responses. Add a minimum poll interval + consecutive empty counter:

MIN_POLL_INTERVAL = 2.0  # seconds
MAX_CONSECUTIVE_EMPTY = 10  # ~20 seconds of empty polls → remove

consecutive_empty = 0
while session.active:
    msgs = await get_updates(...)
    if not msgs:
        consecutive_empty += 1
        if consecutive_empty >= MAX_CONSECUTIVE_EMPTY:
            await remove_session(session)
            break
        await asyncio.sleep(MIN_POLL_INTERVAL)
        continue
    consecutive_empty = 0
    # ... process msgs

Common Pitfalls

Python logging silent by default

Python

logging

wechat.*

basicConfig

import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s [%(levelname)s] %(message)s")

Without this, you'll see zero logs from iLink/bridge modules, making debugging nearly impossible.

Debugging Checklist

Sending Messages

1. Message silently lost? → Check

   base_info

   ,

   from_user_id=""

   , unique

   client_id

2. sendmessage returns

   {}

   → This is normal success, not an error

Receiving/Downloading Media

3. CDN 400 Bad Request? → Param name must be

   encrypted_query_param

   (with d) + URL-encode the value
4. CDN 404? → Wrong endpoint; use CDN GET, there is no

   /ilink/bot/getmedia

5. Downloaded image is garbage? → AES key format wrong; try all 3 decode methods (hex / b64→raw / b64→hex→bytes)
6. "Missing cdn_url" warning? → Received images don't have

   cdn_url

   ; use

   media.encrypt_query_param

Sending Media (Upload)

7. Upload rejected / CDN error? →

   getuploadurl

   aeskey must be hex (not b64),

   filekey

   = md5 of ciphertext
8. CDN upload 405? → Must be POST, not PUT
9. Image sent but shows broken? →

   sendmessage

   media.aes_key

   must be

   base64(hex_string)

   , not raw hex

Voice / Audio

10. STT rejects audio? → SILK not converted to WAV; check

    silk-python

    installed (not

    pysilk

    )
11. pysilk.encode TypeError "4 positional arguments"? →

    pysilk.encode(input, output, sample_rate, bit_rate)

    — bit_rate is required
12. Sent voice is 1-second bar? → Two causes: (a) missing

    playtime

    in voice_item (b) double

    \x02

    prefix from manual prepend +

    tencent=True

13. SILK conversion failed "ffmpeg not found"? → Add

    ffmpeg

    to Dockerfile:

    apt-get install -y ffmpeg

Session / QR

14. Multiple sessions from one scan? → QR confirmed status returned repeatedly, not deduplicated
15. QR shows numbers not image? →

    qrcode_img_content

    is a URL, render it as QR with

    qrcode

    lib
16. Polling storm after restart? → Zombie sessions without

    from_user_id

    restored; filter them out
17. ConnectTimeout to ilinkai? → System proxy interfering; set

    proxy=None

    explicitly

References

• Go SDK:

  github.com/openilink/openilink-sdk-go

  (see

  cdn.go

  for CDN download impl)
• Protocol docs: https://www.wechatbot.dev/zh/protocol
• npm source:

  @tencent-weixin/openclaw-weixin@1.0.2

  (see

  send.ts

  ,

  api.ts

  )