飞书机器人接入指南：从事件订阅到消息回复的完整闭环

这篇是飞书接入系列的第 2 篇。上一篇我们把 Demo 的主流程跑通了能够解析客户问题、命中查询规则、返回结果。但当时的结果只在本地终端里打印客户看不到演示的时候还得把屏幕投过去非常被动。这次的核心任务就是把这个 Demo 接上飞书机器人让客户直接在飞书里发消息Demo 收到后自动回复查询结果。这也是企业做 POC概念验证时最常见的姿势不需要部署复杂的客服系统一个飞书应用加一段回调代码就能让客户“上手即用”。这篇文章不追求把飞书开放平台的所有接口都讲一遍只围绕一个闭环展开消息怎么进来、事件怎么订阅、消息怎么解析、结果怎么回出去以及最容易被坑的权限、回调超时、消息重试这几个点。适合正在做 AI 应用、企业内部工具、或者给客户做 Demo 演示的开发者参考。看完之后你就能把一个能“对话”的飞书机器人跑起来并且知道后续怎么扩展成表格卡片、文件回复这些更高级的交互。1. 接入前的方案选型与整体架构1.1 为什么用“应用机器人”而不是“自定义机器人”飞书里做机器人大多数人的第一反应是在群里加一个自定义机器人拿 Webhook 地址往群里推消息。这个方案确实快3 分钟就能跑通但有个致命缺陷它只能单向推送不能接收用户消息更不能针对某个人“有来有回”地对话。客户发来的消息自定义机器人根本收不到。要让 Demo 接收客户话题必须走飞书开放平台里的真正“应用机器人”路线。应用机器人本质上是一个企业自建应用的能力延伸它拥有独立的机器人身份可以在单聊、群里被 并通过事件订阅把消息实时推送给我们的后端服务。这就像一个拥有“微信号”的客服客户直接私聊它消息就通过飞书的服务器转到了我们手里。对比下来我建议任何要做双向交互、有状态会话的 Demo 都直接选应用机器人。虽然配置步骤多一些但换来的是完整的消息收发链路、可管理的权限体系、以及后续扩展消息卡片、日程、文档等企业级能力的可能性。这也是飞书生态里正规军的做法。1.2 从“客户发消息”到“Demo 回复”的完整链路先画一下整体链路心里有数再动手会顺很多。客户在飞书里找到我们的机器人发起私聊或群里 它输入一个话题或问题飞书服务器接收到这条消息后通过你配置的事件订阅地址回调 URL把消息内容 POST 到你的后端服务后端拿到消息先校验是否是飞书发的签名/Token 校验再解析出用户说了什么、发送者的 open_id业务逻辑处理完后调用飞书的消息发送接口以机器人身份把查询结果回复给对应的用户。这条链路里后端服务是核心中转站。它既要做消息的接收者又要做消息的发送者。接收依靠事件订阅回调发送依靠开放平台的 API 接口。两者用的是同一个应用身份只是方向相反。理解这一点后续看代码就不会绕。还需要说清楚“谁在跟飞书通信”。飞书服务器和后端服务之间是 HTTPS 调用所以你的后端必须有一个公网可访问的 HTTPS 地址。本地开发阶段可以用内网穿透工具把本地端口暴露成公网 URL或者干脆部署到一台有公网 IP 的测试服务器上。我用的是后者部署在测试服务器上回调地址稳定排查问题也方便。2. 创建飞书应用与权限配置的完整步骤2.1 在开发者后台创建企业自建应用这一步相当于给我们的机器人办一张“身份证”。打开飞书开放平台open.feishu.cn用管理员账号登录进入开发者后台点击“创建企业自建应用”填写应用名称和描述。名称建议直白一点比如“XX 智能查询 Demo”这样客户在飞书里搜索机器人时一眼能认出来。创建完成后进入应用详情页左侧菜单能看到“凭证与基础信息”“权限管理”“事件订阅”“版本管理与发布”等模块。需要注意刚创建的应用默认只有开发者本人可见如果要让客户在真实环境中测试必须创建版本并发布。所以配置完成后最后一步记得去“版本管理与发布”里创建版本并提交审核企业自建应用通常管理员审核通过就立即生效速度很快。关于应用的“可见范围”如果只给内部几个人测试直接设置成指定成员即可没必要全员可见。这一步直接影响后续客户能否在飞书里搜索到你的机器人属于很多新手会漏掉的配置别等回调都配好了才发现找不到机器人。2.2 开通消息收发必备的两个权限权限是飞书应用里最让人头疼的一块。如果一个接口调用失败十有八九是权限没开。我们需要在“权限管理”里搜索并开通以下两项im:message读取用户发给机器人的单聊消息消息事件回调依赖这个权限。im:message:send_as_bot以机器人的身份发送消息也就是回复查询结果的能力。这两个权限一个是“收”一个是“发”缺一不可。开通权限之后需要等待一段时间生效如果是企业自建应用且管理员审核通过通常几分钟内就会生效。实测中如果调用发送接口报错提示权限不足可以在权限管理里确认状态是否为“已开通”以及版本发布里是否包含最新的权限配置。很多开发者在这里会踩同一个坑在权限管理里加好了权限但没有重新发布版本导致运行时报权限错误。记住企业自建应用的权限配置必须在创建新版本并发布后才真正对运行时生效。2.3 配置事件订阅与回调地址进入“事件订阅”页面这里有几个关键字段请求地址Request URL、Verification Token、Encrypt Key。请求地址就是我们后端接收消息事件回调的 HTTPS 地址比如https://your-domain.com/webhook/feishu。Verification Token 和 Encrypt Key 建议都设置上尤其是 Encrypt Key开启后飞书推送过来的请求体会用 AES 加密虽然多一层解密操作但能避免消息明文在传输过程中被截获安全等级高一个档次。设置好请求地址后飞书会发送一条 URL 验证请求。你的后端收到这个请求后需要原样返回请求中的challenge字段如果是加密模式先解密再返回才能通过验证。这一步很关键我在第 4 章会给出完整代码。验证通过后再在“事件”列表里添加im.message.receive_v1事件这是接收消息的核心事件。还有一点要提醒飞书对回调地址的响应时间要求是 3 秒内超时会被判定为失败。所以回调接口里的业务逻辑一定要异步化先秒回{code:0}告诉飞书“我收到了”再在后台慢慢处理查询逻辑。这个问题在真实流量下会特别明显压测时如果处理耗时过长飞书会反复重推消息造成重复回复。3. 后端服务的基础框架搭建3.1 选择 FastAPI 作为回调服务后端框架我选择了 FastAPI纯 Python、自带异步支持、依赖注入简单、调试方便最关键是它有完善的请求体验证机制。当然你用 Flask、Spring Boot、Node.js 也完全可以核心不是框架而是处理回调的逻辑。我选 FastAPI 还有个原因配合 uvicorn 部署特别轻量一两行命令就能启动一个带自动文档的服务开发阶段可以直接在浏览器里测试回调接口。项目结构我建议保持简单毕竟这是一个 Demo 而不是完整业务系统。一个main.py放路由和消息处理一个config.py存放应用凭证一个business.py放查询逻辑。这样即使后续代码变多也能保持清晰。依赖安装很简单pip install fastapi uvicorn requests三件套就够了。如果你用了官方 SDK可以加lark-oapi但我的建议是先不用 SDK 跑通一次理解协议本身再在后续重构中引入 SDK 简化代码。这样即使哪天 SDK 出了兼容性问题你也能手写请求兜底。3.2 应用凭证的存储与安全处理开放平台里的 App ID、App Secret、Verification Token、Encrypt Key 这四个凭证是机器人的全部身份凭证。App ID 是应用的唯一标识App Secret 是调用接口时的密钥这两个用于获取tenant_access_tokenVerification Token 用于校验回调请求的合法性Encrypt Key 用于加解密回调请求体。我的做法是把它们放在环境变量里而不是硬编码在代码中。Demo 阶段你可以写在config.py里但一定记得不要把带密钥的文件提交到公开仓库尤其是给客户演示时源代码里的密钥泄露会造成直接的经济损失。一个更稳妥的做法是用.env文件管理这些变量并在.gitignore中排除它。这在小团队和 Demo 场景下够用且不折腾。4. 核心代码实现接收消息并回复查询结果4.1 回调接口验证与消息接收下面这段代码是我们整个接入过程的第一个核心组件它负责接收飞书推送过来的事件并响应挑战信息。我把逻辑写在/webhook/feishu路径上用 POST 接收请求。from fastapi import FastAPI, Request import json app FastAPI() # 这里替换成你自己的 Verification Token VERIFICATION_TOKEN your_verification_token app.post(/webhook/feishu) async def feishu_webhook(request: Request): body await request.json() # 处理 URL 验证请求 if body.get(type) url_verification: return {challenge: body.get(challenge)} # 后续消息处理逻辑 handle_message(body) return {code: 0}这个接口的逻辑很直白先判断事件类型如果是飞书发起的 URL 验证请求就把请求里的challenge原样返回如果是消息事件则继续交给handle_message处理。注意我在这里没有做加解密如果你在飞书后台开启了 Encrypt Key请求体里的内容会是加密字符串需要先用 AES 解密才能看到业务数据。为了把基础链路讲清楚我先把加密部分放在后续小节单独说。真实生产环境里还需要校验请求头里的X-Lark-Request-Timestamp和X-Lark-Request-Nonce以及签名。但 Demo 阶段有 Verification Token 校验基本够用且飞书的回调本身是 HTTPS风险可控。如果需要更严格的安全校验可以在后续迭代中补齐。4.2 消息解析提取用户说的话handle_message是消息处理的核心入口。飞书推送的消息事件 JSON 结构比较固定最外层是schema、header、event三个字段。其中header.event_type表示事件类型event.message里是消息内容event.sender里是发送者信息。我们只需要从event.message.content中解析出文本内容以及从event.sender.sender_id中拿到open_id就能完成一次“谁说了什么”的抽取。def handle_message(body): if body.get(header, {}).get(event_type) ! im.message.receive_v1: return event body.get(event, {}) message event.get(message, {}) message_id message.get(message_id) sender event.get(sender, {}) sender_id sender.get(sender_id, {}) open_id sender_id.get(open_id) # 消息内容是 JSON 字符串需要反序列化 content json.loads(message.get(content)) text content.get(text, ).strip() print(f收到消息: {text}, 来自: {open_id})注意这里的content字段是字符串类型里面是一段 JSON比如{text:帮我查一下订单状态}。如果你直接用message[content][text]会报错因为content不是字典。这个细节在官方文档里写得很清楚但初看示例代码时很容易被误导因为示例里经常直接展示解析后的效果。我在调试时就在这里卡了十分钟所以单独拎出来提醒一下。一个值得留心的点消息事件里带message_id这是每条消息的唯一标识也是我们后续做消息去重的关键。飞书在回调响应超时或失败时会重推同一事件如果不能按message_id去重用户可能会收到多条重复回复。我建议在内存里维护一个最近处理过的message_id集合简单场景下直接判断消息是否已处理即可。4.3 调用发送接口以机器人身份回复拿到open_id和文本内容后接下来就是把文本传给 Demo 的查询模块得到结果再调用飞书的发送消息接口把结果回复给用户。这里的发送接口是POST /open-apis/im/v1/messages需要在请求头里带上tenant_access_token并在 body 中指定接收者、消息类型和内容。首先我们需要获取 tenant_access_token。这个 token 是应用访问飞书开放 API 的凭证用 App ID 和 App Secret 换取有效期通常为 2 小时。为了提高效率我在这里加了简单的缓存避免每次发消息都重新获取。下面这段代码展示了获取 token 和发送消息的过程 python import requests import time APP_ID your_app_id APP_SECRET your_app_secret _token_cache {} def get_tenant_access_token(): if _token_cache.get(token) and _token_cache.get(expire_at) time.time(): return _token_cache[token] resp requests.post( https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal, json{app_id: APP_ID, app_secret: APP_SECRET} ) data resp.json() token data.get(tenant_access_token) _token_cache[token] token _token_cache[expire_at] time.time() data.get(expire, 7200) - 60 return token def send_reply(open_id, text): token get_tenant_access_token() resp requests.post( https://open.feishu.cn/open-apis/im/v1/messages, params{receive_id_type: open_id}, headers{ Authorization: fBearer {token}, Content-Type: application/json }, json{ receive_id: open_id, msg_type: text, content: json.dumps({text: text}) } ) result resp.json() if result.get(code) ! 0: print(f发送失败: {result}) return result这里有两个小细节值得说明。第一receive_id_type选择了open_id这是标识一个用户在某个应用下唯一身份的 ID。你也可以选择user_id或email等类型但事件回调里给你的是open_id直接用最省事。第二发送消息时content字段又是一个 JSON 字符串所以需要json.dumps序列化一下再放入请求体和接收消息时正好反过来。这两个“双重重叠的 JSON”容易让人犯迷糊记不住的话就牢记一条经验飞书 API 里很多字段是字符串里面包裹 JSON传参时用json.dumps取数据时用json.loads。设置回调地址时飞书会 GET 请求你的地址带上从“加密策略”生成的challenge参数。如果你的逻辑里没有处理这个 GET 分支验证会一直失败。所以建议在路由里同时处理 GET 和 POST 两种请求。4.4 整体流程串起来到这一步我们已经有了回调接口、消息解析、发送回复三个组件把它们串成完整的流程app.post(/webhook/feishu) async def feishu_webhook(request: Request): body await request.json() if body.get(type) url_verification: return {challenge: body.get(challenge)} handle_message(body) return {code: 0} def handle_message(body): ... text parse_message_text(body) open_id parse_open_id(body) if not text: return # 调用 Demo 的查询逻辑拿到结果 result query_demo(text) send_reply(open_id, result)把query_demo换成你自己的业务查询逻辑比如查订单状态、查库存、查知识库等整个飞书接入工作就算完成了。这段链路总结下来就是飞书推送消息给你你解析完回一个“已收到”然后在后台做查询再主动调用发送 API 把结果发回去。5. 进阶用消息卡片返回表格类查询结果5.1 飞书消息卡片的基本结构在实际业务演示中纯文本回复往往不够直观。客户问“这个月各区域的销售额是多少”你回一段文字对方还得一行行看体验一般。飞书的消息卡片interactive可以解决这个问题它支持标题、字段、分割线、备注等多种模块甚至能直接渲染一个小型表格视觉上干净很多。卡片本质上是一段 JSON通过msg_type: interactive发送。1.0 版卡片结构是{config: {}, header: {}, elements: []}新版 2.0 卡片则是{schema: 2.0, body: {elements: []}}两种格式飞书都兼容但我建议直接用 2.0 版也是官方现在主推的模式。下面是一个最简单的卡片示例{ schema: 2.0, body: { elements: [ { tag: markdown, content: **查询结果**\n订单号A12345\n状态已发货 }, { tag: hr }, { tag: note, elements: [ {tag: plain_text, content: 数据更新于 2025-01-07 10:00} ] } ] } }把这段 JSON 作为content的值发给用户就会渲染出一张带标题、分隔线和备注的小卡片。对于 3 到 5 行内的查询结果用markdown模块就足够了比纯文本美观得多也不需要额外学习复杂组件。5.2 用 column_set 组件渲染真正的表格当结果超过 5 行比如客户查一份订单列表每行有订单号、客户名、金额、状态四列再用 markdown 拼接就会显得很乱。飞书 2.0 卡片提供了column_set组件本质上是多列布局我们可以用它来模拟表头和数据行对齐效果好很多。下面是一个两列三行的表格卡片示例{ schema: 2.0, body: { elements: [ { tag: column_set, flex_mode: none, background_style: grey, columns: [ {tag: column, width: weighted, weight: 1, elements: [{tag: markdown, content: **订单号**}]}, {tag: column, width: weighted, weight: 1, elements: [{tag: markdown, content: **状态**}]} ] }, { tag: column_set, flex_mode: none, columns: [ {tag: column, width: weighted, weight: 1, elements: [{tag: plain_text, content: A12345}]}, {tag: column, width: weighted, weight: 1, elements: [{tag: plain_text, content: 已发货}]} ] } ] } }在实际代码里你只需要在后端写一个函数接收查询结果的二维数组循环生成column_set数组再塞进卡片的elements里即可。这样面对几十行数据也能整洁展示。但要注意飞书卡片对单次内容的长度有限制如果结果特别长建议通过分页按钮或者让用户输入更具体的筛选条件来缩小范围。5.3 适合用卡片返回的典型查询场景我实际在 Demo 中经常用卡片返回的结果有两类。第一类是状态查询比如“工单进度”“订单状态”“审批流程”每条记录 3 到 5 个字段用列布局展示非常直观客户扫一眼就能看到关键信息。第二类是聚合统计比如“本周销售额按渠道分布”可以把渠道名作为一列金额作为另一列再配合卡片的颜色背景做出层级感。如果你的查询结果本身来自飞书多维表格还可以用多维表格的开放 API 把记录拉出来再渲染成卡片。这就形成了一个很实用的闭环客户在聊天里问“这周有哪些未处理的工单”机器人调用多维表格 API 查出记录回填卡片客户点击卡片里的链接直接打开多维表格处理。这个方案在团队内部工具中非常受欢迎也是飞书生态里“机器人 多维表格”的经典用法。6. 常见问题排查与避坑实录6.1 回调验证失败challenge 没有正确返回这是接入过程中遇到最多的报错通常表现为在飞书后台点击“保存”时提示“请求地址验证失败”。原因基本逃不过两个一是你的服务器没有正确响应 GET 请求比如路由只写了 POST 方法或者响应时没有把challenge字段带回来二是你开启了 Encrypt Key但没有对请求体先解密就直接返回导致飞书期望的解密内容没对上。排查思路分几步先用 curl 模拟飞书的验证请求看看返回结果是否符合预期。如果开启了加密模式先确认后端有没有正确实现解密逻辑。一个快速验证的办法是在飞书后台暂时关闭 Encrypt Key只保留 Verification Token看验证是否能通过。通过后再把加密开回来定位问题边界。6.2 消息能收到但不回复权限与 token 的坑有时候回调地址验证通过了消息事件也推送到后端了日志里也能看到收到的文本但用户那边就是收不到回复。这时先看发送接口的返回值。如果返回code: 9499或类似权限不足的错误码很可能是im:message:send_as_bot权限没生效。排查方法是回到权限管理页面确认状态然后重新创建版本并发布。另一种情况是tenant_access_token获取失败比如 App Secret 填错、或者应用被冻结。这类问题通常会在调用 token 接口的响应中直接体现把响应体的msg字段打出来看一眼就能定位。还有一种不太明显的坑发送消息时不带receive_id_type参数或者传了但你用的是user_id飞书会报参数错误。事件回调里给你的open_id和发送接口的receive_id必须匹配同一个类型体系。6.3 消息重复回复理解飞书的重试机制飞书的回调协议设计了对接收方的可靠性要求。如果接收方没有在 3 秒内响应 2xx飞书会认为推送失败并按照一定策略重新推送同一事件直到成功或达到最大重试次数。这就导致一个常见现象你的后端处理逻辑慢于 3 秒飞书重推你又处理了一遍用户就收到了两条相同回复。解决思路有两个一是在业务处理链路前加一个内存去重表按message_id判断是否已经处理过已经处理的直接返回成功二是把耗时操作放到异步队列里回调接口只负责确认收到拒绝在请求内同步执行长时间查询。我的建议是两者都做先按message_id去重再把查询任务丢进后台线程或消息队列。这两招加上消息重复的问题基本不会再出现。6.4 找不到机器人版本发布与可见范围配置全部完成代码也部署好了但在飞书里搜不到机器人这个问题太常见了。原因基本都在应用配置侧要么应用没有创建版本并发布要么可见范围没有包含你测试用的账号。注意企业自建应用在“版本管理与发布”里创建版本并提交后需要管理员审核通过才算是真正上线。开发模式下只有应用创建者和被添加为协作者的人才能看到。另外机器人私聊还有一个前置条件应用必须开启至少一个消息相关的权限并且用户需要在飞书里先从“通讯录”或“工作台”入口添加这个机器人应用。这个入口位置在不同版本的飞书客户端略有差异通常是在工作台里点击应用图标进入然后选择“添加应用”或直接开始聊天。6.5 长连接模式不需要公网回调的另一种方案如果你是在本地开发没有公网地址又不想用隧道工具可以了解一下飞书的长连接模式。飞书开放平台支持 WebSocket 长连接方式接收事件即你的本地服务主动和飞书服务器建立一个长连接通道事件推送通过这个通道实时到达完全不需要配置公网回调 URL。官方称这种方式为“长连接模式”适合本机调试、内网部署等场景。使用长连接模式不需要在事件订阅里配置请求地址只需在代码里用飞书官方 SDK 启动长连接客户端并注册事件处理器。SDK 会自动维护连接和重连。这个方案对 Demo 阶段尤其友好不用折腾公网地址和 HTTPS 证书直接在本地就能调试整个收发链路。不过它也有短板长连接模式不支持手动指定回调地址对于某些需要自定义事件分发规则的企业场景可能不够灵活。但如果你只是开发测试它绝对是效率最高的方案。写在最后的几个实操建议第一个建议不要把回调接口里的逻辑写得太重。我之前图省事直接在回调里同步调用了 Demo 的查询逻辑结果查询模块偶尔需要几秒中时间飞书一重试就重复回复。后来改成回调收到就立即返回查询放到后台线程执行消息用message_id去重整个链路就稳定下来了。第二个建议给发送消息的逻辑加日志和返回值校验。演示给客户看的时候最怕的就是“用户发了消息机器人没反应”而你在现场无法快速定位。把每次收到的消息、解析后的文本、查询结果、发送接口的返回码都打出来一旦出问题当场就能看出是解析问题、权限问题还是 Demo 逻辑问题整个演示过程会从容很多。第三个建议接入飞书只是第一步后续可以往两个方向扩展。一是做会话上下文让 Demo 能承接多轮对话而不是每条消息独立处理二是把回复内容从纯文本升级成卡片、图片、文件乃至音视频让查询结果的呈现方式更专业。这块我在后续的文章中再逐一展开。这篇先把“能对话、能回复”这个基础闭环打好后续的任何高级功能都建立在这套消息收发机制之上。