企业微信机器人 + 自部署 AI 服务 30 分钟复现

1. 登录 work.weixin.qq.com（企业管理员）→ 应用管理 → 自建 → 创建应用
2. 填应用名/头像 → 创建后进入应用详情，记录三个值：
   • CorpID（我的企业 → 企业信息 → 最底部）
   • AgentId（应用详情页，纯数字）
   • Secret（应用详情 → Secret → 查看，关页面后要重置才能再看）
3. 应用详情 → 企业可信 IP：填你服务器的公网出口 IP（少了它调 API 报 60020，见文末报错 3）
4. 应用详情 → 接收消息 → 设置 API 接收：填 Token（自己编一串随机字符串）和 EncodingAESKey（点随机生成）
5. 「接收消息」页会要你填回调 URL——先不填，第 3 步用 ngrok 拿到公网地址再回来填

企业微信回调验证比飞书多一层：GET 请求带 msg_signature，echostr 是 AES 加密的，必须先解密再原样返回。这段代码直接拷：

import hashlib, base64, struct
from flask import Flask, request, Response
from Crypto.Cipher import AES

app = Flask(__name__)

TOKEN = "你填的Token"
AES_KEY = "你填的EncodingAESKey"  # 43位，代码里补一个"="
CORP_ID = "你的CorpID"

def decrypt_msg(encrypted):
    key = base64.b64decode(AES_KEY + "=")
    cipher = AES.new(key, AES.MODE_CBC, key[:16])
    plain = cipher.decrypt(base64.b64decode(encrypted))
    pad = plain[-1]
    plain = plain[:-pad]                      # 去 PKCS7 填充
    plain = plain[16:]                        # 去 16 字节随机串
    msg_len = struct.unpack("!i", plain[:4])[0]
    return plain[4:4 + msg_len].decode()      # 消息体

def verify_signature(signature, timestamp, nonce, echostr):
    s = "".join(sorted([TOKEN, timestamp, nonce, echostr]))
    return hashlib.sha1(s.encode()).hexdigest() == signature

@app.route("/wecom/callback", methods=["GET"])
def verify_url():
    args = request.args
    if verify_signature(args["msg_signature"], args["timestamp"],
                        args["nonce"], args["echostr"]):
        return Response(decrypt_msg(args["echostr"]), mimetype="text/plain")
    return "signature error", 403

@app.route("/wecom/callback", methods=["POST"])
def receive_msg():
    # 企业微信 POST 的消息体是 { Encrypt: "..." }，同样先验签再解密
    # 解密后是 XML：<MsgType>text</MsgType> + Content + FromUserName...
    # 这里先打印日志，第 4 步换成 AI 调用
    return Response("success", mimetype="text/plain")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

⚠️ 依赖：pip install flask pycryptodome。企业微信要求回调返回纯文本 success（不是 JSON），返回别的会一直重试推送。

1. 运行：ngrok http 8080（或 cloudflared / bore 都行）
2. 复制 https://xxxx.ngrok-free.app 这样的 URL
3. 回企业微信「接收消息」→ 填 https://xxxx.ngrok-free.app/wecom/callback → 保存
4. 保存时企业微信会立刻 GET 一次验证 URL——你的 verify_url() 返回解密后的 echostr 就通过，看到「保存成功」即通

POST 回调解密后是 XML，用 xml.etree 解析，然后调 AI，再用「发送应用消息」API 回给员工：

import requests, xml.etree.ElementTree as ET

def get_access_token():
    url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
    r = requests.get(url, params={"corpid": CORP_ID, "corpsecret": SECRET},
                     timeout=10)
    return r.json()["access_token"]          # 2 小时有效，建议缓存

def send_to_user(user_id, content):
    token = get_access_token()
    url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
    requests.post(url, params={"access_token": token}, json={
        "touser": user_id,                    # UserID，不是手机号/姓名
        "msgtype": "text",
        "agentid": AGENT_ID,
        "text": {"content": content}
    }, timeout=10)

# 在 receive_msg() 里（解密出 xml_str 后）：
root = ET.fromstring(xml_str)
user_id = root.findtext("FromUserName")      # 发消息的员工 UserID
content = root.findtext("Content")           # 员工发的文字

reply = ask_ai(content)                       # 调你的自部署 AI 服务
send_to_user(user_id, reply)

第一次跑通后，再加：

• 记忆（redis 存 user_id → 对话历史）
• 知识库检索（先查知识库再调 LLM）
• Skill 调用（按模板处理固定流程）

1. 应用详情 → 可见范围：先选一个部门（如 IT 部）小范围试点
2. 把 ngrok 地址换成本机 Nginx 反代（proxy_pass http://127.0.0.1:8080）或直接部署到服务器
3. 企业微信「我的企业 → 微信插件」——员工在企业微信里就能搜到你的应用
4. 给应用发条消息 → 收到 AI 回复 → 完成