上周有个老客户找我，说他的机器人"最近回复突然少了一半"——不是接口报错，是收消息的回调端"看起来正常但其实没处理到"。排查下来的结果很典型：Webhook 回调链路在他的工程里被三个细节同时坑了，丢在 3 秒应答超时、去重键设成了空字符串、群消息正文没剥前缀。三个坑叠加在一起，线上看起来一切正常（接口 200 应答、日志没报错），但一半的消息从链路里直接漏过。

这类线上问题最讨厌——开发时本地测通了，上线后无声无息地亏业务。所以这篇把"消息收发链路"当成一个完整工程来讲：从回调收到消息的那一刻，到你的业务逻辑处理完再发回微信，中间有哪几个节点可能丢、怎么防、怎么查。代码全部可跑，照着抄就能把这条链路的健壮性拉起来。

先看清整条链路

消息进入你的机器人时，经过的完整路径是这样的：

微信端 → 平台 Webhook 推送 → 你的回调服务 → 业务逻辑处理 → 发送接口发回微信
         ↑                     ↑                ↑
         1. 是否推到你？        2. 你是否收对？   3. 你是否处理对？

排查丢消息时，就按这三段断点来：推不出去、推到了没收对、根本没推。每段有明确的判断依据，不是猜。

这套机器人通道走 WTAPI——基于 iPad/Mac 协议的微信机器人接口服务，把消息、好友、群、朋友圈这些能力做成标准 HTTP API。说直白点：你的程序调用它给的网址，就能用这个微信号收消息、发消息、管群，扫码登录、掉线重连这些协议层的活它来干。回调链路的文档是它的"消息订阅收发"模块。

链路一：推不出去

第一个判断：你用的平台到底有没有把消息推给你？

排查步骤：

1. 回调地址有没有在平台控制台/接口侧设置？ 用它的登录接口模块下有个"设置回调地址"接口，或者直接在控制台后台配置。没配地址等于收不到。
2. 你的服务器公网可达吗？ 用手机 4G 网络访问你配的回调地址，看能不能返回 200。本地服务或内网服务不行，平台推不到。
3. 回调地址有没有正确应答？ 平台对回调有硬约束：3 秒内必须返回 HTTP 200（ok 或 success 都行，关键是 3 秒内返回）。超时没回应，平台会重推，重推几次之后如果还是超时，平台侧就放弃了——这也是我上面那个客户踩到的第一个坑：他在回调函数里同步调 AI，AI 要等十几秒才返回，3 秒超时全部丢。

所以回调函数的第一行应该是：先应答、后处理。业务逻辑（调 AI、写数据库、拉群成员）全部丢线程异步执行，主线程只负责返回 ok。这是一条不能妥协的纪律。

链路二：推到了没收对（解析端）

平台确实推过来了，但你解析错了。这一段坑最多，每一个都要对照文档原文核对：

1. 报文结构：回调报文的顶层一定是 dict。如果 request.get_json(silent=True) 返回了 None，说明请求体不是 JSON——通常是你在回调函数里多做了一步中间件转换，或者把回调地址配成了另一个服务的入口。

2. TypeName 是事件类型。正常消息的 TypeName 是 AddMsg，其他事件（进群、朋友圈等）都有各自的 TypeName。排查时可以先只处理 AddMsg，其他事件打日志。

3. 正文在 Data.Content.string 里，不是 Data.Content 本身。这是嵌套结构：Data.Content 是个 dict，里面有个 string 键才是真正的文本内容。很多人直接取 Content 当字符串用，结果拿到的是 {"string": "真实内容"} 整个 dict，喂给 AI 当然答非所问。

4. 群消息正文带前缀。群聊里收到的消息，正文格式是 发言人wxid:\n真实内容——必须先剥掉前缀再处理。剥法很简单：找 ":\n" 分割，取第二段。不剥前缀直接喂业务逻辑，机器人会把 wxid_xxx: 当成内容的一部分。

5. FromUserName 的 @chatroom 结尾是群消息标记。单聊消息的 FromUserName 是好友 wxid，群聊消息的 FromUserName 以 @chatroom 结尾。判断群聊还是单聊就靠这个后缀。

6. 去重键必须是 Appid + Data.NewMsgId。平台会重推未确认的报文（比如你 3 秒内答了 ok 但业务还没跑），去重靠的就是这对值的组合。我上面那个客户把去重键设成了空字符串，等于所有消息都跳过了去重逻辑——但这不是丢消息，是重复处理；反过来，把去重键设成 FromUserName 也不对，同一人连续发两条消息会被去重掉第二条。

把这六条写进一个 parse 函数里，每次回调都走这个函数，别直接在 webhook 函数里散落取字段的逻辑。

链路三：根本没推（平台侧状态）

如果你前两段都排完了，还是没收到消息，最后查平台侧状态：

• 登录态：机器人对应的 appId 是不是在线？平台控制台能看到设备状态。离线就没有消息可推。
• appId 是否有更新：登录接口里有个参数（文档里叫 appId），首次登录空传、重登必须传上次返回的原值。如果登录态丢了又空传 appId 重登，会生成一个新的设备实例，之前的回调配置就对不上了——消息推给了旧的设备实例，新的实例收不到。

