做微信自动化开发的朋友应该都有体会最头疼的不是业务逻辑而是微信协议维护。itchat 停更了、wechaty 要跟进版本、自己逆向的话改到怀疑人生。踩了不少坑之后发现用第三方平台封装的 API 接口是最省心的方案。Eyun 就是这样的平台——把微信能力封装成标准 HTTP 接口不用管协议调接口就行。这篇把 Eyun 的核心接口、调用方式、常见坑点整理出来看完直接能上手。官网微信二次开发 API 文档与在线调试 | Eyun e云管家方案架构Eyun 的架构分四层开发者只需要跟最外层的 API 网关和最内层的事件回调打交道中间的协议维护、实例管理由平台屏蔽API 网关层统一鉴权、路由、限流所有接口入口能力调度层消息编排、批量发送、定时任务、多账号分流协议执行层微信协议维护、实例管理保障实例在线事件回调层Webhook 推送将微信事件实时推送到业务系统核心能力概览Eyun 覆盖了微信客户端的大部分常见操作按类型划分消息类文本、图片、文件、语音、视频、链接、名片、动图、小程序、群、撤回支持 Webhook 实时接收和消息转发。联系人/群聊类通讯录同步、好友搜索、群聊管理、标签管理。实例管理类多账号挂载、在线状态监控、断线重连、动态代理。内容类朋友圈发布、视频号操作、收藏夹管理。通用约定基础配置项目说明Base URLhttp://你的域名地址请求格式Content-Type: application/json认证方式Authorization: Bearer {token}请求方法除特殊说明外均为 POST成功码code: 1000失败码code: 1001核心概念名称含义wId登录实例标识每次登录可能变化业务接口主要使用它wcId微信 ID / 接收方 ID群聊 ID 通常以chatroom结尾messageTypeWebhook 消息类型接收回调时按此字段分发处理注意wId和wcId千万别搞混。wId是你自己的实例IDwcId是对方的微信号。搞反了消息全发给自己别问我怎么知道的。快速接入第一步创建实例和获取凭证前往 Eyun 控制台 注册账号在后台创建执行实例扫码登录微信账号获取 API Key就是Authorization里的那个 token第二步发送消息发送消息的核心是调POST /sendText接口。以下是一个带错误分类重试的封装直接复制就能用import requests import time import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) class EyunClient: def __init__(self, base_url, api_key): self.base_url base_url.rstrip(/) self.headers { Authorization: fBearer {api_key}, Content-Type: application/json } # 永久错误码参数错误、鉴权失败、资源不存在 # 碰到这些直接放弃重试一万次也是错的 self.fatal_codes {1001, 1002, 1004} def send_text(self, wid, wcid, content, max_retry3): 发送文本消息 :param wid: 实例ID登录后获取 :param wcid: 接收方微信ID :param content: 消息内容≤500字符 :param max_retry: 最大重试次数 url f{self.base_url}/sendText payload {wId: wid, wcId: wcid, content: content} for attempt in range(max_retry): try: # 连接超时5s读取超时15s大文件传输慢 resp requests.post(url, jsonpayload, headersself.headers, timeout(5, 15)) result resp.json() code result.get(code) if code 1000: logging.info(f发送成功: wcid{wcid}) return True, result.get(data) if code in self.fatal_codes: logging.warning(f永久错误不重试: {result}) return False, result logging.info(f临时错误第{attempt1}次重试: code{code}) time.sleep(2) except requests.Timeout: logging.warning(f请求超时第{attempt1}次尝试) if attempt max_retry - 1: return False, {error: timeout} time.sleep(2) except Exception as e: logging.error(f请求异常: {e}) return False, {error: str(e)} return False, {error: retry_exhausted} # 使用示例 if __name__ __main__: client EyunClient( base_urlhttp://your-domain, api_keysk-xxxxxxxx ) # 给文件传输助手发条测试消息 ok, data client.send_text( wid0000016e-63eb-f319-0001-ed01076abf1f, wcidfilehelper, contentHello from Eyun )第三步接收消息Webhook接收消息需要在 Eyun 后台配置回调地址必须公网可访问的 HTTPS。微信收到消息后平台会 POST 到你的回调地址from flask import Flask, request import redis import threading import logging app Flask(__name__) r redis.Redis(hostlocalhost, port6379, db0) logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) app.route(/webhook, methods[POST]) def webhook(): Webhook 回调处理 注意必须在3秒内响应超时会触发平台重试 data request.json msg_id data.get(msgId) msg_type data.get(messageType) from_user data.get(fromUserName) content data.get(content, ) # 幂等处理防止重复回调 # Webhook 可能因为网络波动或超时被重复推送 # 用 Redis SET NX 做去重处理过的消息直接跳过 if r.set(feyun:msg:processed:{msg_id}, 1, nxTrue, ex86400): logging.info(f收到新消息: id{msg_id}, type{msg_type}, from{from_user}) # 异步处理业务逻辑避免阻塞回调响应 thread threading.Thread( targetprocess_message, args(msg_type, from_user, content), daemonTrue ) thread.start() else: logging.info(f跳过重复消息: id{msg_id}) # 3秒内必须返回否则 Eyun 会重推 return {code: ok} def process_message(msg_type, from_user, content): 消息业务处理 messageType 对应关系 101000 - 文本消息 101001 - 图片消息 101002 - 语音消息 101003 - 视频消息 try: if msg_type 101000: # 文本 logging.info(f文本消息: {from_user} - {content}) # 这里加你的业务逻辑AI回复、查订单、转人工等 elif msg_type 101001: # 图片 logging.info(f图片消息: {from_user}) else: logging.info(f未处理类型: {msg_type}, from{from_user}) except Exception as e: logging.error(f消息处理异常: {e}) if __name__ __main__: app.run(host0.0.0.0, port8080)常见坑点与避坑指南1. 发送频率控制实测单账号 30 条/分钟无风控40 条开始触发限制。建议单账号发送间隔 ≥ 3 秒批量发送加随机抖动3-6秒单账号单批次 ≤ 100 条单日总量 ≤ 500 条2. 超时设置文本消息连接超时 5s读取超时 10s图片/文件连接超时 5s读取超时 15s大文件传输慢Webhook 回调必须 3 秒内响应3. 错误码处理错误码含义处理方式1000成功正常流程1001参数错误永久错误不重试1002鉴权失败永久错误检查 Token1004资源不存在永久错误不重试其他临时错误可重试最多 3 次4. Token 安全不要硬编码在代码里放环境变量不要提交到 Git泄露后立即刷新不同业务用不同 Token隔离风险5. 日志规范所有日志带msgId和wId排查问题时能快速定位logging.info(f发送消息: msgId{msg_id}, wId{wid}, wcid{wcid}, result{result})适用场景AI 客服自动回复AI Agent 接收消息后通过 API 自动完成意图识别与回复订单通知与触达业务系统触发事件后通过 API 自动向微信用户发送通知群自动运营批量建群、成员管理、定时推送CRM 数据沉淀微信事件通过 Webhook 实时回调至 CRM自动更新客户画像
