☰
飞书机器人接入实战:从事件订阅到消息回复的完整链路
2026/9/26 14:41:04 网站建设 项目流程

这次要做的接入其实很典型:客户那边希望有一个“发消息就能拿到结果”的入口,不想装重客户端,也不愿意为一个小场景去开一套完整的前后端页面。我打算用飞书机器人把这个事情跑通——把现有的查询能力挂到飞书上,客户在聊天窗口里发一句“查一下订单状态”,机器人就把查询结果回出来。我的 Demo 重点不在算法,也不在业务复杂度,而在把“接收客户话题 → 解析查询意图 → 调用内部查询 → 把结果回给客户”这条链路完整闭环。

先把结论放前面:飞书机器人这套东西,官方叫“企业自建应用”,本质上就是一个能收发消息的机器人账号加上一组开放接口。对 Demo 来说,你不需要把全部系统搬上去,只需要解决两个问题——如何收到客户在聊天框里说的话,以及如何用机器人身份把结果发回去。这篇文章会把这两条链路拆开讲,包含后台配置、权限申请、代码骨架和常见坑,适合正在做类似 Demo 或者想快速验证“IM 客服式查询”场景的人参考。

1. 接入前的整体设计与选型思路

1.1 为什么最终选定飞书机器人

做客户沟通入口,摆在我面前的选择有不少:自建网页聊天窗、邮件自动回复、企业微信群机器人、飞书机器人。我最后选了飞书机器人,原因比较实际。

第一是免登录。客户只要在飞书里,直接跟机器人对话就行,不需要注册账号、不需要记住域名、不需要二次验证。这个体验对 Demo 阶段特别重要,因为你的目标不是做用户体系,而是验证“客户话题进来后能不能正确查询并回复”。把账号体系引进来,问题复杂度就上去了。

第二是飞书开放平台的机器人接口足够成熟。它有事件订阅机制能实时推送消息,也有主动发送消息的 API,可以做到“你一句我一句”的对话体验。人工客服系统要做的“转人工”之类操作,在消息卡片里也能做,后续扩展空间很大。

第三是我做 Demo 时希望尽可能少写前端代码。飞书端天然提供了输入框、消息气泡、卡片渲染,我只需要处理文本和 JSON,不用管浏览器兼容、移动端适配、消息推送 SDK 这些东西。如果自建网页聊天窗,还要考虑 WebSocket 维护、消息持久化、前端轮询,开发量翻一倍。

当然,飞书机器人也有它的限制。比如应用必须经过租户管理员或开发者的审核配置,事件订阅回调需要公网可访问的地址,发送消息受频率限制。这些限制在 Demo 阶段都可以接受,但我在选型时也把它们记在了评分表里。

1.2 消息链路方案对比:事件订阅比轮询更合适

接入飞书之前,我梳理过三种常见方案。很多人第一步会想:我是不是可以定时去拉取客户发给机器人的消息?这种“轮询”思路在飞书体系里不太好实现。

飞书开放平台没有提供“拉取某个机器人所有未读消息”的通用接口。它的消息记录查询接口一般是围绕具体会话或者具体消息 ID 来做的,你没法像查数据库一样直接扫一遍收件箱。这就要求开发思路转成“事件驱动”——让飞书在有新消息时主动推给我的服务。

所以最终方案是:飞书开放平台配置事件订阅,收到im.message.receive_v2事件后,我的服务端解析事件内容,再调用飞书发送消息接口把查询结果回过去。整个链路是推拉结合的,接收靠推送,发送靠 API。

这里做一个对比表格,方便直观理解:

方案实时性开发量适用场景
自建聊天页面高,但要维护连接高大型产品、需要深度定制 UI
邮件自动回复低中异步客服、工单系统
飞书机器人事件订阅高低快速验证 IM 客服、查询机器人

对 Demo 而言,事件订阅 + 主动回复的组合,开发量最小,链路最短,也最接近真实产品形态。后面生产化也只需要在架构里加消息队列和任务调度,不用推翻重来。

1.3 Demo 数据流设计与模块边界

我把整个接入拆成了四个相对独立的模块,方便单独调试。

