当前位置: 首页 > news >正文

飞书 Agent 集成(Channel SDK)lark-channel-sdk入门

前言

近期在研究将开发的智能体接入飞书。看了一些视频、博客都是将已有网页应用接入飞书,俗称飞书应用机器人。例如飞书开发智能体博客 - KempY - 博客园

这篇博客入门还是不错的,只是并未涉及具体代码实现和agent部分,对应智能助手:从自然语言到可执行代码 - 服务端 API - 开发文档 - 飞书开放平台 飞书文档中服务端API部分。

当然可以通过手动封装智能体回复,去重逻辑,长连接等实现飞书与agent的集成,不过这样还是不够智能,灵活。

真正要看的文档应该是这个:飞书 Agent 集成能力概述 - 飞书 CLI - 开发文档 - 飞书开放平台 对应飞书文档中飞书CLI部分。

飞书 Agent 集成能力概述部分分为三个版块感兴趣的可以去看看。本篇文章只针对Channel SDK进行介绍。

更多API可以参考larksuite/channel-sdk-python

Channel SDK 做了什么

如果你需要开发 AI 助手、客服机器人、知识库问答、审批/工单助手、群内自动化协作、卡片按钮交互等场景,Channel 能显著减少接入飞书时的重复工程量。

业务目标 未接入 SDK 要做的 接入 SDK 后要做的
建立与飞书的双向通道 申请公网回调地址,或自行实现 websocket 长连、断线重连、心跳检测 createLarkChannel() 一行代码全自动实现
解析飞书事件 im.message.receive_v1 原始 payload:chat_id/sender/message_type/post 嵌套结构,富文本/图片/文件/at 各种分支解析 直接获取 NormalizedMessage / CardActionEvent / CommentEvent 的统一结构化消息和事件
流式输出 card_update API、处理限流、转换 markdown 为飞书卡片 schema channel.stream()里的 ctrl.update()自动刷新界面
机器人回复策略 区分单聊/群聊行为、@机器人策略、群白名单、消息去重、防刷屏策略 设置safetypolicy 配置参数,SDK 内置完成
  • 流式回复:Agent 边推理边输出,用户在飞书里看到的卡片实时刷新进度和内容。
  • 卡片交互:在卡片按钮、表单、下拉,用户点击后回调事件直接送到你的 Agent。
  • 云文档评论:用户在飞书云文档里 @ 你的 bot 提问,Agent 收到评论事件后可以直接回复评论,无需切换到群聊。

第 1 章 · 第一个机器人:Hello World


1.1 认识 FeishuChannel

FeishuChannel 是整个 SDK 的核心入口类,它整合了:

  • 事件接收:WebSocket 或 Webhook
  • 消息归一化:将原始飞书消息转为统一的 InboundMessage 对象
  • 消息发送:各种类型的消息发送与回复
  • 流式输出:CardKit 流式卡片更新
  • 媒体管理:上传与下载
  • 安全策略:安全模式、去重、策略控制

最小化构造如下:

from lark_channel import FeishuChannelchannel = FeishuChannel(app_id=os.environ["LARK_APP_ID"],app_secret=os.environ["LARK_APP_SECRET"],
)

关键构造参数(入门版)

参数 必填 说明
app_id 你的机器人应用 ID,形如 cli_xxx
app_secret 你的机器人应用密钥
domain 开放平台域名。国内版默认 https://open.feishu.cn,Lark 用户需传 https://open.larksuite.com

如果你的机器人服务面向国际版 Lark 用户,记得指定 domain

channel = FeishuChannel(app_id=os.environ["LARK_APP_ID"],app_secret=os.environ["LARK_APP_SECRET"],domain="https://open.larksuite.com",
)

1.2 事件监听机制

FeishuChannel 使用事件驱动的设计。你需要注册一个"当收到消息时该做什么"的处理函数:

async def on_message(msg):# msg 是一个 InboundMessage 对象print(f"收到消息: {msg.content_text}")channel.on("message", on_message)

这里 "message" 是事件名。channel.on(event_name, handler) 注册了一个异步处理函数,当对应事件发生时会被调用。

Events 枚举提供了所有事件名的常量,例如 Events.MESSAGEEvents.CARD_ACTION 等,这是更规范的写法:

from lark_channel import Events
channel.on(Events.MESSAGE, on_message)

1.3 认识 InboundMessage

当收到用户消息时,你的处理函数会收到一个 InboundMessage 对象。以下是入门阶段最常用的字段:

字段 类型 说明
msg.chat_id str 消息所在会话的 ID(回复消息时需要)
msg.content_text str 消息的纯文本内容,已做归一化
msg.message_id str 消息的唯一 ID(用于引用回复)
msg.sender Identity 发送者的身份信息
msg.chat_type str 会话类型:p2p(单聊)、group(群聊) 等

1.4 发送消息:channel.send()

发送消息的基本形式:

await channel.send(chat_id,           # 目标会话 ID{"text": "你好!"}, # 消息内容
)

消息内容可以是多种形式,入门阶段我们先认识三种:

形式 1:纯文本

await channel.send(chat_id, {"text": "这是一条纯文本消息"})

形式 2:Markdown

await channel.send(chat_id, {"markdown": "**加粗**,*斜体*,`代码`"})

形式 3:直接传入 Markdown 字符串

await channel.send(chat_id, "# 标题\n这是正文")

SDK 会自动将裸字符串识别为 Markdown 并转换为飞书富文本格式。

回复指定消息(带引用)

await channel.send(chat_id,{"markdown": "收到你的消息了!"},{"reply_to": msg.message_id},  # 引用这条消息
)

1.5 启动机器人

在 WebSocket 模式下,启动非常简单:

import asyncio
asyncio.run(channel.connect())

