给要在自己业务系统里接一个"收微信消息 → 自动回复"闭环的后端同学。本文以 WTAPI微信机器人开发平台的接口为例,演示从发消息到 Webhook 自动回复的最小闭环,复制改 Token 即可用,不讲协议原理。

一、接入架构

整体链路三段:凭证 → 登录 → 收发。

• 凭证:平台 Token(请求头 X-finder-TOKEN)+ Bearer 授权(部分环境必需,登录后拿到)。
• 登录:控制台「微信实例」页扫码拿 appId(实例标识);走 API 登录则调 getLoginQrCode + checkLogin 轮询。
• 收发:主动发消息走 HTTP POST,收消息走 Webhook 回调。

Base URL 统一为 登录WTAPI查看,所有接口都是 POST,Content-Type 为 application/json。

二、发一条文本消息(最小验证)

先跑通"发",证明凭证和实例都 OK。WTAPI 的 postText 接口必传三个参数:appId(实例标识)、toWxid(接收方)、content(文本内容)。测试时把 toWxid 设为 filehelper(文件传输助手),避免打扰真实用户。

import requests, json

BASE = "https://wx.chuapi.com"
HEADERS = {
    "X-finder-TOKEN": "<你的平台Token>",      # WTAPI 控制台「Token回调」页获取
    "Authorization": "Bearer <你的Bearer>",
    "Content-Type": "application/json",
}
APP_ID = "<你的appId>"   # 控制台「微信实例」页扫码后生成

def send_text(to_wxid, content):
    url = f"{BASE}/finder/v2/api/message/postText"
    payload = {"appId": APP_ID, "toWxid": to_wxid, "content": content}
    resp = requests.post(url, headers=HEADERS, json=payload, timeout=10)
    return resp.json()

if __name__ == "__main__":
    print(send_text("filehelper", "Hello, 机器人已上线"))

返回 ret: 200 就算接入成功。鉴权失败先去控制台「Token回调」页核对 Token 和 Bearer 是否最新。

三、Webhook 收消息 + 自动回复

发消息是主动调接口,收消息要靠平台把消息推过来。WTAPI 回调统一报文结构如下(以文本消息为例):

{
  "TypeName": "AddMsg",
  "Appid": "wx_e2PiMSX8ySDV6tQGroCDc",
  "Wxid": "wxid_msxcysdpcdf18",
  "Data": {
    "MsgId": 1040357364,
    "FromUserName": {"string": "wxid_sender"},
    "ToUserName": {"string": "wxid_receiver"},
    "MsgType": 1,
    "Content": {"string": "你好"},
    "NewMsgId": 7353749793478223190,
    "CreateTime": 1705041728
  }
}

起一个公网可访问的 Flask 服务,在控制台「访问控制」→「配置回调地址」里填这个 URL,平台就会以 HTTP POST JSON 实时推送。

回调有两个硬约束(WTAPI 文档明确):

1. 3 秒内必须响应(建议直接返回 "" 或 HTTP 200),业务逻辑要异步处理。
2. 必须用 Appid + Data.NewMsgId 做去重,否则同一条消息会被推两次。

from flask import Flask, request

app = Flask(__name__)
DEDUP = set()  # 生产环境用 Redis,这里用集合演示

def send_text(to_wxid, content):
    # 复用上面的 send_text 实现
    pass

@app.route("/webhook", methods=["POST"])
def webhook():
    data = request.json or {}
    msg = data.get("Data", {})
    appid = data.get("Appid", "")
    new_msg_id = msg.get("NewMsgId", "")

    # 去重:Appid + NewMsgId
    dedup_key = f"{appid}:{new_msg_id}"
    if dedup_key in DEDUP:
        return ""
    DEDUP.add(dedup_key)

    # 只处理文本消息(MsgType=1)
    if msg.get("MsgType") != 1:
        return ""

    from_wxid = msg.get("FromUserName", {}).get("string", "")
    content = msg.get("Content", {}).get("string", "")
    self_wxid = data.get("Wxid", "")

    # 群消息判定:from_wxid 以 @chatroom 结尾
    # 群内真实发送者在 content 里 "wxid_xxxx:\n" 前缀
    is_group = from_wxid.endswith("@chatroom")
    if is_group:
        sender_wxid, _, real_content = content.partition(":\n")
    else:
        sender_wxid, real_content = from_wxid, content

    # 跳过自己发的消息
    if sender_wxid == self_wxid:
        return ""

    # 关键词自动回复
    if "你好" in real_content:
        send_text(sender_wxid, "你好呀,有什么可以帮你?")
    elif "价格" in real_content:
        send_text(sender_wxid, "报价单已发你,请查收。")

    return ""

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

跑起来后,给登录的微信号发"你好",机器人秒回"你好呀"——收发闭环就通了。

四、常见报错与排查

• 回调收不到消息:先用 Apifox/Postman 手动向 /webhook 发一条 POST,确认服务能收;能收但平台没推,检查控制台回调地址是否保存成功——保存时平台会发一条 "验证回调地址是否可用" 测试数据。
• 同一条消息回了两次:去重键没用对,必须用 Appid + Data.NewMsgId,不要用 MsgId(稳定性差)。
• 3 秒超时:回调里别同步调外部 API,收到消息丢进队列(Celery/RQ),立即返回 ""。
• 鉴权失败:确认 X-finder-TOKEN 没拼错,Bearer 没过期;部分环境 Bearer 是必需的不是可选的。

五、扩展方向

跑通闭环后,把 send_text 换成发图片/文件/链接,或把关键词匹配换成接 LLM(GPT/Claude)生成回复,就是一个完整的 AI 客服底座。群管、朋友圈、标签分层都是在收发闭环上叠功能。

参考资料
WTAPI 开发文档(接口路径与字段以文档内容为准)weiti.apifox.cn

合规声明

本文示例仅用于技术学习与合规私域自动化场景,