客户发来话题后,飞书推送事件到我的回调服务。回调服务先做验签和去重,然后把消息内容丢给话题解析模块。话题解析模块负责从文本里找出客户想问什么、有没有附带参数。比如“订单号 SO-12345 到哪了”和“查一下物流”,背后的查询意图是不同的。解析完成后,查询服务根据意图去查数据源,这个数据源可以是写死的 JSON、本地 SQLite、内部接口,Demo 阶段我建议先用一个 mock 服务顶着,等链路通了你再换成真实数据。最后回复组装模块拿到查询结果,选择纯文本或者消息卡片,通过 API 回给客户。

这套设计的好处是每个模块都能独立测试。回调服务可以先打印日志,不接业务;解析模块可以先不接飞书,直接用命令行喂文本;查询服务可以做成 HTTP 接口,用 curl 验证。如果一上来就把全部逻辑写在一个回调函数里,出了问题你根本不知道是哪一环挂了。

我在实际操作中体会最深的一点是:不要在事件回调里写重逻辑。因为飞书对回调响应时间有要求,事件收到后你必须在几秒内返回成功应答,否则飞书会认为发送失败并重试。Demo 虽然数据量小,但养成这个边界意识没坏处。

2. 开发者后台配置与权限准备

2.1 创建自建应用与获取凭证

飞书接入的第一步,是在飞书开放平台创建一个企业自建应用。打开开发者后台,进入“开发者后台 → 创建企业自建应用”,填一个应用名称和图标,名称我建议直接写成机器人将要展示给客户的名字,这样后面调试的时候不容易迷惑。

创建完成之后,最重要的一件事是拿到两个凭证:App ID 和 App Secret。App ID 相当于应用的身份证号,App Secret 相当于密码,调用大部分 API 都要用这两个东西换取租户访问令牌(tenant_access_token)。

这里有个安全问题要特别注意:App Secret 一定不要写死在前端代码或者公开仓库里,哪怕 Demo 也不行。我自己吃过亏,曾经把 App Secret 放在代码仓库的配置文件里,后来整改的时候要批量换密钥,非常麻烦。正确做法是放在环境变量或者本地配置文件中,并且加入.gitignore。

在后台的“凭证与基础信息”页面,还能看到应用是否启用、版本号、可见范围等信息。如果后续发现接口提示没有权限,第一件事就是回这个页面检查应用是否已发布。测试阶段的“创建版本并发布”操作很多人会漏掉,只停留在后台编辑状态,这样机器人永远无法真正收到消息。

2.2 事件订阅与回调配置

应用创建好后,进入“事件与回调”页面,添加事件监听。Demo 只需要一个事件:接收消息。在飞书的事件列表里,它的标识是im.message.receive_v2,涵盖单聊和群聊中发给机器人的消息。如果你的客户场景主要是单聊,订阅im.message.p2p_msg_receive前缀的事件也可以;但为了链路简单,我直接订阅 v2 这个通用事件,在代码里再用chat_type字段区分单聊还是群聊。

接下来设置回调地址。这里要求提供一个公网 HTTPS 地址。本机开发时可以用内网穿透工具把本地服务映射成一个临时公网地址,或者把服务部署到一台带公网 IP 的测试服务器上,两种方式我都试过,后者更稳定,因为穿透工具的免费域名经常变,回调地址一改就要在后台重新配置。

配置回调地址时,飞书会立刻发一个验证请求,这是一个 POST 请求,请求体长这样的:

{ "challenge": "xxxxxx", "token": "xxxxxx", "type": "url_verification" }

你的服务端收到这个请求后,必须原样返回请求体中的challenge字段。比如用 FastAPI 写,大概是这样:

from fastapi import FastAPI, Request from fastapi.responses import PlainTextResponse app = FastAPI() @app.post("/webhook/feishu") async def feishu_callback(request: Request): body = await request.json() if body.get("type") == "url_verification": return PlainTextResponse(body.get("challenge", "")) return {"ok": True}

注意,这里返回的一定要是纯文本的 challenge 值,不能包装成 JSON,也不能加其他字段,否则飞书后台会提示验证失败。

