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

企业微信接入WorkBuddy全流程指南

news 2026/6/1 11:27:04

企业微信接入 WorkBuddy 的技术实现

WorkBuddy 作为一款智能工作助手，接入企业微信可显著提升企业内部协作效率。以下从配置流程、API 调用、消息处理等角度，结合代码示例详细说明技术实现方案。

企业微信应用配置

在企业微信管理后台创建自建应用，获取以下关键信息：

CorpID：企业唯一标识
AgentID：应用ID
Secret：应用密钥

# 示例：配置参数存储 WORKBUDDY_CONFIG = { "corp_id": "wwxxxxxx", "agent_id": 1000002, "secret": "xxxxxxxxxxxxxxxx" }

获取 Access Token

调用企业微信API需携带有效的Access Token，通过GET请求获取并缓存：

import requests import time def get_access_token(corp_id, secret): url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corp_id}&corpsecret={secret}" response = requests.get(url).json() if response["errcode"] == 0: return response["access_token"] else: raise Exception(f"Token获取失败: {response}") # 缓存Token示例 token_cache = { "token": None, "expire_time": 0 } def ensure_valid_token(): if time.time() < token_cache["expire_time"]: return token_cache["token"] token = get_access_token(WORKBUDDY_CONFIG["corp_id"], WORKBUDDY_CONFIG["secret"]) token_cache.update({"token": token, "expire_time": time.time() + 7000}) # 有效期7200秒 return token

消息接收与解析

企业微信通过回调推送用户消息，需配置可信域名并验证签名。以下为消息解密示例：

from Crypto.Cipher import AES import base64 import xml.etree.ElementTree as ET def decrypt_msg(encrypt_msg, encoding_aes_key): key = base64.b64decode(encoding_aes_key + "=") cipher = AES.new(key, AES.MODE_CBC, key[:16]) decrypted = cipher.decrypt(base64.b64decode(encrypt_msg)) pad = ord(decrypted[-1:]) content = decrypted[:-pad].decode("utf-8") return ET.fromstring(content.split(">")[1] + ">") # 提取XML内容

主动发送消息

通过企业微信API向用户发送文本消息：

def send_text_message(user_id, content): token = ensure_valid_token() url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}" payload = { "touser": user_id, "msgtype": "text", "agentid": WORKBUDDY_CONFIG["agent_id"], "text": {"content": content}, "safe": 0 } response = requests.post(url, json=payload).json() if response["errcode"] != 0: print(f"消息发送失败: {response}")

用户身份验证

通过OAuth2.0获取用户详细信息，实现WorkBuddy与企业微信账号绑定：

def get_user_info(code): token = ensure_valid_token() url = f"https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo?access_token={token}&code={code}" response = requests.get(url).json() if response["errcode"] == 0: return response["UserId"] # 企业微信用户唯一标识

事件处理框架

处理订阅、菜单点击等事件需实现回调接口：

from flask import Flask, request app = Flask(__name__) @app.route("/workbuddy/callback", methods=["POST"]) def handle_callback(): msg_signature = request.args.get("msg_signature") timestamp = request.args.get("timestamp") nonce = request.args.get("nonce") data = request.get_data() # 验证签名并解密（略） xml_data = decrypt_msg(data, encoding_aes_key) msg_type = xml_data.find("MsgType").text if msg_type == "event": event = xml_data.find("Event").text if event == "click": handle_menu_click(xml_data) return "success"