实战开发一个飞书机器人：webhook 告警推送 + lark-oapi 长连接 AI 对话，完整项目从零跑通

一、先分清：你要的是"推送器"还是"对话机器人"

很多人第一次做飞书机器人，在群里加了个"自定义机器人"，拿到 webhook 地址兴冲冲地想做"群里 @它它就回答"，结果发现机器人根本收不到消息——因为群自定义机器人从设计上就只能"往外推"，不能"往里收"。

飞书机器人其实分两条完全不同的路线：

维度

群自定义机器人（Webhook）

企业自建应用（SDK）

一句话：要它"说话"用 webhook，要它"听懂你说话"用企业自建应用。

二、路线 A：群自定义机器人 webhook 推送（10 分钟）

2.1 建机器人拿 webhook

https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxxxxxxxxxxx

2.2 用 Python 推一条消息

import requests

WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/你的hook"

def send_text(text: str):
resp = requests.post(
WEBHOOK,
json={"msg_type": "text", "content": {"text": text}},
timeout=5,
)
print(resp.status_code, resp.json())

if __name__ == "__main__":
send_text("[告警] 生产环境 CPU 超过 90%，请值班同学关注")

curl 版一行就能测：

curl -X POST -H "Content-Type: application/json" \\
-d '{"msg_type":"text","content":{"text":"[测试] 机器人上线了"}}' \\
https://open.feishu.cn/open-apis/bot/v2/hook/你的hook

这就是为什么 webhook 机器人适合做 CI/CD 通知、监控告警——它就是一个"HTTP POST 到群里"的简单通道，不需要任何服务端常驻进程。

三、路线 B：企业自建应用 + 长连接，做 AI 对话机器人

接下来是重点。我们要做的效果是：在群里 @这个机器人 问一句，它调大模型回答，再把结果发回群里。

3.1 创建应用并开权限

3.2 为什么用"长连接"而不是公网回调

传统做法是你在事件订阅里填一个公网可访问的 HTTPS URL，飞书把事件 POST 过来。这意味着你得有一台公网服务器 + 域名 + 证书，本地开发都得用 ngrok 内网穿透，很麻烦。

2026 年飞书 SDK 提供了**长连接（WebSocket 全双工）**接收事件的能力（官方 2026-06-11 更新文档）：你的服务器主动和飞书建立一条 WebSocket 通道，事件顺着这条通道推下来。不需要公网回调域名，本地一台能出网的机器就能跑，这对个人开发者和内网开发友好太多。

3.3 完整可跑代码（lark-oapi Python）

先装 SDK：

pip install lark-oapi requests

下面这个骨架就是官方 echo_bot 改造版，把"原样回显"换成了"调大模型回答"：

import lark_oapi as lark
from lark_oapi.api.im.v1 import (
P2ImMessageReceiveV1,
CreateMessageRequest,
CreateMessageRequestBody,
)

APP_ID = "cli_xxxxxxxx"
APP_SECRET = "yyyyyyyy"

# —- 这里换成你自己的 LLM 调用，下面是占位 —-
def call_llm(prompt: str) -> str:
# 真实项目里换成 OpenAI 兼容接口 / DeepSeek / 本地 vLLM
return f"你问的是：{prompt}\\n（这是一个占位回复，接上 LLM 后就是真智能）"

# —- 收到群消息后的处理 —-
def on_message(data: P2ImMessageReceiveV1) -> None:
try:
msg = data.event.message
chat_id = msg.chat_id
# 取用户文本（简化：msg_type=text）
content = msg.content # 形如 {"text":"@_user_1 你好"}
user_text = content.replace("@_user_1", "").strip()

answer = call_llm(user_text)

# 用同一个 client 发回群里
client = lark.Client.builder().app_id(APP_ID).app_secret(APP_SECRET).build()
req = (
CreateMessageRequest.builder()
.receive_id_type("chat_id")
.request_body(
CreateMessageRequestBody.builder()
.receive_id(chat_id)
.msg_type("text")
.content('{"text":"%s"}' % answer.replace('"', "'"))
.build()
)
.build()
)
client.im.v1.message.create(req)
except Exception as e:
lark.logger.error(f"处理消息出错: {e}")

# —- 注册事件处理器并以长连接启动 —-
handler = (
lark.EventDispatcherHandler.builder("", "")
.register_p2_im_message_receive_v1(on_message)
.build()
)

client = lark.ws.Client(
APP_ID,
APP_SECRET,
event_handler=handler,
log_level=lark.LogLevel.DEBUG,
)

if __name__ == "__main__":
# 长连接方式启动：本地进程主动连飞书，无需公网回调URL
client.start()

说明：lark-oapi 不同版本的个别类名/方法名可能有微调，跑之前先 pip show lark-oapi 看下版本，对照官方 echo_bot 示例 核对。骨架逻辑（注册 register_p2_im_message_receive_v1 → lark.ws.Client(…).start()）是官方稳定模式。

3.4 接上真大模型

把上面 call_llm 换成真实的 OpenAI 兼容调用即可，比如接 DeepSeek 或本地 vLLM：

import requests

def call_llm(prompt: str) -> str:
r = requests.post(
"http://localhost:8000/v1/chat/completions", # 本地 vLLM
json={"model": "mycode", "messages": [{"role": "user", "content": prompt}]},
timeout=30,
)
return r.json()["choices"][0]["message"]["content"]

这样你的飞书群就有了一个"@它就用本地大模型回答"的 AI 助手。

四、部署、排坑与安全

坑1：机器人收不到消息？ 先检查事件订阅有没有加 im.message.receive_v1、权限有没有审批通过、应用有没有发布版本——这三步漏一步都收不到。

坑2：消息重复处理？ 飞书可能重推事件。Lark Channel SDK 已经内置了去重和过期消息过滤（S3），但自己业务层最好用 message_id 再去一次重。

坑3：webhook 泄露。 webhook 地址就是发消息的钥匙，别提交进 git。建议用环境变量注入，并开启"自定义关键词"安全设置做兜底。

坑4：长连接进程退出。 生产环境用 systemd/supervisor 托管，挂了自动重启，毕竟 WebSocket 偶尔会断。

五、小结

• 群自定义机器人（webhook）= 单向推送器，5 分钟搞定，适合告警/通知；

• 企业自建应用 + lark-oapi 长连接 = 双向对话机器人，2026 年免公网域名，适合接 LLM 做群聊 AI 助手；

• 两条路线不是二选一，通常一个团队两者都用：webhook 推告警，应用机器人做对话。

这套东西的真正价值不在"调通接口"，而在于：把大模型能力送进你每天都在用的 IM 里，而不是让同事再开一个新工具。

参考资料