当前位置：首页 > news >正文

个人微信API与业务系统对接：从需求到落地的完整路径

news 2026/8/12 22:18:13

这篇就把"微信API和业务系统对接"这件事完整捋一遍，从需求分析到方案设计、开发实现、测试上线，希望能给要做类似项目的同学一个参考。

一、需求分析：先把场景定清楚

对接微信API之前，最容易犯的错就是"想要的功能太多"。客户一开始列了一堆需求：自动加好友、群发、自动回复、群管理、朋友圈、聊天归档……恨不得把微信所有能力都用上。

我拉着他把需求按优先级分了三档：

P0（必须做）：CRM内发消息给客户、接收客户回复、聊天记录入库
P1（二期做）：自动加好友、关键词自动回复、群消息同步
P2（看效果）：朋友圈发布、群管理、数据报表

这样拆完，第一期的范围就清楚了——核心是"消息收发 + 记录归档"。范围一定，选型和方案设计就有抓手了。

二、方案设计：选对API服务很关键

需求定了，接下来是选微信API服务。选型时我重点看三点：

消息收发能力：能不能覆盖文本、图片、文件这些常见类型
回调机制：客户回复能不能实时推到我的服务，而不是轮询
稳定性：登录态能保持多久，掉线重连机制怎么样

对比下来，Eyun的文档结构清晰，回调机制和错误码规范做得不错，还提供了完整的对接示例代码，对快速落地比较友好。最终选了Eyun，具体可以看 Eyun开发文档了解能力范围。

整体架构设计成这样：

CRM系统 → 对接服务层 → Eyun微信API → 客户微信 ↓ 消息归档（MySQL + 对象存储） ↓ 回调服务 ← Eyun推送客户消息

核心是一个"对接服务层"，负责把CRM的业务请求翻译成微信API调用，同时接收回调把消息写回CRM。

三、开发实现：消息收发和归档

开发分三块：发消息、收消息、记录归档。先看发消息的实现：

import requests import uuid class WeChatBridge: def __init__(self, base_url, token, wx_id): self.base_url = base_url self.token = token self.wx_id = wx_id def send_text(self, to_wx, content, crm_msg_id=None): """从CRM发消息到客户微信""" payload = { "from_wx": self.wx_id, "to_wx": to_wx, "content": content, "client_msg_id": crm_msg_id or str(uuid.uuid4()) } resp = requests.post( f"{self.base_url}/send/text", json=payload, headers={"Authorization": f"Bearer {self.token}"}, timeout=10 ) return resp.json() def archive_message(self, msg_id, from_wx, to_wx, content, direction): """消息归档到CRM数据库（示意）""" # 实际会写入MySQL，图片/文件存对象存储 print(f"归档: {direction} {from_wx}->{to_wx}: {content}")

注意client_msg_id这个字段，它用于幂等——CRM里同一条消息重复发送时，微信侧能识别去重。这个字段Eyun是明确提供的，对接时省了很多麻烦。

收消息靠回调服务，用Flask简单搭一个：

from flask import Flask, request app = Flask(__name__) @app.route("/callback/message", methods=["POST"]) def on_message(): data = request.json # data里包含 from_wx, to_wx, msg_type, content 等 bridge.archive_message( msg_id=data.get("msg_id"), from_wx=data.get("from_wx"), to_wx=data.get("to_wx"), content=data.get("content"), direction="inbound" ) # 同步到CRM（调用CRM的API或直接写库） sync_to_crm(data) return {"code": 0, "msg": "ok"} if __name__ == "__main__": app.run(port=8080)

这块的难点不在代码，在于保证回调不丢。网络抖动、服务重启都可能导致回调漏收。我的做法是加一层消息队列（用的Redis），回调先入队，消费失败能重试。