WXBridge

让任何 AI 产品都可以通过腾讯 iLink Bot API 接入微信个人号——而不仅仅是特定平台。

WXBridge 负责全部微信协议细节（认证、长轮询、游标持久化、消息解析、媒体加密传输、发送），你只需实现一个方法：

微信用户
    ↓ 发送消息（文本 / 图片 / 文件 / 视频）
腾讯 iLink Bot API  (长轮询，服务端 35s 超时)
    ↓
WeixinBridge  (核心循环 + 媒体下载/上传)
    ↓ 调用
AIAdapter  ←──── 你只需实现此接口
    ↓ 返回 str 或 Reply（文本 + 媒体任意组合）
sendmessage → 微信用户

安装

# 核心库（仅文本收发）
pip install wxbridge

# 含 Redis 支持（生产推荐）
pip install "wxbridge[redis]"

# 含媒体传输支持（图片/文件/视频，需要 cryptography）
pip install "wxbridge[media]"

# 全部功能（Redis + 媒体）
pip install "wxbridge[full]"

需要 Python 3.10+

快速开始

文本适配器（最简）

from wxbridge import AIAdapter, WeixinMessage

class MyAdapter(AIAdapter):
    async def reply(self, message: WeixinMessage) -> str:
        # message.text          — 用户文本（或语音转写结果）
        # message.from_user_id  — 稳定的 iLink UID（用作用户唯一标识）
        return await my_ai.chat(message.from_user_id, message.text)

启动桥接

import asyncio
from wxbridge import WeixinBridge

async def main():
    bridge = WeixinBridge(adapter=MyAdapter(), redis_url="redis://localhost")

    # 首次使用需要扫码登录
    if not await bridge.auth.load_token():
        qrcode_token, qrcode_img = await bridge.auth.start_login()
        print("请扫描二维码")
        await bridge.auth.poll_login()  # 等待用户扫码确认

    await bridge.start()
    await asyncio.Event().wait()  # 保持运行

asyncio.run(main())

媒体收发

需要 pip install "wxbridge[media]"（依赖 cryptography）。

媒体文件经 AES-128-ECB 加密后通过腾讯 CDN 传输。WXBridge 封装了全部加密/解密细节。

接收媒体（入站）

开启 auto_download_media=True 后，bridge 在调用 adapter.reply() 前自动下载并解密媒体内容，填充到 item.media_bytes：

from wxbridge import AIAdapter, WeixinMessage, Reply
from wxbridge.models import AdapterReply

class MediaAdapter(AIAdapter):
    async def reply(self, message: WeixinMessage) -> AdapterReply:
        for item in message.media_items:   # 图片/文件/视频条目
            if item.media_bytes is None:
                return Reply.text("媒体下载失败")

            if item.type == 2:   # 图片
                # item.media_bytes 是解密后的原始图片字节
                return Reply.image(item.media_bytes)   # 直接回传

            elif item.type == 4:  # 文件
                return Reply.file(item.media_bytes, filename=item.filename or "file")

        return Reply.text(f"你说：{message.text}")

bridge = WeixinBridge(
    adapter=MediaAdapter(),
    redis_url="redis://localhost",
    auto_download_media=True,    # 开启自动下载
)

message.media_items 只返回图片（type=2）、文件（type=4）、视频（type=5）条目； 语音（type=3）已在服务端完成 STT 转写，直接通过 message.text 读取。

发送媒体（出站）

返回 Reply 对象即可，WXBridge 自动处理加密上传：

from wxbridge.models import Reply

# 纯图片
return Reply.image(png_bytes)

# 纯文件
return Reply.file(pdf_bytes, filename="report.pdf")

# 视频
return Reply.video(mp4_bytes, filename="clip.mp4")

# 文本 + 图片（顺序发送两条消息）
return Reply(items=[
    TextReplyItem("这是结果图："),
    MediaReplyItem(data=chart_bytes, media_type="image"),
])

