Connect AI Agent to Enterprise IM: 4 Steps for Feishu, DingTalk & WeCom

• Feishu: Use a Feishu self-built app, fill in App ID + App Secret
• DingTalk: Internal enterprise app, fill in Corp ID + App Key + App Secret
• WeCom (企业微信): Self-built enterprise app, fill in Corp ID + Agent ID + App Secret

https://your-agent-domain.com/webhook/feishu

1. Go to Feishu Open Platform → find your self-built app
2. Add App Capabilities → Bot, enable bot capability
3. Event Subscriptions → check im.message.receive_v1
4. Request URL → paste the Mule Agent callback URL (HTTPS required)
5. Publish the app version, wait for enterprise admin approval

1. Go to DingTalk Open Platform → Internal Development → find your app
2. Message Push → Bot → Enable custom bot
3. Message Receive → configure callback URL (Token + EncodingAESKey auto-generated by Mule Agent)
4. Subscribe to event: im.message.receive
5. Save — DingTalk will send a test message to verify connectivity

1. WeCom Admin Console → Apps → find your self-built app
2. Trusted IPs → add Mule Agent server IP (can skip if Agent is on intranet)
3. Receive Messages → configure callback URL (HTTPS required)
4. Enable App Messaging permission, save

Minimal Flask service that receives IM callbacks and forwards to Mule Agent:

import os
from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

MULE_AGENT_TOKEN = os.environ.get("MULE_AGENT_TOKEN", "your-agent-token")
MULE_AGENT_API = "https://your-agent-domain.com/api/chat"

@app.route("/webhook/feishu", methods=["POST"])
def feishu_webhook():
    body = request.json
    user_msg = body.get("text", {}).get("content", "")
    resp = requests.post(
        MULE_AGENT_API,
        headers={"Authorization": f"Bearer {MULE_AGENT_TOKEN}"},
        json={"message": user_msg},
        timeout=30
    )
    agent_reply = resp.json().get("reply", "")
    return jsonify({"msg_type": "text", "content": {"text": agent_reply}})

@app.route("/webhook/dingtalk", methods=["POST"])
def dingtalk_webhook():
    body = request.json
    user_msg = body.get("text", {}).get("content", "")
    resp = requests.post(
        MULE_AGENT_API,
        headers={"Authorization": f"Bearer {MULE_AGENT_TOKEN}"},
        json={"message": user_msg},
        timeout=30
    )
    agent_reply = resp.json().get("reply", "")
    return jsonify({"msgtype": "text", "text": {"content": agent_reply}})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=False)

Deploy this on a server with public access (or use cloud functions / containers). For production: add signature verification, timeout handling, and logging.

Traditional mode waits for the full response before sending — users wait 5-30 seconds. Stream mode pushes chunks as they're generated, so users see "typing" feedback:

import os, json
from flask import stream_with_context, Response
import requests

MULE_AGENT_API = "https://your-agent-domain.com/api/chat/stream"
MULE_AGENT_TOKEN = os.environ.get("MULE_AGENT_TOKEN", "your-agent-token")

@app.route("/webhook/feishu/stream", methods=["POST"])
def feishu_stream():
    body = request.json
    user_msg = body.get("text", {}).get("content", "")

    def generate():
        resp = requests.post(
            MULE_AGENT_API,
            headers={
                "Authorization": f"Bearer {MULE_AGENT_TOKEN}",
                "Accept": "text/event-stream"
            },
            json={"message": user_msg},
            stream=True, timeout=60
        )
        for line in resp.iter_lines():
            if line:
                yield f"data: {json.dumps({'msg_type':'text','content':line.decode()})}\n\n"

    return Response(
        stream_with_context(generate()),
        mimetype="text/event-stream"
    )

Stream mode shines for long-form generation (reports, code, document summaries) — users get instant feedback instead of staring at a blank screen.