不用改网络、不用买服务器，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。
下面是一个最小可用的 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():
# 飞书签名验证（生产环境必须验证）
# headers["X-Lark-Signature"] = request.headers.get("X-Lark-Signature", "")
body = request.json
# 提取消息内容（不同平台字段名略有差异）
user_msg = body.get("text", {}).get("content", "")
# 转发给 Mule Agent
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
)
# 按 SSE 格式逐块推给飞书
for line in resp.iter_lines():
if line:
# Mule Agent 返回 SSE 格式：data: {"content":"xxx"}\n\n
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，服务没有验签，飞书拒绝重试（飞书对无响应 endpoint 会逐渐降级推送频率）。
{"text": {"content": "..."}}，钉钉是 {"text": {"content": "..."}}（但父级 key 不同），企微是另一个结构。