# 仍然可以直接返回 str（向后兼容）
return "你好！"

Reply 类型参考

AdapterReply = str | Reply，二者均可作为 reply() 的返回值。

AIAdapter 接口

from wxbridge import AIAdapter, WeixinMessage
from wxbridge.models import AdapterReply

class AIAdapter(ABC):
    @abstractmethod
    async def reply(self, message: WeixinMessage) -> AdapterReply:
        """
        处理一条微信消息并返回回复。

        message 字段：
          message.text           str | None  — 用户文本或语音 STT 转写
          message.from_user_id   str         — 稳定的 iLink UID（用作用户唯一标识）
          message.session_id     str         — iLink 会话 ID
          message.context_token  str         — 由桥接层回传，适配器无需关心
          message.media_items    list        — 媒体条目（auto_download_media=True 时含 media_bytes）
          message.message_id     str         — 消息唯一 ID
          message.create_time_ms int         — 消息时间戳（毫秒）
        """
        ...

    async def on_new_session(self, from_user_id: str) -> None:
        """会话超时后首条消息触发（默认 no-op，可覆写以重置对话历史）"""

支持的入站消息类型：

WeixinBridge 参数

WeixinBridge(
    adapter,                      # AI 适配器实例
    storage=None,                 # 存储后端，None 时自动使用 RedisStorage
    redis_url="redis://localhost", # Redis URL（storage=None 时有效）
    session_ttl=3600,             # 会话空闲 TTL（秒），超时后触发 on_new_session
    max_concurrent_tasks=10,      # 最大并发消息处理数
    auto_download_media=False,    # True=自动下载入站媒体到 item.media_bytes（需 cryptography）
)

工作原理

长轮询消息接收

WXBridge 通过 /ilink/bot/getupdates 长轮询（服务端最长 35 秒返回）。每次收到消息后：

1. 持久化新游标到 Redis（服务重启后从断点继续，不丢消息）
2. 过滤掉 Bot 自身发出的消息（message_type=2）
3. 对每条用户消息创建独立的 asyncio.Task，并发调用 adapter.reply()

媒体传输流程

下载（入站）：

消息 JSON image_item.media.encrypt_query_param + aes_key
    ↓
GET https://novac2c.cdn.weixin.qq.com/c2c/download
        ?encrypted_query_param={url_encoded_param}
    ↓ 响应为 AES-128-ECB 加密字节
AES-128-ECB 解密（需先将 aes_key 从 base64/hex 双重编码还原为 16 字节真实密钥）
    ↓
item.media_bytes（原始文件字节）

上传（出站）：MediaReplyItem.data → 生成随机 AES 密钥 → 加密 → getuploadurl 申请 URL → PUT CDN → sendmessage 携带 CDN 引用

注意：入站消息中 encrypt_query_param 和 aes_key 嵌套在 image_item.media 子对象内，而非 image_item 顶层。AES key 有两种编码：直接 base64 编码的 16 字节，或 base64 编码的 32 字符 hex 字符串（需再次 hex 解码）。

errcode=-14 处理

iLink 返回 errcode=-14 表示 token 已过期。WXBridge 会：

• 自动清除存储中的 token
• 停止桥接循环
• 需调用 bridge.auth.start_login() 重新扫码登录

登录流程

微信登录通过扫描二维码完成，token 持久化到 Redis（重启后自动恢复，无需重新扫码）。

# 检查是否已登录
if await bridge.auth.load_token():
    print("已登录，直接启动")
else:
    # 申请二维码
    qrcode_token, qrcode_img = await bridge.auth.start_login()
    # qrcode_img 是二维码图片数据，展示给用户扫描

    # 等待扫码确认（阻塞，自动处理过期重试）
    status = await bridge.auth.poll_login()
    # status: "confirmed" | "expired" | "error"