另外,后台还有两个可选配置项:Verification Token 和 Encrypt Key。Verification Token 是一个简单的校验字符串,Encrypt Key 则用于对回调内容做 AES 加密。Demo 阶段我建议先不开启 Encrypt Key,直接走明文事件,把链路跑通后再加密。一旦开启加密,你不仅要处理解密逻辑,还要注意解密后的 JSON 解析,排查问题会多一层。

2.3 给机器人申请消息权限

事件订阅解决了“收消息”,发送消息还需要另一组权限。在开发者后台的“权限管理”页面,需要申请和消息相关的权限。常见的有读取单聊消息、读取群聊消息、以机器人身份发送消息等。每个租户的权限标识可能略有差异,而且飞书后台更新过权限体系,所以我的经验是先在“权限管理”里搜“消息”“机器人”关键词,把候选权限都申请上。

权限申请之后,还有一个关键步骤:发布版本。飞书的权限生效机制是“申请权限 → 创建版本 → 发布版本”,发布后应用的状态才会更新,机器人也才会真正出现在客户的组织架构里。很多人在这里卡住:后台明明都配置好了,但自己给机器人发消息没反应,十有八九是应用没有发布版本。

我在实际测试中还发现,飞书对“机器人与用户会话”有两种模式。一种是我作为开发者直接跟机器人对话,这要求机器人已经启用;另一种是让其他客户找到这个机器人并开启会话。应用发布后,最好先用管理员身份找到机器人并主动发一条消息,确认事件能推送到回调服务,然后再进入代码联调。

3. 核心代码实现与回复链路

3.1 回调服务骨架与事件解析

我用的技术栈是 Python + FastAPI,部署在一台测试服务器上。原因不用多说,Python 做文本处理和 API 调用快,FastAPI 自带的异步能力也够用。回调服务的核心代码不复杂,重大戏在事件类型判断和消息内容解析上。

事件订阅验证通过后,飞书推送的消息事件长这个大致结构:

{ "schema": "2.0", "header": { "event_id": "xxx", "event_type": "im.message.receive_v2", "tenant_key": "xxx" }, "event": { "sender": { "sender_id": { "open_id": "ou_xxx" } }, "message": { "message_id": "om_xxx", "chat_id": "oc_xxx", "chat_type": "p2p", "content": "{\"text\":\"查一下订单\"}", "message_type": "text" } } }

需要澄清一点:content字段是一个 JSON 字符串,不是直接能用的文本。我第一次写的时候直接读content["text"],结果拿到的是字符串对象而不是字典,定位半天才发现。正确姿势是:

import json message = event["message"] content_type = message.get("message_type", "text") content_obj = json.loads(message.get("content", "{}")) if content_type == "text": user_text = content_obj.get("text", "")

如果客户发的是图片、文件、语音这类富媒体消息,content的结构会不一样。Demo 阶段我直接忽略非文本消息,统一回一句“目前只支持文字查询”,省去一堆不同消息类型的解析工作。

3.2 客户话题解析与查询意图匹配

客户发来的话是自然语言,不会按标准格式走。这个 Demo 里我没有上大模型做意图识别,而是用关键词匹配 + 正则抽取参数,因为业务场景固定,规则解法最快也最可解释。

我定义了一个简单话题模型,把查询拆成“意图 + 参数”两个维度。比如客户说“查一下订单 HELLO-829 到哪里了”,意图是“物流查询”,参数是HELLO-829;客户说“报价单有没有更新”,意图是“报价查询”,参数为空。实现上用一张关键词映射表:

intent_rules = { "order_query": ["订单", "订单号", "查订单"], "logistics_query": ["物流", "快递", "到哪", "到哪里"], "price_query": ["报价", "价格", "报价单"] } param_patterns = { "order_query": r"(?:订单号|单号)[::\s]*([A-Za-z0-9-]{3,})", "logistics_query": r"([A-Za-z0-9-]{3,})" }

