IM Bot 平台协议剖析与 Python 原生复刻实践

一、为什么要把协议“读”出来

IM 平台的机器人（Bot）开放能力通常有两种接入路径：官方 SDK 或插件，以及直接调用 HTTP 接口。SDK 上手快，但依赖链长、运行时约束多，且把协议细节封装在黑盒里；直接走 HTTP 则灵活透明，适合把消息能力嵌入自研系统——例如告警推送、审批联动、接入自建对话服务等场景。问题在于，很多平台的 HTTP 协议文档并不完整，这时“从官方公开源码包反推协议”就成了最直接的学习路径。

先划清边界：本文讨论的对象，是平台主动公开、未混淆的功能源码及其暴露出的 HTTP 行为，目的限于技术学习与合规集成。任何协议分析都应遵守目标平台的服务条款与频率限制，不得用于绕过鉴权、批量注册或骚扰用户。

二、第一步：用源码目录建立“协议地图”

以某平台发布在公共包仓库的机器人插件为例，其源码未混淆、未打包，模块划分清晰，是很好的分析样本。拿到这类源码后，推荐先建立“目录 → 职责 → 协议行为”的三级映射：

源码模块	职责	对应协议行为
认证模块	扫码登录、凭据持久化	登录态获取与本地存储
API 层	HTTP 封装 + 类型定义	接口清单、请求头约定、错误码
消息模块	入站接收、发送、处理链路	长轮询、上下文令牌管理
媒体模块	加密与上传	对象存储直传、内容加密

通读时建议边读边画时序图，重点回答三个问题：会话如何建立（登录态从哪来）、消息如何流动（推还是拉）、副通道如何工作（输入状态、媒体上传）。这三个答案基本构成协议主干。

三、协议全貌：五个 POST 接口覆盖核心能力

所有业务接口均为 POST + JSON，基地址下文以 https://bot-api.example.com 占位示意（真实地址以目标平台为准）。核心接口只有五个：

接口路径	用途
/bot/getupdates	长轮询拉取新消息
/bot/sendmessage	发送文本/结构化消息
/bot/getuploadurl	获取媒体上传地址
/bot/getconfig	获取输入状态所需票据
/bot/sendtyping	上报“正在输入”状态

另有两个登录态接口（GET）：获取二维码、轮询扫码状态。

这个设计值得注意：接口数量被压缩到极小，收与发彻底分离，并用“拉模式”（长轮询）取代 WebSocket 或回调，使客户端无需公网可达即可运行。这也是许多 Bot 平台选择长轮询的原因：部署门槛低、可穿透 NAT、实现简单。

四、通用请求头：三个容易踩坑的细节

headers = {
    "Content-Type": "application/json",
    "AuthorizationType": "bot_token",           # 平台自定义的鉴权类型标识
    "Authorization": f"Bearer {bot_token}",     # 扫码登录后获得
    "X-SESSION-UIN": base64.b64encode(
        struct.pack(">I", random.getrandbits(32))
    ).decode(),                                 # 随机会话标识
    "Content-Length": str(len(body_bytes)),     # 必须精确
}

三个细节值得展开：

1. Content-Length 必须是 UTF-8 字节长度。请求体若含中文，len(字符串) 得到的是字符数，而服务端网关校验的是字节数。正确姿势是先 body = json.dumps(...).encode("utf-8")，再取 len(body)。长度不一致时，部分带严格校验的网关会直接拒绝连接或返回 400，且报错信息往往很含糊，这是新手最常踩的坑。
2. 双字段鉴权。除标准的 Authorization: Bearer 外，还需一个自定义的鉴权类型字段。这类设计通常是为了让网关在完整解析令牌之前，就能按类型做路由和前置校验；缺失它，即使 token 正确也可能被拦在最外层。
3. 随机会话标识。每个请求携带一个随机 32 位整数经 Base64 编码后的值。其确切语义因平台而异，一般用于链路追踪与会话粘性（如把同一会话的请求落到同一后端）。实测中缺失或固定不变该值都可能导致请求被拒，稳妥做法是每个请求随机生成。

五、扫码登录：一个极简状态机

登录流程本质是一个四状态的小状态机：

获取二维码 ──→ 等待扫码 ──→ 已扫码待确认 ──→ 已确认（下发 bot_token）
                  │
                  └──→ 已过期（重新取码）

客户端先请求二维码接口，拿到二维码内容与凭据 ID；把二维码内容渲染出来（终端或网页均可）供用户扫码；随后以 2~3 秒间隔轮询状态接口，直到确认或过期。确认后服务端下发 bot_token，它是后续所有请求的凭据，应立即持久化。仅用标准库即可完成：

import json, time, urllib.request, urllib.parse
BASE = "https://bot-api.example.com"   # 占位示意，实际以平台为准
def _get(path):
    with urllib.request.urlopen(BASE + path, timeout=15) as r:
        return json.loads(r.read().decode("utf-8"))
def login():
    qr = _get("/bot/qrcode?bot_type=3")
    print("请扫码：", qr["qrcode_content"])   # 可交给任意二维码渲染方案展示
    while True:
        st = _get("/bot/qrcode/status?" + urllib.parse.urlencode({"qrcode": qr["qrcode_id"]}))
        if st["status"] == "confirmed":
            return st["bot_token"]
        if st["status"] == "expired":
            return login()            # 过期重取
        time.sleep(2)

