拓冰网站定制 
官方文档平台介绍 - QiWe API企微 API 开发文档一、业务痛点与技术背景自动回复、入群欢迎、消息存档、SOP 触达都依赖回调。QiWe 官方约束配置方式控制台「应用凭证 → 配置」或 API「设置回调地址」POST application/json3 秒内响应可返回空串 / HTTP 200必须公网可达配置成功会推送「验证回调地址是否可用」Token 下所有已登录账号消息都会转到同一回调 URL接口主动发送的消息不触发回调好友/群消息与手机端发送会触发SCRM 若在回调里同步写库 调 LLM必丢事件。二、核心架构设计与数据流转QiWe Webhook │ ▼ SCRM Ingress (TLS, ≤300ms ACK) │ 验签/限流/去重 ▼ Event Bus (Kafka topic: qiwe.inbound) │ ├─ customer-profile-updater ├─ ai-cs-worker ├─ group-sop-engine └─ message-archive-writer多guid共用回调时第一字段分流键就是guid再进租户路由。三、关键代码与配置示例3.1 IngressFastAPIfrom fastapi import FastAPI, Request, Response import hashlib, json, time import redis, aiokafka app FastAPI() r redis.Redis.from_url(os.environ[REDIS_URL]) producer None # 启动时初始化 def dedupe_key(body: dict) - str: # 以实际回调字段为准guid msgId / uniqueId raw f{body.get(guid)}:{body.get(msgId) or body.get(uniqueId) or body.get(id)} return qiwe:dedupe: hashlib.sha256(raw.encode()).hexdigest() app.post(/webhook/qiwe) async def qiwe_webhook(request: Request): body await request.json() # 先准备 ACK业务异步 text json.dumps(body, ensure_asciiFalse) if 验证回调地址是否可用 in text: return Response(content, media_typetext/plain) key dedupe_key(body) if not r.set(key, 1, nxTrue, ex72 * 3600): return Response(content, media_typetext/plain) await producer.send_and_wait(qiwe.inbound, json.dumps(body).encode(utf-8)) return Response(content, media_typetext/plain)3.2 控制台 / API 配置回调# 方式 A控制台手动配置推荐 # 控制台 → 应用凭证 → 配置 # 方式 B程序化method 以文档「设置回调地址」为准 curl -X POST https://new.qiweapi.com/qiwe/api/qw/doApi \ -H X-QIWEI-TOKEN: $QIWEI_TOKEN \ -H Content-Type: application/json \ -d { method: /callback/setUrl, params: { url: https://scrm.example.com/webhook/qiwe } }3.3 租户与 guid 路由async function routeEvent(evt: any) { const tenant await tenantRepo.byGuid(evt.guid); if (!tenant) { metrics.unknownGuid.inc(); return; } await bus.publish(qiwe.tenant.${tenant.id}, { ...evt, tenantId: tenant.id }); }3.4 事件类型分发器HANDLERS { text: handle_text, image: handle_image, friend_add: handle_friend, room_member: handle_room_member, } def dispatch(evt: dict): t normalize_type(evt) # 按文档字段映射 HANDLERS.get(t, handle_unknown)(evt)四、生产环境避坑与安全风控3 秒铁律Ingress 只做去重入队排障看队列积压而不是回调线程。共用回调一个 Token 多账号务必按guid隔离数据防串租户。验证包配置后立刻打日志确认失败先用 Apifox 公网自测。内网穿透仅开发用生产用正式域名 TLS。主动发送无回调投递状态靠 API 响应 自建发送日志。细节见文首官方文档中的 Webhook 专题。五、本篇交付清单3 秒 ACK Ingress去重 Kafka 事件总线guid 租户路由回调配置两种方式 企业微信API开发：Webhook 回调与 SCRM 事件总线架构
官方文档平台介绍 - QiWe API企微 API 开发文档一、业务痛点与技术背景自动回复、入群欢迎、消息存档、SOP 触达都依赖回调。QiWe 官方约束配置方式控制台「应用凭证 → 配置」或 API「设置回调地址」POST application/json3 秒内响应可返回空串 / HTTP 200必须公网可达配置成功会推送「验证回调地址是否可用」Token 下所有已登录账号消息都会转到同一回调 URL接口主动发送的消息不触发回调好友/群消息与手机端发送会触发SCRM 若在回调里同步写库 调 LLM必丢事件。二、核心架构设计与数据流转QiWe Webhook │ ▼ SCRM Ingress (TLS, ≤300ms ACK) │ 验签/限流/去重 ▼ Event Bus (Kafka topic: qiwe.inbound) │ ├─ customer-profile-updater ├─ ai-cs-worker ├─ group-sop-engine └─ message-archive-writer多guid共用回调时第一字段分流键就是guid再进租户路由。三、关键代码与配置示例3.1 IngressFastAPIfrom fastapi import FastAPI, Request, Response import hashlib, json, time import redis, aiokafka app FastAPI() r redis.Redis.from_url(os.environ[REDIS_URL]) producer None # 启动时初始化 def dedupe_key(body: dict) - str: # 以实际回调字段为准guid msgId / uniqueId raw f{body.get(guid)}:{body.get(msgId) or body.get(uniqueId) or body.get(id)} return qiwe:dedupe: hashlib.sha256(raw.encode()).hexdigest() app.post(/webhook/qiwe) async def qiwe_webhook(request: Request): body await request.json() # 先准备 ACK业务异步 text json.dumps(body, ensure_asciiFalse) if 验证回调地址是否可用 in text: return Response(content, media_typetext/plain) key dedupe_key(body) if not r.set(key, 1, nxTrue, ex72 * 3600): return Response(content, media_typetext/plain) await producer.send_and_wait(qiwe.inbound, json.dumps(body).encode(utf-8)) return Response(content, media_typetext/plain)3.2 控制台 / API 配置回调# 方式 A控制台手动配置推荐 # 控制台 → 应用凭证 → 配置 # 方式 B程序化method 以文档「设置回调地址」为准 curl -X POST https://new.qiweapi.com/qiwe/api/qw/doApi \ -H X-QIWEI-TOKEN: $QIWEI_TOKEN \ -H Content-Type: application/json \ -d { method: /callback/setUrl, params: { url: https://scrm.example.com/webhook/qiwe } }3.3 租户与 guid 路由async function routeEvent(evt: any) { const tenant await tenantRepo.byGuid(evt.guid); if (!tenant) { metrics.unknownGuid.inc(); return; } await bus.publish(qiwe.tenant.${tenant.id}, { ...evt, tenantId: tenant.id }); }3.4 事件类型分发器HANDLERS { text: handle_text, image: handle_image, friend_add: handle_friend, room_member: handle_room_member, } def dispatch(evt: dict): t normalize_type(evt) # 按文档字段映射 HANDLERS.get(t, handle_unknown)(evt)四、生产环境避坑与安全风控3 秒铁律Ingress 只做去重入队排障看队列积压而不是回调线程。共用回调一个 Token 多账号务必按guid隔离数据防串租户。验证包配置后立刻打日志确认失败先用 Apifox 公网自测。内网穿透仅开发用生产用正式域名 TLS。主动发送无回调投递状态靠 API 响应 自建发送日志。细节见文首官方文档中的 Webhook 专题。五、本篇交付清单3 秒 ACK Ingress去重 Kafka 事件总线guid 租户路由回调配置两种方式