企业微信自建应用：Python 实现消息收发功能

安装依赖包：

pip install requests

第三步：集成加解密库

下载微信官方的加解密文件（Python 版本），解压并导入项目。

下载地址：https://developer.work.weixin.qq.com/document/path/90468

主要使用 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)

第五步：配置回调地址

1. 进入企业微信后台配置 - 接收消息。
2. 提供处理请求回调的 URL，随机生成 Token、EncodingAESKey，并将这些数据配置到上方代码中。
3. 确保拥有一个可访问的域名，并在服务器配置反向代理指向项目文件。
4. 上传代码并启动服务，验证回调服务是否正常工作。
5. 保存企业微信的 URL 及其他配置信息。

启动程序后，可通过日志查看消息收发情况。