之前用 Dify 接过一轮微信消息，体验不错，但团队里做运营的同事更熟 Coze——插件市场现成的多，工作流拖拽也顺手。那就把微信消息接到 Coze 智能体上，让机器人回复直接复用 Coze 里配好的知识库和插件。

链路本身不复杂：微信来消息 → Webhook 推给我的 Python 服务 → 调 Coze 对话接口拿到回答 → 调发消息接口回给用户。真正花时间的只有两处：Coze 的非流式调用要分三步走（不是一次请求拿结果），以及同一个用户连发两条消息会直接报 4016。下面按我实际搭的顺序来。

第一步：收微信消息

微信侧用托管 API 的 Webhook 回调，消息来了平台主动 POST 到我起的 HTTP 服务。先起一个最简单的接收端：

# WTAPI 微信机器人接口平台 | 文档: https://weiti.apifox.cn
# 本文依赖:
#   - 收消息: Webhook 回调 TypeName=AddMsg (3秒内必须响应, 去重键=Appid+Data.NewMsgId)
#   - 发消息: postText (appId/toWxid/content)
# 相关能力: 群管理 / 朋友圈 / 小程序卡片模块见文档
import time
import hashlib
import threading
import requests
from flask import Flask, request

BASE_URL = "https://wx.chuapi.com"
HEADERS = {
    "Content-Type": "application/json",
    "X-finder-TOKEN": "<your_token>",
}

app = Flask(__name__)
seen_msg_ids = set()  # 简单去重,生产环境换 Redis

@app.route("/callback", methods=["POST"])
def callback():
    data = request.json
    # 先应答,再慢慢处理——平台要求3秒内响应
    threading.Thread(target=handle_message, args=(data,)).start()
    return ""

两个细节：

1. 3 秒响应是硬要求。收到回调先返回空串或 200，处理逻辑扔到线程里慢慢跑。如果同步处理，Coze 响应慢一点就会被平台判定超时重推，一条消息回三遍。
2. 去重键是 Appid + Data.NewMsgId，不是 MsgId。网络抖动时平台会重推同一个事件，不去重用户就会收到重复回复。

第二步：解析消息，只处理该处理的

回调报文长这样（示例数据）：

{
  "TypeName": "AddMsg",
  "Appid": "wx_xxxx",
  "Wxid": "机器人本号的wxid",
  "Data": {
    "MsgId": 123456,
    "NewMsgId": 789012345,
    "FromUserName": {"string": "发送者wxid"},
    "MsgType": 1,
    "Content": {"string": "你好"},
    "CreateTime": 1729000000
  }
}

解析时注意三点：

def handle_message(data: dict):
    if data.get("TypeName") != "AddMsg":
        return  # 只处理新消息,Offline/ModContacts 等事件跳过

    msg = data["Data"]
    msg_key = f'{data["Appid"]}_{msg["NewMsgId"]}'
    if msg_key in seen_msg_ids:
        return  # 重复事件,直接丢弃
    seen_msg_ids.add(msg_key)

    from_wxid = msg["FromUserName"]["string"]
    if from_wxid.endswith("@chatroom"):
        return  # 这篇先只接私聊,群消息的处理是另一个话题

    if msg.get("MsgType") != 1:
        return  # 只处理文本消息(1),图片语音等先不管

    content = msg["Content"]["string"]  # 注意 Content 是嵌套对象
    reply = ask_coze(from_wxid, content)
    send_text(data["Appid"], from_wxid, reply)

Content 是个嵌套对象，取值要 msg["Content"]["string"]，直接当字符串用会拿到一个 dict。这个地方我卡了十分钟。

第三步：调 Coze，三步缺一不可

Coze 国内版域名是 api.coze.cn。非流式模式不是"发一个问题等一个回答"，而是三步：

1. POST /v3/chat 发起对话，拿到 conversation_id
2. GET /v3/chat/retrieve 轮询状态，直到 completed（轮询间隔至少 1 秒）
3. GET /v3/chat/message/list 取回消息列表，找 assistant 角色、type 为 answer 的那条

