前面两篇讲了原理和字段,这篇不讲新概念,只做一件事:给你一份照着做就能跑通的清单。每一步都标死"做完怎么验证"——做过环境搭建的人都知道,坑不在难,在于你不知道自己卡在哪一步:回调填了没反应、二维码扫了没上线、消息发了没回,全是黑盒。

所以本文的核心方法论是:拆成 6 个带验证点的步骤,每一步验证通过再进行下一步。任何一步卡住,你都能立刻定位。全程讲清楚大约 30 分钟,前提是你有一台能跑 Python 的机器和一个能对外提供回调地址的环境。

接口形态为个人微信托管 API(HTTP 主动调用 + Webhook 回调),代码以 WTAPI 的接口为例,字段对照官方文档核实,以文档站实时内容为准。

--

一、先看全局:30 分钟时间表

步骤内容时间验证点
第 0 步账号与凭证准备5 min拿到 appId + Token
第 1 步服务器与回调地址5 min外网能访问到你的回调路径
第 2 步Python 环境与依赖3 minimport flask, requests 不报错
第 3 步验证发送链路5 min文件传输助手收到消息
第 4 步验证回调链路7 min控制台/日志看到真实推送报文
第 5 步接上自动回复5 min发关键词,收到自动回复

注意:第 1 步是唯一可能超时的环节。有公网服务器的话 5 分钟搞定;没有的话需要内网穿透,第一次配置可能多花几分钟。其余步骤都是确定性的。


二、第 0 步:账号与凭证(5 分钟)

去平台控制台注册账号后,你要拿两样东西:

要拿的东西在哪拿说明
appId控制台「微信实例」页,扫码登录后实例列表直接显示这就是你的机器人号的身份标识
Token 凭证控制台「Token回调」页请求头用,见下文鉴权格式

官方快捷路径:控制台「微信实例」页支持直接在线扫码登录——省掉自己调登录接口推二维码的环节。这也是推荐新手入门的方式(用 API 登录留给需要批量管理或自动重登的场景)。

鉴权头格式(三个头):

HEADERS = {
    "X-finder-TOKEN": "YOUR_TOKEN",          # 控制台「Token回调」页查看
    "Authorization": "Bearer YOUR_JWT",      # 部分环境需要
    "Content-Type": "application/json",
}

✅ 验证点:appId 和 Token 都在手上了。扫码登录的微信号,用另一台手机搜一下确认该号在线。


三、第 1 步:服务器与回调地址(5 分钟)

为什么要有回调地址:微信里的消息是平台主动 POST 给你的,你得有一个外网能访问的地址接住它。

两种情况:

情况做法
有公网服务器直接用它,回调地址形如 http://你的域名或IP:8080/callback
只有本地电脑用内网穿透工具(natapp、frp、ngrok 等),把本机 8080 映射出去,得到一个公网地址

✅ 验证点:先用最简单的方式确认这个地址"外网真的能访问到"。启动一个只返回 200 的空服务:

# probe.py —— 只用来验证地址可达,不处理任何业务
from flask import Flask
app = Flask(__name__)

@app.route("/callback", methods=["POST", "GET"])
def probe():
    return "", 200

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

启动后用手机浏览器打开 http://你的地址/callback,看到空白页(不是超时/拒绝连接)就说明通了。先验证地址,再验证逻辑——顺序反了,出问题时你分不清是地址不通还是代码写错。


四、第 2 步:环境与依赖(3 分钟)

python --version          # 需要 3.8+
pip install flask requests

flask 负责提供回调 HTTP 服务,requests 负责调用平台接口。就这两个,没有别的依赖。

✅ 验证点:

python -c "import flask, requests; print('ok')"

打印 ok 即通过。


五、第 3 步:验证发送链路(5 分钟)

先测"发",再测"收"——发送是你主动调用,成功失败一目了然;接收依赖外部推送,变量多。从问题少的一端开始。

发消息接口 POST /finder/v2/api/message/postText,参数只有三个:appId、toWxid、content。

自测神器:toWxid 填 filehelper,消息会发到你自己微信的"文件传输助手",调试全程不打扰任何真实用户。

# send_test.py —— 第一个验证脚本:只做一次发送
import json
import requests

API_BASE = "https://wx.chuapi.com"
APP_ID   = "YOUR_APPID"
HEADERS  = {
    "X-finder-TOKEN": "YOUR_TOKEN",
    "Authorization": "Bearer YOUR_JWT",   # 部分环境需要
    "Content-Type": "application/json",
}

def post_text(to_wxid, content):
    """官方发消息接口 postText,参数仅 appId/toWxid/content"""
    return requests.post(f"{API_BASE}/finder/v2/api/message/postText",
        headers=HEADERS,
        data=json.dumps({"appId": APP_ID, "toWxid": to_wxid, "content": content}),
        timeout=10).json()

if __name__ == "__main__":
    print(post_text("filehelper", "第一条测试消息 ✅"))

✅ 验证点:运行脚本,微信的"文件传输助手"收到「第一条测试消息 ✅」。

没收到? 先把返回的 JSON 打印出来看 ret 和 msg:常见原因是 appId 对应的实例不在线(回控制台确认实例状态),或 Token/JWT 填错(返回鉴权失败)。这一步不通,不要往下走——后面所有功能都建立在这条链路上。


六、第 4 步:验证回调链路(7 分钟)

