企业微信 API 二次开发:外部群主动推送消息
·
在企业微信的常规开发体系中,外部群(包含企业外部联系人的群聊)的消息推送一直属于限制较多的环节。传统的群机器人(Webhook)大多仅支持内部群或被动响应,无法满足主动、灵活控制外部群聊的需求。
本文拆解如何通过纯 HTTP RESTful 接口架构,实现对外部群的主动消息精准推送与系统集成。
一、 整体技术实现架构
实现“主动推送”的核心逻辑在于:将底层消息调度与上层业务接口解耦。业务系统只需关注发送什么内容、发给谁,无需感知复杂的连接维护与通信细节。
┌────────────────┐ HTTP POST ┌────────────────┐ API 调起 ┌────────────────┐
│ 内部业务系统 │ ─────────────────> │ 二次开发网关 │ ────────────────> │ 企业微信外部群 │
│ (CRM/ERP/运维) │ (JSON Payload) │ (API Engine) │ (Direct Push) │ (External Group)│
└────────────────┘ └────────────────┘ └────────────────┘
- 业务发起端:由业务系统(如订单系统、告警系统)在触发事件后,组装 JSON 格式的数据包。
- 二次开发网关:校验 Token 鉴权,解析目标群 ID(
chat_id)与消息类型。 - 消息投递层:网关接收到请求后,将消息直接投递进指定的外部群中。
二、 核心 API 请求与数据结构设计
为了保证接口的易用性与扩展性,推荐采用标准的 RESTful 接口风格。
1. 基础请求头 (Headers)
POST /v1/group/send_message HTTP/1.1
Host: api.your-domain.com
Content-Type: application/json
Authorization: Bearer your_access_token_here
2. 发送文本消息 Payload
{
"chat_id": "external_chat_8899112233",
"msg_type": "text",
"text": {
"content": "【系统通知】您的售后工单 #20260813 已处理完成,请在附件中查看明细。",
"mentioned_list": ["@all"]
}
}
3. 发送 Markdown 富文本 Payload
{
"chat_id": "external_chat_8899112233",
"msg_type": "markdown",
"markdown": {
"content": "### 📊 每日数据汇总\n> 日期:<font color=\"comment\">2026-08-13</font>\n\n* 今日新增客户:**128** 人\n* 外部群活跃度:**94%**\n\n详情请查看[控制台文档](https://doc.qiweapi.com/)"
}
}
三、 Python 核心调用实现
以下为生产环境下具备超时控制与重试机制的完整调用示例代码:
import requests
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class QiWeiGroupClient:
def __init__(self, base_url: str, token: str):
self.base_url = base_url.rstrip('/')
self.headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {token}"
}
def send_text_msg(self, chat_id: str, content: str, at_all: bool = False) -> bool:
url = f"{self.base_url}/v1/group/send_message"
payload = {
"chat_id": chat_id,
"msg_type": "text",
"text": {
"content": content,
"mentioned_list": ["@all"] if at_all else []
}
}
# 指数退避重试逻辑
max_retries = 3
for attempt in range(max_retries):
try:
response = requests.post(url, json=payload, headers=self.headers, timeout=5)
res_json = response.json()
if response.status_code == 200 and res_json.get("code") == 200:
logging.info(f"消息成功发送至群 [{chat_id}]")
return True
else:
logging.warning(f"发送失败(尝试 {attempt + 1}/{max_retries}): {res_json.get('msg')}")
except requests.RequestException as e:
logging.error(f"网络请求异常(尝试 {attempt + 1}/{max_retries}): {e}")
time.sleep(2 ** attempt) # 1s, 2s, 4s 退避
return False
# 执行代码
if __name__ == "__main__":
CLIENT = QiWeiGroupClient(
base_url="https://api.your-domain.com",
token="your_secret_token"
)
TARGET_CHAT = "external_chat_8899112233"
MSG = "这是一条来自二次开发接口的主动推送测试消息。"
CLIENT.send_text_msg(chat_id=TARGET_CHAT, content=MSG)
四、 避坑指南与高可用设计
在实际二次开发部署时,外部群主动推送需要重点解决频控与并发问题:
- 接口频率控制 (Rate Limiting)
- 企微对外部群消息频控极其严格。建议在二次开发网关层引入 令牌桶算法 (Token Bucket)。
- 单个群建议发送频率控制在 不超过 1 条/秒,批量群推送时必须设置消费队列(如 Redis Stream 或 RabbitMQ)做削峰填谷。
- 异步队列处理
- 业务端发起 HTTP 请求后,网关应立即返回
202 Accepted和task_id,避免业务端因等待长连接而超时。 - 依靠后台 Worker 异步消费队列并执行实际的 API 投递。
- 异常状态捕获
- 若返回错误码提示
chat_id无效,需检查机器人或账号是否已被移出该外部群。 - 建立自动风控熔断机制:当连续 5 次出现频控限制错误码时,自动暂停该账号所在通道的推送 10 分钟。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐


所有评论(0)