工程化回调实现

把上面三个链路的所有要点收进一个可直接跑的实现里，依赖 pip install flask requests：

# WTAPI 接口平台 | 文档: https://weiti.apifox.cn
# 本文依赖:
#   - 收消息: Webhook 回调 TypeName=AddMsg (3秒内必须返回 200, 业务逻辑异步执行)
#   - 解析: Data.Content.string 取正文; FromUserName 以 @chatroom 结尾=群消息; 群消息剥 "wxid:\n" 前缀
#   - 去重: 键=Appid + Data.NewMsgId (平台会重推未确认报文, 去重靠此组合)
#   - 发送: POST /finder/v2/api/message/postText (appId / toWxid / content)
# 下一步: 先用 filehelper 自发送, 把整条回调→回复链路验通再接业务
# 官网: www.chuapi.com

import threading
import requests
from flask import Flask, request

app = Flask(__name__)

TOKEN = "YOUR_TOKEN"
APP_ID = "YOUR_APPID"
SEND_URL = "https://wx.chuapi.com/finder/v2/api/message/postText"

# 去重集合: 演示版用内存 set, 多进程部署换 Redis, 过期时间设 5 分钟
seen = set()
SEEN_TTL_SECONDS = 300

def parse_message(data: dict) -> dict | None:
    """把原始回调报文解析成统一的消息结构; 返回 None 表示非 AddMsg 事件或解析失败。"""
    type_name = data.get("TypeName", "")
    if type_name != "AddMsg":
        print(f"[跳过事件] TypeName={type_name}")
        return None

    msg = data.get("Data", {})
    from_wxid = msg.get("FromUserName", {}).get("string", "")
    raw_content = msg.get("Content", {}).get("string", "")

    is_group = from_wxid.endswith("@chatroom")
    if is_group and ":\n" in raw_content:
        # 群消息正文格式: "发言人wxid:\n真实内容", 剥前缀
        speaker, real_content = raw_content.split(":\n", 1)
    else:
        real_content = raw_content

    if not real_content.strip():
        return None

    return {
        "from_wxid": from_wxid,
        "is_group": is_group,
        "content": real_content.strip(),
        "msg_id": msg.get("NewMsgId"),
    }

def reply(to_wxid: str, content: str) -> None:
    resp = requests.post(
        SEND_URL,
        headers={"Content-Type": "application/json", "X-finder-TOKEN": TOKEN},
        json={"appId": APP_ID, "toWxid": to_wxid, "content": content},
        timeout=15,
    )
    print("回复:", resp.json().get("msg"))

def handle(parsed: dict) -> None:
    """业务逻辑: 这里接 AI、写数据库、群管等; 线程里跑, 不阻塞回调应答"""
    if parsed["is_group"]:
        # 群消息默认只记录不回复, 避免刷屏; 群管/AI 场景再单独处理
        print(f"[群] {parsed['from_wxid']}: {parsed['content'][:50]}")
        return

    print(f"[单聊] {parsed['from_wxid']}: {parsed['content'][:50]}")
    reply(parsed["from_wxid"], f"收到: {parsed['content']}")

@app.route("/webhook", methods=["POST"])
def webhook() -> str:
    # --- 断点 1: 3秒内必须返回 200, 所有业务逻辑放线程 ---
    raw = request.get_json(silent=True)
    if not isinstance(raw, dict):
        print("[断点1] 报文不是 JSON -> 检查回调地址配置")
        return "ok"

    # --- 断点 2: 去重 ---
    key = (raw.get("Appid", ""), raw.get("Data", {}).get("NewMsgId"))
    if key in seen:
        print(f"[去重] 跳过重复报文 msg_id={key[1]}")
        return "ok"
    seen.add(key)
    # 演示版 seen 永不清理, 生产用带 TTL 的集合或 Redis

    # --- 断点 3: 解析 ---
    parsed = parse_message(raw)
    if parsed is None:
        return "ok"

    # --- 断点 4: 业务异步执行 ---
    threading.Thread(target=handle, args=(parsed,), daemon=True).start()
    return "ok"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

这段代码里四个 [断点 X] 标记就是你排查线上丢消息的锚点——报文中途没了，日志会停在某一个断点，直接看停在哪就知道哪段逻辑没跑。

自检清单

1. 3 秒应答：回调函数里没有任何同步阻塞调用（AI、数据库、HTTP 请求），所有业务逻辑都在子线程
2. 解析六条：正文从 Data.Content.string 取；群消息剥 wxid:\n 前缀；FromUserName 判群聊；TypeName 只处理 AddMsg；NewMsgId 存在；Appid 存在
3. 去重：键是 Appid+NewMsgId 组合，不是任一单字段；seen 集合有过期清理
4. 登录态：appId 在线；重登用了上次返回的原值，没有空传生成新实例
5. 公网可达：手机 4G 能访问回调地址且返回 200