只调 sendText 是单向推送。AI 客服、关键词应答、工单采集都要收消息 → 处理 → 再发。RPA 连接器在控制台配置 Webhook 后执行节点收到的好友消息、群消息会 POST 到你的服务你再调发送接口回复。和官方「应用回调」不同这里通常不需要企业侧部署消息加解密服务但事件结构以平台文档为准联调时先全量打日志。方案说明数据方向用户发企微消息 ↓ 执行节点接收 ↓ 平台 POST WebhookJSON ↓ 你的服务幂等 → 入队 → 业务逻辑 → sendText 回复回调服务三原则快速 200— 几秒内返回调 LLM、写库放异步 worker。幂等— 用消息 ID 去重避免重复回复。分类型— 文本、图片、群 等分支不同别写死只处理一种。测收消息的常见误区用 sendText 发出去再等同一条回调往往收不到。应让外部用户或同事给该节点发消息再看 Webhook 是否触发。操作步骤节点在线Token 就绪。部署公网 HTTPS 的/webhook/qw本地用内网穿透。控制台保存 Webhook URL确认验证请求能 200。外部号发测试私聊服务端打印完整 Body。写 handler文本 → 调 sendText 回声。外部群测 或关键词私聊与群聊分支分开。生产用 Redis 存已处理消息 ID。示例代码from flask import Flask, request from threading import Thread import requests import os app Flask(__name__) GATEWAY os.environ[QW_GATEWAY] TOKEN os.environ[QW_API_TOKEN] _seen set() def send_text(guid, toid, content): requests.post( f{GATEWAY}/qw/doApi, json{method: /msg/sendText, params: { guid: guid, toid: toid, content: content, }}, headers{X-API-TOKEN: TOKEN, Content-Type: application/json}, timeout15, ) def handle(event): msg_id event.get(msgId) or event.get(id) if msg_id and msg_id in _seen: return if event.get(msgType) ! text: return guid event[guid] sender event.get(fromId) room event.get(roomId) toid room if room else sender text event.get(content, ) send_text(guid, toid, f已收到{text}) if msg_id: _seen.add(msg_id) app.post(/webhook/qw) def webhook(): event request.get_json(forceTrue) Thread(targethandle, args(event,), daemonTrue).start() return {code: 0}, 200注意群消息回复时toid 通常是 roomId发送者 ID 用于识别谁触发。实例掉线后回调不稳定先查节点状态。内存去重重启会丢生产用 Redis。日志注意脱敏控制聊天记录留存周期。前面有 Nginx 时注意超时和 Body 大小限制。文档API 文档 · Webhook · 官网平台