六、收发主循环：长轮询的正确姿势

长轮询的关键约定是“把等待交给服务端”：请求里带 timeout（如 30 秒），服务端在没有新消息时挂住请求，有消息或超时才返回；客户端的读超时要略大于服务端 timeout，避免自己先断。核心循环如下：

import json, time, random, base64, struct, urllib.request
def make_post(bot_token):
    def post(path, payload, timeout=35):
        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        headers = {
            "Content-Type": "application/json",
            "Content-Length": str(len(body)),      # 字节长度，不是字符数
            "Authorization": f"Bearer {bot_token}",
            "AuthorizationType": "bot_token",
            "X-SESSION-UIN": base64.b64encode(
                struct.pack(">I", random.getrandbits(32))).decode(),
        }
        req = urllib.request.Request(BASE + path, data=body,
                                     headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return json.loads(r.read().decode("utf-8"))
    return post
def poll_loop(post, on_message):
    offset, backoff = 0, 1
    while True:
        try:
            batch = post("/bot/getupdates", {"offset": offset, "timeout": 30})
            backoff = 1
        except Exception:
            time.sleep(min(30, backoff))           # 指数退避，加随机抖动更佳
            backoff = min(30, backoff * 2)
            continue
        for ev in batch.get("events", []):
            on_message(ev)                         # 建议投入队列，由工作线程处理
            offset = max(offset, ev.get("seq", 0))

工程上有四条经验：其一，收到响应后立即发起下一轮轮询，不要额外 sleep，否则白丢实时性；其二，用服务端返回的序列号（seq/offset）去重与续传，断线重连后从上次位置继续，防止重复消费；其三，网络异常要指数退避并加抖动，避免服务端故障时形成重试风暴；其四，把“轮询”与“业务处理”解耦成生产者—消费者模型，处理慢时先落队列，不阻塞拉取。发送则非常直白：

def send_text(post, to, text):
    return post("/bot/sendmessage",
                {"to": to, "msg_type": "text", "body": {"text": text}})

七、输入状态与上下文令牌：两段式防滥用设计

“正在输入”这类高频、易滥用的状态，很多平台不做成单接口，而是拆成两步：先调 getconfig 换取一张短时效票据，再携带票据调用 sendtyping。两段式的好处是网关可以做精细频控与防重放——票据一次性或短时效，抓包重放很快失效。

def send_typing(post, context_token):
    cfg = post("/bot/getconfig", {})
    return post("/bot/sendtyping",
                {"context_token": context_token, "ticket": cfg["typing_ticket"]})

这里出现了协议里最隐蔽的概念：context_token（会话上下文令牌）。它随每条入站消息下发，客户端需要按会话维度缓存、随新消息滚动更新，并在发送与上报状态时附带；其生命周期通常与会话窗口绑定，过期即失效。实践建议是维护一个“会话 ID → (context_token, 时间戳)”的字典，读取时检查时效，过期就等下一条消息自然刷新，不要拿旧令牌硬发。

八、媒体上传：预签名直传 + 应用层加密

发送图片、文件等媒体消息，典型流程是三步：

1. 调 getuploadurl，服务端返回对象存储的预签名上传地址与文件标识；
2. 客户端在上传前按约定算法（示例平台采用 AES-ECB）加密内容后直传存储；
3. 上传完成后，把文件标识填入消息体，走 sendmessage 正常发送。

预签名直传的好处是媒体流量不经过 Bot 接入层，由对象存储直接承接，扩展性好；应用层加密则保证存储与链路中间环节不出现明文。需要指出：ECB 模式对重复明文分组会产生重复密文分组，安全性偏弱，若平台允许协商，工程上更推荐带认证的加密模式，密钥的来源与轮换策略也应在设计期明确。上传后建议校验返回的摘要（如 ETag/MD5），再进入发送环节。

九、从协议到可用客户端：工程化清单

把上面的零件组装成可长期运行的服务，还需补齐：

• 凭据管理：bot_token 落盘持久化（文件权限 0600），启动时加载、失效时自动重登；日志中绝不打印完整 token。
• 幂等与去重：以 seq/消息 ID 建立去重窗口，处理长轮询重试可能带来的重复投递。
• 可观测性：结构化日志记录请求耗时、错误码、轮询延迟，接入日志服务与监控告警，及时发现令牌过期、接口变更等问题。
• 部署形态：作为常驻进程跑在容器平台，优雅退出时保存 offset 并结束轮询；这类服务天然适合“单进程长连接”模型，无需复杂弹性伸缩。
• 版本防御：协议毕竟是分析所得，平台升级可能带来字段变化，客户端应对未知字段宽容（只取所需），解析失败时报警而非崩溃。

十、写在最后：方法比结论更有价值

回顾整个过程：从一份结构清晰的公开源码出发，建立模块映射 → 提炼接口清单 → 还原请求头约定 → 复现登录状态机 → 实现收发主循环 → 补齐媒体与状态副通道，最后用标准库收拢成轻量客户端。这套方法论适用于任何“文档不全但有公开实现”的平台协议分析。同时必须再次强调合规底线：协议复刻应用于自有账号的自动化集成与学习研究，遵守平台条款与频率限制，尊重用户隐私与个人信息保护相关法规——技术的边界感，决定了它能走多远。