connect() 会建立 WebSocket 连接,接收事件,并调用你注册的处理函数。它会阻塞运行,直到你中断程序(Ctrl+C)。


1.6 示例 1:Echo Bot(回音机器人)

这是一个经典的入门示例:机器人将用户发送的内容原样回显。

# chapter01/echo_bot.py
# 最简单的 Echo Bot:收到什么就回复什么import asyncio
import osfrom lark_channel import FeishuChannel# 从环境变量中读取凭据
APP_ID = os.environ.get("LARK_APP_ID")
APP_SECRET = os.environ.get("LARK_APP_SECRET")if not APP_ID or not APP_SECRET:raise ValueError("请先设置 LARK_APP_ID 和 LARK_APP_SECRET 环境变量")# 创建 Channel 实例(默认使用 WebSocket 模式)
channel = FeishuChannel(app_id=APP_ID, app_secret=APP_SECRET)async def on_message(msg):"""收到消息时的处理函数。"""# 获取消息的纯文本内容text = msg.content_text# 打印到控制台,方便调试print(f"[{msg.chat_type}] {msg.sender_id}: {text}")# 回显:在同一会话中回复await channel.send(msg.chat_id,{"text": f"echo: {text}"},)# 注册"收到消息"事件的处理函数
channel.on("message", on_message)# 也可以注册错误处理函数
async def on_error(err):print(f"channel error: {err}")channel.on("error", on_error)# 启动(会阻塞当前线程,直到 Ctrl+C)
print("Echo Bot 已启动,等待消息...")
asyncio.run(channel.connect())

运行方式

# 设置环境变量(Windows PowerShell)
$env:LARK_APP_ID = "cli_xxxxxxxxxxxx"
$env:LARK_APP_SECRET = "your_app_secret"# 或使用 .env 文件
# pip install python-dotenv# 运行
python chapter01/echo_bot.py

在飞书中测试

  1. 在飞书中搜索你的机器人并打开单聊
  2. 发送任意消息,例如:"你好"
  3. 预期回复:echo: 你好
  4. 控制台会打印:[p2p] ou_xxx: 你好
http://www.gsyq.cn/news/1555375.html

相关文章:

  • 2026杭州手表回收避坑指南|正规中检备案门店,杜绝虚高引流、恶意压价套路 - 薛定谔的梨花猫
  • 机器学习生产化:从模型上线到系统韧性工程
  • 设备管理系统团队博客第五篇
  • 南宁品牌首饰回收避坑指南:2026年七大品牌实力排行榜 - 薛定谔的梨花猫
  • 颜值的降维打击!“报名管家”全版本免费,用王炸级UI撑起你的品牌门面 在这个“颜值即正义”的时代,别让丑陋的表单劝退用户 - 亲测好用工具
  • 杭州本地人私藏!2026 全城 6 家靠谱名牌包包回收门店整理 - 开心测评
  • 2026武汉黄金回收避坑干货|拆解虚高报价套路,拒绝到店扣损耗,正规门店全汇总 - 名奢变现站
  • 二手腕表哪些成色最保值?昆明市场真实行情实测 - 开心测评
  • 2026青岛卡地亚首饰回收分级评分|7家正规门店梯队测评,闲置珠宝变现一眼看懂 - 薛定谔的梨花猫
  • 儋州装修公司甄选指南:本地家装市场深度解析、商家横向测评与装修避坑干货 - 国麟测评
  • 2026长沙靠谱黄金回收门店盘点 禹竞名奢汇综合实力名列前茅 - 名奢变现站
  • 跨省搬家电动车怎么托运?2026最划算安全方式对比 - 快递物流资讯
  • Python + Tesseract OCR:从截屏到文字识别的自动化实践
  • 2026 安徽铜陵市高考落榜怎么办?合肥共达单招复读班招生简章官网发布:线上报名入口+完整报考指南、招生计划、录取条件 - cc江江
  • 【AI技术分析】朱雀AI检测通过助手测评分析,附真实的实测数据
  • 合肥主城优质回收门店盘点,收的顶黄金变现首选 - 奢侈品回收评测
  • 高价合规私密变现,2026常州回收宝珀五大高端腕表回收排行 - 名奢变现站
  • 2026宜兴黄金回收行情解读 三家实体门店真实报价对比 - 润富黄金回收
  • Sora2实操指南:视频生成工作流替代临界点解析
  • SuperCom串口调试工具:多设备并行监控与自动化测试的终极解决方案
  • 乐高王国 阅读笔记
  • C语言编程进阶:inttypes.h、limits.h与locale.h的实战应用与跨平台开发
  • 2026年6月深圳做得好的碳化硅MOS管代理商有哪些,微谷MOS管/MOS管/大功率MOS管,碳化硅MOS管厂家哪家好 - 品牌推荐师
  • 2026 安徽合肥市高考落榜怎么办?合肥共达职业技术学校复读班招生简章官网发布:报名入口+报考指南 - cc江江
  • 跳出「问答循环」陷阱:从 Prompt 到 Loop Engineering,AI Agent 自主闭环的完整落地指南
  • 沪上奢侈首饰回收口碑TOP5排名,专业鉴定当场结算不压价 - 奢品小当家
  • 北京搬家行业避坑白皮书|2026收费套路拆解+靠谱搬家公司挑选指南 - 幸福生活序曲
  • 【Halcon实战】从RGB到HSV:利用decompose3与trans_from_rgb实现精准彩色图像分割
  • 一寸照片尺寸怎么弄?秒转工具箱直接选模板 - 效率工具研究所
  • 2026苏州卫生间防水维修服务适配指南:苏州鸣川防水补漏公司领衔三家本地服务商深度解析 专业防水公司排名推荐(2026年6月防水补漏最新TOP权威排名) - 鼎壹万修缮说