企业微信 AI 机器人怎么搭建？三种方案对比（附代码）

管理后台 →「应用管理」→「自建」→「创建应用」，可见范围先选一个小部门做灰度。记下三个值：

• corpid：「我的企业」页 → 企业 ID
• AgentId：应用详情页
• Secret：应用详情页（只显示一次，立即存入密钥管理，不要进代码库）

应用详情 →「接收消息」→「设置 API 接收」，填三项：URL（你服务器的 HTTPS 地址）、Token、EncodingAESKey（可随机生成）。点保存时，企微会向你的 URL 发一个 GET 验证请求，要求验签通过并原样返回解密后的 echostr。

# callback_verify.py —— 回调 URL 验证（GET）
# 验签规则：对 token、timestamp、nonce、密文四者排序拼接后取 sha1
import hashlib

def check_signature(token, timestamp, nonce, encrypt_msg):
    items = sorted([token, timestamp, nonce, encrypt_msg])
    return hashlib.sha1("".join(items).encode()).hexdigest()

@app.route("/wecom/callback", methods=["GET"])
def verify():
    encrypt = request.args["echostr"]
    expect = request.args["msg_signature"]
    actual = check_signature(TOKEN, request.args["timestamp"],
                             request.args["nonce"], encrypt)
    if actual != expect:
        return "bad signature", 403
    # 用官方加解密库（WXBizMsgCrypt）做 AES 解密，拿到明文后原样返回
    return decrypt_echostr(encrypt, AES_KEY)

验签失败的三大原因：Token 填错、服务器时间漂移（做 NTP 对时）、只写了 POST 没写 GET。调试期把「排序后参与计算的四个字符串」打进日志，一眼就能看出差在哪。

员工在应用里发消息后，企微会 POST 一段加密 XML 到你的回调 URL。两个硬约束：5 秒内必须响应（超时企微会重试，最多三次），重试的消息内容一模一样（不去重就会重复回复）。所以标准写法是：收到立即去重并返回，AI 处理放异步，算完再主动推给员工。

# wecom_bot.py —— 收消息：MsgId 去重 + 异步处理
import threading

replying = set()          # 生产环境换 Redis，防多实例重复处理

@app.route("/wecom/callback", methods=["POST"])
def on_message():
    xml = decrypt_msg(request.data, request.args)   # 官方加解密库解开 XML
    msg_id = xml.get("MsgId")
    if msg_id in replying:
        return "success"        # 重试的重复消息，直接吞掉
    replying.add(msg_id)
    threading.Thread(target=handle, args=(xml["FromUserName"],
                                          xml["Content"], msg_id)).start()
    return "success"            # 立即响应，不等 AI

def handle(user, question, msg_id):
    try:
        answer = ask_ai(question)     # 你自己的 AI 服务/知识库
        send_text(user, answer)       # 用应用消息接口主动回复
    except Exception:
        send_text(user, "抱歉，刚才处理超时了，请再问一次。")
    finally:
        replying.discard(msg_id)

access_token 用 corpid + Secret 换取，有效期 7200 秒，且有获取频率限制——每个地方都现取是典型反模式，必须集中缓存、单点刷新。另外：调用发送接口的服务器 IP 必须先加入应用的「企业可信 IP」，否则报错 60020。

# wecom_client.py —— token 集中缓存 + 发送文本消息
import time, requests

_token, _ts = "", 0

def get_token():
    global _token, _ts
    if _token and time.time() - _ts < 7000:      # 留 200 秒余量
        return _token
    r = requests.get("https://qyapi.weixin.qq.com/cgi-bin/gettoken",
                     params={"corpid": CORP_ID, "corpsecret": SECRET},
                     timeout=5).json()
    _token, _ts = r["access_token"], time.time()
    return _token

def send_text(user, content):
    requests.post(
        "https://qyapi.weixin.qq.com/cgi-bin/message/send",
        params={"access_token": get_token()},
        json={"touser": user, "msgtype": "text", "agentid": AGENT_ID,
              "text": {"content": content[:2000]}},     # 超长先截断/分段
        timeout=5)

到这里方案 A 就通了。把 ask_ai() 换成你的 RAG 知识库或大模型服务，就是一个能用的企业 AI 助手。