COZE_TOKEN = "<coze_pat_token>"
BOT_ID = "<your_bot_id>"
COZE_BASE = "https://api.coze.cn"

def ask_coze(user_wxid: str, question: str) -> str:
    headers = {"Authorization": f"Bearer {COZE_TOKEN}"}

    # 1. 发起对话 (首轮不传 conversation_id)
    # user_id 做哈希,避免把真实 wxid 透传给三方平台
    user_id = hashlib.md5(user_wxid.encode()).hexdigest()[:16]
    chat = requests.post(
        f"{COZE_BASE}/v3/chat",
        headers=headers,
        json={"bot_id": BOT_ID, "user_id": user_id,
              "additional_messages": [{"role": "user", "content": question,
                                       "content_type": "text"}]},
        timeout=10,
    ).json()
    conv_id = chat["data"]["conversation_id"]
    chat_id = chat["data"]["id"]

    # 2. 轮询状态,间隔至少1秒
    while True:
        time.sleep(1)
        status = requests.get(
            f"{COZE_BASE}/v3/chat/retrieve",
            headers=headers,
            params={"conversation_id": conv_id, "chat_id": chat_id},
            timeout=10,
        ).json()
        if status["data"]["status"] == "completed":
            break
        if status["data"]["status"] == "failed":
            return "我这边出了点状况,稍后再试"

    # 3. 取回答: 找 role=assistant 且 type=answer 的那条
    messages = requests.get(
        f"{COZE_BASE}/v3/chat/message/list",
        headers=headers,
        params={"conversation_id": conv_id, "chat_id": chat_id},
        timeout=10,
    ).json()
    for m in messages["data"]:
        if m["role"] == "assistant" and m["type"] == "answer":
            return m["content"]
    return "没拿到回复,换个问法试试?"

第四步：4016 并发错误，按用户加锁

跑通之后让同事帮忙测试，他手快连发了两条，第二条直接报错：code=4016，同一会话存在进行中的对话。

原因：Coze 同一个会话不支持并发请求。解决也直接——按 wxid 加一把锁，同一个用户的消息串行处理：

user_locks = {}
locks_guard = threading.Lock()

def get_user_lock(wxid: str) -> threading.Lock:
    with locks_guard:
        if wxid not in user_locks:
            user_locks[wxid] = threading.Lock()
        return user_locks[wxid]

def handle_message(data: dict):
    # ...前面的解析逻辑不变...
    lock = get_user_lock(from_wxid)
    with lock:  # 同一用户串行,不同用户之间仍然并行
        reply = ask_coze(from_wxid, content)
        send_text(data["Appid"], from_wxid, reply)

不同用户之间不受影响，同一个用户的消息排队处理——正好也符合真人聊天的节奏。

第五步：把回答发回微信

def send_text(app_id: str, to_wxid: str, content: str):
    resp = requests.post(
        f"{BASE_URL}/finder/v2/api/message/postText",
        headers=HEADERS,
        json={"appId": app_id, "toWxid": to_wxid, "content": content},
        timeout=10,
    ).json()
    if resp.get("ret") != 200:
        print(f"发送失败: {resp}")

跑起来的效果（示例）：

用户: 你们的退款政策是什么
机器人: (Coze 知识库命中的回答, 约2秒后到达)

小结

整条链路拆开就五件事：回调收消息（先应答再处理）、Appid+NewMsgId 去重、Coze 三步调用（发起→轮询→取答案）、按 wxid 加锁防 4016、postText 回发。Coze 侧的知识库和插件改动完全不用动 Python 代码，这是接智能体平台比自己裸调 LLM 最省心的一点。

接口路径、参数与回调字段以 WTAPI 文档和 Coze 开放平台文档为准。本文涉及的回调字段与发送接口均已核实，代码中的 token、wxid 为占位符，示例消息内容为演示数据。

---