流程是:先遍历intent_rules,统计命中的关键词权重,得分最高的作为意图;再用对应的正则表达式抽取参数。抽不到参数时,我会让机器人反问一句“请提供订单号或物流单号”,而不是直接查一个空条件。

用这种方式的好处是,规则你可以写进代码配置里,哪天客户新增了一种查询,加两行关键词就行,不用改主流程。我见过不少团队在 Demo 阶段就直接接大模型,看起来灵活,但排查问题时“为什么这个说法没识别出来”会变成一个黑盒,反而耽误时间。规则版本先跑通,后续需要再平滑升级。

3.3 把查询结果回复给客户

查询服务拿到意图和参数后,调一个 mock 接口。真正项目里这里可能是查数据库、查订单系统、查 ERP,但接口形态是一样的——给它参数,返回结构化 JSON。我的 mock 返回长这样:

{ "code": 0, "data": { "order_id": "HELLO-829", "status": "已发货", "logistics_company": "顺丰", "tracking_no": "SF123456789" } }

拿到结果后,回复方式有两种选择。第一种是纯文本,把结果拼成一行文字发回去。优点是简单,缺点是字段一多就看不清。第二种是消息卡片,飞书原生支持交互式卡片,展示结构化结果非常好看,甚至还能加按钮、加跳转链接。

热词里很多人问“飞书机器人发送表格”,其实在消息卡片里就能实现。比如把查询结果渲染成字段列表或者表格,用交互卡片消息类型发送。我给 Demo 做了一个简单版本,用卡片展示订单信息:

send_msg_body = { "receive_id": open_id, "msg_type": "interactive", "content": json.dumps({ "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": "订单查询结果"} }, "elements": [ { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "kdl", "content": "**订单状态**\n已发货"}}, {"is_short": True, "text": {"tag": "kdl", "content": "**物流公司**\n顺丰"}} ] }, { "tag": "hr" }, { "tag": "note", "elements": [{"tag": "plain_text", "content": "HELLO-829 正在运输途中"}] } ] }) }

这里的send_msg_body会通过飞书发送消息接口,POST 到/open-apis/im/v1/messages,鉴权用tenant_access_token。要注意的是content字段必须是 JSON 字符串,不能直接放 Python 字典,否则飞书接口会报格式错误。我第一次就栽在这里,以为是签名问题,后来发现是请求体序列化没做好。

3.4 异步处理、去重与日志

回调服务收到飞书事件后,我强烈建议先应答再处理。前面提到飞书对回调有超时要求和重试机制,如果你的回调函数里直接执行查询 + 发送消息,整个过程超过 2 秒就可能超时。飞书重试事件时如果没做去重,客户会收到重复回复。

我用的方案是“立即应答 + 后台任务处理”。FastAPI 里有BackgroundTasks,也可以在回调里把事件丢到一个内存队列,由 worker 线程去消费。Demo 阶段,用BackgroundTasks足够,代码也简单:

from fastapi import BackgroundTasks def handle_message(event: dict): # 解析、查询、回复都在这里执行 pass @app.post("/webhook/feishu") async def feishu_callback(request: Request, background_tasks: BackgroundTasks): body = await request.json() if body.get("type") == "url_verification": return PlainTextResponse(body.get("challenge", "")) background_tasks.add_task(handle_message, body) return {"ok": True}

先把{"ok": true}立刻返回给飞书,再在后台慢慢处理业务。既不会因为飞书重试导致重复,也腾出了处理时间。

去重我是这么做的:飞书每个事件都有event_id,我用 Redis 存最近 5 分钟的 event_id,处理前先检查是否已存在,存在就直接跳过。没有 Redis 的话,用内存里的set加过期时间也可以,就是分布式部署时不共享。日志方面,我建议在收到事件、解析出意图、查完数据、发送消息这四个节点各打一条日志,带上 message_id,排查问题直接按消息 ID 串日志。

4. 常见问题与排查技巧实录

4.1 回调验证不通过,反复报“请求验证失败”