现在反向验证。在控制台「Token回调」页把回调地址填上(就是第 1 步验证过的那个 /callback)。

这一步的关键技巧:先只打印、不处理。 用下面的"裸回调打印器"把平台推给你的原始报文完整打出来——你要亲眼看一遍真实数据长什么样(这是上一篇讲的字段表最好的对照物):

# callback_probe.py —— 只打印,不处理,专门用来验证回调
from flask import Flask, request

app = Flask(__name__)

@app.route("/callback", methods=["POST"])
def callback():
    msg = request.get_json(silent=True) or {}
    print("==== 收到回调 ====")
    print("TypeName:", msg.get("TypeName"))
    print("Appid   :", msg.get("Appid"))
    print("Data    :", msg.get("Data"))
    return "", 200            # 必须 3 秒内返回

两条必须记住的规则(回调相关的坑基本都出在这里):

  1. 3 秒内必须返回响应(空串或 200 都行)。业务逻辑一律异步处理——不要在处理函数里做耗时的数据库查询或 HTTP 调用,否则平台判定回调失败。
  2. 去重键是 Appid + Data.NewMsgId。网络重试会重复推送同一条消息,后面写业务时必须做幂等,否则一条消息回复两三次。

✅ 验证点:用另一个微信号给机器人号随便发一条消息,你的服务终端打印出完整报文,TypeName 为 AddMsg,Data.MsgType 为 1(文本)。

没打印? 按这个顺序查:
① 回调地址填的路径和代码里的路由是否一致;
② 用手机浏览器再访问一次 /callback(GET 也会返回 200 说明服务活着);
③ 确认实例是在线状态;
④ 回调地址必须是外网可达——本地 127.0.0.1 平台访问不到。


七、第 5 步:接上自动回复(5 分钟)

把前面两段合起来,加上异步队列和关键词规则,就是完整的机器人:

# bot.py —— 完整闭环:收消息 → 异步处理 → 自动回复
import json
import queue
import threading
import requests
from flask import Flask, request

API_BASE = "https://wx.chuapi.com"
APP_ID   = "YOUR_APPID"
HEADERS  = {
    "X-finder-TOKEN": "YOUR_TOKEN",
    "Authorization": "Bearer YOUR_JWT",   # 部分环境需要
    "Content-Type": "application/json",
}

RULES = {                                  # 按自己的业务改
    "价格": "价格表已发您,回复【人工】可详细咨询~",
    "帮助": "回复【价格】看价格、【资料】领资料、【人工】转客服",
}

def post_text(to_wxid, content):
    requests.post(f"{API_BASE}/finder/v2/api/message/postText",
        headers=HEADERS,
        data=json.dumps({"appId": APP_ID, "toWxid": to_wxid, "content": content}),
        timeout=10)

app = Flask(__name__)
_seen = set()                              # 正式环境换 Redis 并设过期时间
task_queue = queue.Queue()

def consumer():
    """异步消费者:3 秒响应靠它把耗时逻辑挪出回调函数"""
    while True:
        msg = task_queue.get()
        d = msg.get("Data", {})
        if d.get("MsgType") != 1:          # 只处理文本
            continue
        from_user = d.get("FromUserName", {}).get("string", "")
        to_user   = d.get("ToUserName", {}).get("string", "")
        if from_user.endswith("@chatroom"):  # 群消息不自动回复
            continue
        if from_user == to_user:           # 自己发的消息不回(防自循环)
            continue

        text = d.get("Content", {}).get("string", "").strip()
        for kw, reply in RULES.items():
            if kw in text:
                post_text(from_user, reply)
                break

@app.route("/callback", methods=["POST"])
def callback():
    msg = request.get_json(silent=True) or {}
    if msg.get("TypeName") == "AddMsg":
        d = msg.get("Data", {})
        key = f"{msg.get('Appid')}:{d.get('NewMsgId')}"   # 官方去重键
        if key not in _seen:
            _seen.add(key)
            task_queue.put(msg)            # 入队,立刻返回
    return "", 200                         # 3 秒内必须返回

if __name__ == "__main__":
    threading.Thread(target=consumer, daemon=True).start()
    app.run(host="0.0.0.0", port=8080)

✅ 验证点(三条都过才算完成):

  1. 启动时用 post_text("filehelper", "启动成功") 自测,文件传输助手收到 → 发送链路 OK;
  2. 用另一个微信发「价格」→ 收到自动回复 → 接收 + 回复闭环 OK;
  3. 同一条内容连发两次 → 只回一条 → 去重生效。

八、卡住时的自查表

照着顺序走,卡住基本都在这几种情况里:

现象大概率原因动作
发送返回失败实例不在线 / Token 或 JWT 填错回控制台看实例状态、核对凭证
文件传输助手没收到但返回成功看错了号(发到了别的实例)核对 appId 与登录号是否一套
回调地址填了但完全没推送地址外网不可达 / 路径不一致手机浏览器访问验证;核对路由名
回调收到但回复两三次没做去重用 Appid + NewMsgId 幂等
回调时有时无、消息丢失回调函数处理超 3 秒业务移到异步队列,回调只入队
消息一直循环回复没过滤自己发的消息判断 FromUserName == ToUserName

合规声明:自动回复应用于正当客户服务与信息通知场景,请控制发送频次、避免骚扰用户,遵守微信平台规则及相关法律法规。接口能力、回调字段与平台限制以官方文档实时内容为准。


Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