这篇是飞书接入系列的第 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_token；Verification 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": f"Bearer {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 能承接多轮对话，而不是每条消息独立处理；二是把回复内容从纯文本升级成卡片、图片、文件乃至音视频，让查询结果的呈现方式更专业。这块我在后续的文章中再逐一展开。这篇先把“能对话、能回复”这个基础闭环打好，后续的任何高级功能都建立在这套消息收发机制之上。