这个坑几乎人人都会踩一次。后台配置回调地址时,飞书发来的验证请求要求响应体是干净的原文challenge。如果你用的是 FastAPI 等框架,注意不要写return {"challenge": body["challenge"]},这样返回的是 JSON,飞书不认识。另外还要检查是否在路由函数上加了额外的响应模型或者日志中间件,中间件可能会把响应包一层,导致响应体多了字段。

如果回调地址是 HTTPS 而且是自签名证书,飞书也可能验证失败。测试阶段直接用正规证书,不管是云服务器自带的还是从证书服务商申请的,都不要用自签名证书。我遇到过在本地能用 curl 访问、但飞书后台一直验证失败的情况,最后发现是证书链不完整。

4.2 后台事件配置好了,机器人就是收不到客户消息

先说一个最容易被忽略的:应用是否已经发布。飞书应用在开发者后台编辑状态下,机器人是不对外生效的。哪怕你自己是管理员,也得“创建版本 → 发布”,应用状态变成“已发布可用”之后,机器人才能正常收发。

第二个检查点是权限。你虽然订阅了im.message.receive_v2事件,但如果没有申请对应的消息读取权限,飞书可能不会推送事件。去权限管理页面确认“读取用户发给机器人的单聊消息”这类权限已经申请并随版本发布。

还有一个我实际踩过的点:客户对机器人发消息之前,可能需要先在飞书里搜到并主动发起会话。如果应用没有设置“可用范围”,或者可用范围里没有包含测试账号,你是搜不到那个机器人的。测试阶段把“可用范围”设成全公司或者明确包含测试用户,能省很多折腾。

4.3 发送消息报错:没有权限或“应用未开启”

飞书发送消息接口用的是tenant_access_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} ) token = resp.json().get("tenant_access_token")

拿到 token 后,调发送接口时要在请求头带Authorization: Bearer <token>。我之前犯过一个低级错误,把 token 拼进 URL 里当 query 参数,服务端一直返回鉴权失败。

如果 token 没问题,错误信息却提示“没有权限”,大概率是权限没生效。飞书的权限修改不是即时生效的,发布新版本后需要等一会儿。重新发布版本时,要仔细看发布页面里有没有提醒新增的权限项。

4.4 消息卡片或表格样式没有按预期展示

卡片不展示时,先检查msg_type是不是interactive,再看content是不是 JSON 字符串而不是 Python 对象。飞书对卡片 JSON 的格式要求很严格,多一个逗号都会导致整个卡片渲染失败,它不会提示你语法错误,只会默默发不出来。

我测试时发现,卡片里的fields字段要用kdl标签而不是markdown标签。飞书的卡片文本引擎对 Markdown 语法的支持是有限制的,像加粗这类语法,在markdown标签下支持,但在plain_text标签下就是普通字符串。不要想当然地把 Markdown 语法用在全文本标签里。

4.5 从 Demo 走向真实使用前要处理的事

Demo 链路跑通后,后面还有几道坎要过。第一个是消息频率限制。飞书开放平台对单个应用发消息有 QPS 限制,如果客户量大,消息发送必须做缓冲和限流。Demo 里直接同步发送没关系,真实场景要设计重试队列。

第二个是回调安全性。开启 Encrypt Key 加密后,所有事件体都变成一坨密文,需要你在本地用 AES 解密。有些团队为了省事一直不开加密,一旦回调地址被别人扫到,消息内容就裸奔了。建议 Demo 验证之后就开启。

第三个是数据和查询服务要解耦。不要在回调代码里直接拼 SQL、查库,最好把查询封装成独立服务或者独立函数。真实项目里,订单、物流、报价这些可能来自不同系统,回调服务只做消息转发和结果转发,边界清晰才不会越写越乱。

这套链路我调试了两天左右,大部分时间花在权限申请和回调验证上,真正写业务逻辑的时间很少。踩过几次坑之后,我的体会是:飞书接入这类工作,最值钱的不是代码怎么写,而是对“事件必须立即应答”“权限必须随版本发布”“内容字段必须是 JSON 字符串”这几个规则的敏感度。把这些规则内化成习惯,再做集成类 Demo 就会顺很多。后面如果客户话题更复杂,我打算把规则解析替换成大模型意图识别,但消息链路的骨架不需要动。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询