企业微信自建应用:Python 实现消息收发功能
如何使用 Python 和 FastAPI 实现企业微信自建应用的消息收发功能。主要步骤包括在企业微信后台创建应用并获取凭证,测试基础发送接口,下载官方加解密库处理消息安全,编写 FastAPI 服务处理回调请求及消息解密,最后配置服务器回调地址完成对接。通过该方案可实现自动化消息通知与交互。

如何使用 Python 和 FastAPI 实现企业微信自建应用的消息收发功能。主要步骤包括在企业微信后台创建应用并获取凭证,测试基础发送接口,下载官方加解密库处理消息安全,编写 FastAPI 服务处理回调请求及消息解密,最后配置服务器回调地址完成对接。通过该方案可实现自动化消息通知与交互。

将保存好的信息填入代码段进行初步测试。
import json
import urllib.parse
import requests
# 企业 ID
corpid = 'your_corp_id'
# 应用 ID
agentid = 'your_agent_id'
# 应用 Secret
corpsecret = 'your_secret'
# 接收消息的用户
touser = '@all'
# 企业微信 API 的基础 URL
base = 'https://qyapi.weixin.qq.com'
# 构造函数 (获取 access-token 的 API URL)
access_token_api = urllib.parse.urljoin(base, '/cgi-bin/gettoken')
# 定义请求参数
params = {'corpid': corpid, 'corpsecret': corpsecret}
# 发送 GET 请求获取 access-token
response = requests.get(url=access_token_api, params=params).json()
access_token = response['access_token']
# 构建发送消息的完整 URL
message_send_api = urllib.parse.urljoin(base, f'/cgi-bin/message/send?access_token={access_token}')
# 定义要发送的消息数据(文本格式)
data = {
'touser': touser,
'msgtype': 'text',
'agentid': agentid,
'text': {'content': '测试数据:hello world!'}
}
# 发送 POST 请求以发送消息
response = requests.post(url=message_send_api, data=json.dumps(data)).json()
if response['errcode'] == 0:
print('发送成功')
else:
print(response)
安装依赖包:
pip install requests
下载微信官方的加解密文件(Python 版本),解压并导入项目。
主要使用 WXBizMsgCrypt3.py 文件。注意 Crypto.Cipher 包已不再维护,建议安装最新版加解密工具包:
pip install pycryptodome
同时需要导入错误码提示文件 ierror.py。
本示例使用 FastAPI 实现接口开发。逻辑代码中需配置接收消息设置里的 Token、EncodingAESKey、CorpID 等信息。
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import PlainTextResponse, Response
from WXBizMsgCrypt3 import WXBizMsgCrypt
import xmltodict
import logging
import time
import xml.etree.ElementTree as ET
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
# 企业微信配置
TOKEN = 'your_token'
ENCODING_AES_KEY = 'your_encoding_aes_key'
CORP_ID = 'your_corp_id'
try:
wxcpt = WXBizMsgCrypt(TOKEN, ENCODING_AES_KEY, CORP_ID)
except Exception as e:
logger.error(f"初始化 WXBizMsgCrypt 失败:{str(e)}")
raise
@app.get("/callback")
async def verify_url(msg_signature: str, timestamp: str, nonce: str, echostr: str):
"""验证 URL 有效性"""
try:
logger.info(f"收到验证请求:msg_signature={msg_signature}, timestamp={timestamp}, nonce={nonce}")
ret, sEchoStr = wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr)
if ret == 0:
logger.info("URL 验证成功")
return PlainTextResponse(content=sEchoStr)
else:
logger.error(f"URL 验证失败,错误码:{ret}")
raise HTTPException(status_code=400, detail="验证失败")
except Exception as e:
logger.error(f"验证过程发生错误:{str(e)}")
raise HTTPException(status_code=500, detail="服务器内部错误")
@app.post("/callback")
async def receive_message(request: Request):
"""接收并处理企业微信消息"""
try:
body = await request.body()
msg_signature = request.query_params.get("msg_signature")
timestamp = request.query_params.get("timestamp")
nonce = request.query_params.get("nonce")
if not all([msg_signature, timestamp, nonce]):
raise HTTPException(status_code=400, detail="缺少必要的参数")
logger.info(f"收到消息推送:msg_signature={msg_signature}, timestamp={timestamp}, nonce={nonce}")
# 解密消息
ret, sMsg = wxcpt.DecryptMsg(body, msg_signature, timestamp, nonce)
if ret != 0:
logger.error(f"消息解密失败,错误码:{ret}")
raise HTTPException(status_code=400, detail="消息解密失败")
# 解析 XML 消息
xml_dict = xmltodict.parse(sMsg)
logger.info(f"解密后的消息内容:{xml_dict}")
xml_content = xml_dict['xml']
to_user_name = xml_content.get('ToUserName')
from_user_name = xml_content.get('FromUserName')
create_time = xml_content.get('CreateTime')
msg_type = xml_content.get('MsgType')
content = xml_content.get('Content')
msg_id = xml_content.get('MsgId')
agent_id = xml_content.get('AgentID')
logger.info(f"收到消息:{content}")
# 构造回复消息
reply_content = content.replace('吗', '').replace('?', '!').replace('?', '!')
current_time = str(int(time.time()))
reply_msg = f"""<xml>
<ToUserName><![CDATA[{from_user_name}]]></ToUserName>
<FromUserName><![CDATA[{to_user_name}]]></FromUserName>
<CreateTime>{current_time}</CreateTime>
<MsgType><![CDATA[text]]></MsgType>
<Content><![CDATA[{reply_content}]]></Content>
<MsgId>{msg_id}</MsgId>
<AgentID>{agent_id}</AgentID>
</xml>"""
# 加密回复消息
ret, encrypted_msg = wxcpt.EncryptMsg(reply_msg, nonce, current_time)
if ret != 0:
logger.error(f"消息加密失败,错误码:{ret}")
raise HTTPException(status_code=500, detail="消息加密失败")
logger.info("成功构造并加密回复消息")
return Response(content=encrypted_msg, media_type="application/xml")
except Exception as e:
logger.error(f"处理消息时发生错误:{str(e)}")
raise HTTPException(status_code=500, detail="服务器内部错误")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=5000)
启动程序后,可通过日志查看消息收发情况。

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML 转 Markdown 互为补充。 在线工具,Markdown 转 HTML在线工具,online
将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML 转 Markdown在线工具,online
通过删除不必要的空白来缩小和压缩JSON。 在线工具,JSON 压缩在线工具,online