# 查询当前登录状态
status = await bridge.auth.get_login_status()
# "pending" | "confirmed" | "failed" | "none"

# 退出登录
await bridge.auth.clear_token()

存储后端

RedisStorage（生产默认）

bridge = WeixinBridge(adapter=MyAdapter(), redis_url="redis://localhost:6379")

DictStorage（测试/开发）

from wxbridge import DictStorage
bridge = WeixinBridge(adapter=MyAdapter(), storage=DictStorage())

自定义存储后端

实现 Storage Protocol：

class MyStorage:
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str, ttl: int | None = None) -> None: ...
    async def delete(self, *keys: str) -> None: ...
    async def expire(self, key: str, ttl: int) -> None: ...

Redis Key 规范

嵌入 Web 框架

FastAPI

from contextlib import asynccontextmanager
from fastapi import FastAPI
from wxbridge import WeixinBridge

bridge = WeixinBridge(adapter=MyAdapter(), auto_download_media=True)

@asynccontextmanager
async def lifespan(app: FastAPI):
    await bridge.start()
    yield
    await bridge.stop()

app = FastAPI(lifespan=lifespan)

@app.post("/weixin/login/start")
async def login_start():
    token, img = await bridge.auth.start_login()
    return {"qrcode_token": token, "qrcode_img": img}

@app.get("/weixin/login/status")
async def login_status():
    return {"status": await bridge.auth.get_login_status()}

@app.get("/weixin/health")
async def health():
    return {"healthy": await bridge.is_healthy()}

示例

调试 Web 面板

examples/web_server.py 是一个开箱即用的可视化调试工具，适合在接入初期快速验证整条链路——无需编写任何业务代码。

功能

• 扫码登录：页面内直接扫码，token 自动持久化
• 实时消息流：SSE 推送，收发消息即时显示（含图片缩略图、文件信息）
• 主动发送：向任意已联系用户发送文本、图片、文件、视频
• Echo 模式（可切换）：自动将用户消息原样回显，方便测试媒体传输
• Bridge 控制：启动 / 停止 / 查看运行状态

快速启动

# 安装依赖
pip install 'wxbridge[media]' fastapi uvicorn

# 启动（默认使用内存存储，重启后需重新扫码）
python examples/web_server.py

# 使用 Redis 持久化 token（推荐，重启后无需重新扫码）
REDIS_URL=redis://localhost python examples/web_server.py

# 自定义端口
PORT=8080 python examples/web_server.py

启动后访问 http://localhost:8000，点击「获取登录二维码」完成扫码，即可开始收发消息。

API 端点

/api/send/media 的 media_type 可选值：image、file、video。

注意：主动发送需要目标用户先发过一条消息（用于获取 context_token）。

官方 SDK 参考

WXBridge 基于腾讯 iLink Bot 协议实现，协议细节参考官方 TypeScript SDK：

npx -y @tencent-weixin/openclaw-weixin-cli@latest install

官方 SDK 核心文件

已知协议坑点（调试记录）

1. encrypt_query_param 在 media 子对象内：入站消息中该字段嵌套于 image_item.media.encrypt_query_param，不在 image_item 顶层。
2. CDN 下载路径是 /c2c/download：完整 URL = https://novac2c.cdn.weixin.qq.com/c2c/download?encrypted_query_param={url_encoded}，encrypt_query_param 直接 URL 编码使用，不需要 base64 解码。
3. AES key 双重编码：media.aes_key 是 base64(hex_string) 格式（不是 base64(raw_bytes)），解密前需先 base64 decode 得到 hex 字符串，再 hex decode 得到真正的 16 字节密钥。

开发

# 安装（含全部可选依赖）
pip install -e ".[dev]"

# 运行全部测试
pytest

# 运行单个测试
pytest tests/test_bridge.py::test_message_dispatch

# 代码检查
ruff check .
ruff format .

# 类型检查
mypy wxbridge/

许可证

MIT License