调用官方 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")