How to Build a WeCom AI Bot? 3 Approaches Compared (With Code)

In the admin console → "Applications" → "Custom" → "Create". For the visible range, start with one small department as a canary. Write down three values:

• corpid: "My Company" page → Corp ID
• AgentId: on the app detail page
• Secret: app detail page (shown once — store it in a secrets manager, never in git)

App detail → "Receive Messages" → "API endpoint". Fill in: URL (your HTTPS endpoint), Token, and EncodingAESKey (can be auto-generated). On save, WeCom sends a GET verification request to your URL: you must verify the signature and return the decrypted echostr as-is.

# callback_verify.py — callback URL verification (GET)
# Signature: sha1 of the sorted concatenation of token, timestamp, nonce, ciphertext
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
    # AES-decrypt with the official crypto library, return plaintext as-is
    return decrypt_echostr(encrypt, AES_KEY)

The three usual causes of verification failure: wrong Token, server clock drift (run NTP), or implementing POST only and forgetting the GET verification. During debugging, log the four strings that go into the signature — the mismatch becomes obvious.

When an employee messages the app, WeCom POSTs an encrypted XML to your callback URL. Two hard constraints: you must respond within 5 seconds (otherwise WeCom retries, up to three times), and retried messages are identical (no dedup means duplicated replies). The standard pattern: dedupe and return immediately, process asynchronously, then push the reply through the send-message API.

# wecom_bot.py — receive: MsgId dedup + async processing
import threading

replying = set()          # use Redis in production for multi-instance safety

@app.route("/wecom/callback", methods=["POST"])
def on_message():
    xml = decrypt_msg(request.data, request.args)   # official crypto library
    msg_id = xml.get("MsgId")
    if msg_id in replying:
        return "success"        # retried duplicate — swallow it
    replying.add(msg_id)
    threading.Thread(target=handle, args=(xml["FromUserName"],
                                          xml["Content"], msg_id)).start()
    return "success"            # respond immediately, don't wait for the AI

def handle(user, question, msg_id):
    try:
        answer = ask_ai(question)     # your own AI / knowledge-base service
        send_text(user, answer)       # push via the app message API
    except Exception:
        send_text(user, "Sorry, that request timed out. Please try again.")
    finally:
        replying.discard(msg_id)

The access_token is exchanged with corpid + Secret, valid for 7200 seconds, with a refresh rate limit — fetching it fresh at every call site is an anti-pattern. Cache centrally, refresh from one place. Also: the outbound IP of your server must be added to the app's trusted IP list, or send calls fail with error 60020.

# wecom_client.py — centralized token cache + send text message
import time, requests

_token, _ts = "", 0

def get_token():
    global _token, _ts
    if _token and time.time() - _ts < 7000:      # 200s safety margin
        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]}},     # truncate/split long text
        timeout=5)

That's Approach A working end to end. Swap ask_ai() for your RAG knowledge base or LLM service and you have a usable enterprise assistant.