不用改網路、不用買伺服器，30 分鐘把 AI Agent 接進企業微信/飛書/釘釘，支援 Stream 模式即時推送
接入企業 IM 有兩條路徑，選錯了要多走很多彎路：
| 模式 | 原理 | 是否需要公網 IP | 適合場景 |
|---|---|---|---|
| 長連線（Agent→IM） | AI Agent 主動推訊息到 IM | ✅ 需要 | 有固定出口伺服器 |
| Webhook 回調（IM→Agent） | IM 把訊息 POST 到你的服務 | ❌ 不需要 | 內網/無伺服器/不想暴露連接埠 |
| 混合模式 | Webhook 接收 + Agent 回調推送 | ✅ Agent 側需要 | 需要雙向即時對話 |
對大多數企業來說，Webhook 回調模式是最佳起點：不需要改防火牆、不需要買雲端伺服器，Mule Agent 提供了穩定的回調接收端，使用者發訊息到 IM，IM 自動轉發給 Agent，Agent 處理完再推回 IM。
https://your-agent-domain.com/webhook/feishu這個位址就是你要填入 IM 平台設定裡的 Outgoing 回調 URL。
im.message.receive（接收訊息）以下是一個最小可用的 Flask 服務，監聽 IM 平台回調並轉發給 Mule Agent 處理：
import os
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
# Mule Agent Webhook Token（從管理後台取得）
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)
這段程式碼放到一台有公網 IP 的伺服器上（或者用雲端函數/容器服務），即可完成訊息的接收和轉發。生產環境別忘了加：簽章驗證、超時處理、日誌記錄。
傳統模式是等 Agent 處理完再一次性推訊息，使用者要等 5-30 秒，體驗很差。Stream 模式讓 AI 逐句推送，使用者看到的是「正在輸入」效果：
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 模式特別適合長文字生成（報告生成、程式碼編寫、文件總結），使用者在等待過程中就能看到輸出，心理體驗比乾等好得多。
X-Lark-Signature，服務沒有驗簽，飛書拒絕重試（飛書對無回應端點會逐漸降級推送頻率）。
{"text": {"content": "..."}}，釘釘父層 key 不同，企微又是另一種結構。