企业微信机器人 API:接口调用鉴权、消息格式与异常处理详解
·
调用官方 API 最让人头疼也最容易踩坑的环节有两个:第一是怎么证明“我是我”(接口鉴权);第二是为什么辛辛苦苦写的代码发出去总是报错(错误码与异常处理)。今天咱们把企业微信接口调用的底层逻辑一次性彻底说清楚。
核心技术拆解
-
全局凭证 Access Token:企业微信几乎所有的 API(除了群机器人 Webhook)都需要在请求地址里带上一个
access_token。这个 Token 的有效期通常是 2 小时,绝对不能每次请求都去重新申请,否则会瞬间触发频率限制(Rate Limit)。正确的做法是:全局缓存,提前刷新。 -
错误码机制(errcode):企业微信的所有接口无论成功与否,都会返回一个 JSON,其中必定包含
errcode和errmsg。只要errcode !== 0,就代表出错了。必须针对常见的错误码(如 40014:不合法的 access_token,45009:接口调用超过限制)编写专门的重试或降级处理逻辑。
下面我们用 Python 编写一个兼顾 Token 缓存机制与异常错误处理的图文消息发送封装类。
完整代码实现(Python)
use-strict
import requests
import time
import json
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class WeChatClient:
def __init__(self, corpid, corpsecret):
self.corpid = corpid
self.corpsecret = corpsecret
self.access_token = None
self.token_expires_at = 0
def get_access_token(self):
"""
获取 Access Token,内置内存缓存机制,避免频繁请求
"""
current_time = time.time()
# 如果 Token 还有效(提前 200 秒过期刷新),直接返回缓存值
if self.access_token and current_time < self.token_expires_at:
return self.access_token
url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.corpid}&corpsecret={self.corpsecret}"
try:
response = requests.get(url, timeout=5)
res_data = response.json()
if res_data.get("errcode") == 0:
self.access_token = res_data.get("access_token")
# 凭证有效期一般为 7200 秒,这里设定在 7000 秒后强制过期刷新
expires_in = res_data.get("expires_in", 7200)
self.token_expires_at = current_time + expires_in - 200
logging.info("成功刷新并获取到新的 Access Token")
return self.access_token
else:
logging.error(f"获取 Access Token 失败: {res_data}")
return None
except Exception as e:
logging.error(f"获取 Access Token 网络请求异常: {e}")
return None
def send_app_message(self, user_id, title, description, url):
"""
向指定企业员工发送图文类型的应用消息
"""
token = self.get_access_token()
if not token:
logging.error("无法发送消息,因为未获取到有效的 Access Token。")
return False
api_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
payload = {
"touser": user_id,
"msgtype": "news",
"agentid": 1000002, # 替换成你自建应用的 AgentID
"news": {
"articles": [
{
"title": title,
"description": description,
"url": url,
"picurl": "https://example.com/logo.png"
}
]
}
}
try:
response = requests.post(api_url, data=json.dumps(payload), timeout=5)
result = response.json()
errcode = result.get("errcode")
if errcode == 0:
logging.info(f"成功向用户 {user_id} 发送图文消息!")
return True
elif errcode == 40014:
# 专属处理:Token 过期或失效,清空缓存后递归重试一次
logging.warning("Access Token 已失效,正在尝试重新获取并重试...")
self.access_token = None
self.token_expires_at = 0
return self.send_app_message(user_id, title, description, url)
else:
logging.error(f"发送应用消息出错,错误码: {errcode}, 原因: {result.get('errmsg')}")
return False
except Exception as e:
logging.error(f"发送应用消息时发生网络异常: {e}")
return False
# --- 实战演练 ---
if __name__ == "__main__":
client = WeChatClient("YOUR_CORPID", "YOUR_CORPSECRET")
# client.send_app_message("ZhangSan", "系统运维周报发布", "点击查看本周服务器运行健康报告", "https://example.com/report")
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐

所有评论(0)