飞书 - AgentKit 文档

功能说明

FeishuChannelExtension 把飞书机器人的入站消息桥接到 VeADK Runner。

本页是把飞书作为入站消息渠道：飞书用户发来的消息会触发智能体运行。若需让智能体主动调用飞书能力，请参见飞书工具。

它监听飞书机器人的消息事件，并把飞书的会话身份映射到 VeADK：以消息发送者作为 Runner 的用户标识，以消息所在的话题作为会话标识；话题不存在时，回退到当前群聊或单聊作为会话标识。1.0.5 支持提取文本消息、互动卡片与合并转发消息中的文本内容，并兼容同步 Channel SDK 生命周期。

话题历史与引用消息默认开启。扩展会读取相关飞书消息并将内容发送给模型；应用配置了会话、长期记忆或日志时，这些内容还可能按对应策略保存。启用机器人前应确认消息读取范围、模型数据处理和保留策略符合业务要求；不需要上下文时关闭 include_thread_history 与 include_parent_message。

收到话题消息或回复消息时，扩展会自动获取话题历史或被引用的父消息，并将这些上下文拼接到用户消息前再转发给智能体，使智能体能够参考完整对话上下文。详见话题历史与引用消息。

环境变量与前提

安装：

pip install "veadk-python[extensions]"

如果只想要这一项能力，也可以单独安装：

pip install lark-oapi

环境变量：

TOOL_FEISHU_CHANNEL_APP_ID
TOOL_FEISHU_CHANNEL_APP_SECRET
TOOL_FEISHU_CHANNEL_TRANSPORT：默认 ws
TOOL_FEISHU_CHANNEL_STREAMING：是否开启流式输出，默认 false
TOOL_FEISHU_CHANNEL_REACTIONS：是否在收到消息时回复“收到”表情，默认 false
TOOL_FEISHU_CHANNEL_RECONNECT_INTERVAL：WebSocket 断开后重连的间隔秒数，默认 3
TOOL_FEISHU_CHANNEL_DRAIN_TIMEOUT：优雅关停时等待已接收消息处理完成的最大秒数，默认 300

或在 config.yaml 中配置：

config.yaml

tool:
  feishu_channel:
    app_id: cli_xxx
    transport: ws
    streaming: true
    reactions: true

在飞书开放平台启用应用的机器人能力，配置接收消息事件，并开通发送消息和读取所需消息上下文的权限。将应用发布到测试范围后，把机器人加入测试会话。先完成模型配置，再在启动进程前设置凭据

export TOOL_FEISHU_CHANNEL_APP_ID="your-app-id"
export TOOL_FEISHU_CHANNEL_APP_SECRET="your-app-secret"

使用方法

feishu_bot.py

import asyncio

from veadk import Agent, Runner
from veadk.extensions import FeishuChannelExtension
from veadk.memory.short_term_memory import ShortTermMemory

agent = Agent(
    name="feishu_agent",
    instruction="你是一个通过飞书机器人与用户沟通的助手。",
)

runner = Runner(
    agent=agent,
    app_name="veadk_feishu_demo",
    user_id="veadk_feishu_default_user",
    short_term_memory=ShortTermMemory(),
)

channel = FeishuChannelExtension(
    runner=runner,
    channel_kwargs={"transport": "ws"},
)


async def main():
    await channel.connect()


if __name__ == "__main__":
    asyncio.run(main())

优雅关停

FeishuChannelExtension 提供独立的生命周期方法，用于在 ASGI 应用（如 FastAPI、Starlette）的启动与关闭钩子中管理 WebSocket 连接，尽量完成滚动更新或退出时已接收的消息。关停流程分两步：先关闭 WebSocket 以停止接收新消息（飞书会将新事件重新路由到仍持有连接的其他实例），再等待已接收的消息完成回复。消息回复通过飞书 OpenAPI HTTP 客户端发送，不依赖 WebSocket，因此关停期间已接收的消息仍能正常回复。

仅使用 connect() 直接运行的长连接场景不需要调用 start() / shutdown()；适用于进程退出前无需排空消息的简单脚本。在 ASGI 服务中托管时，使用 start() 与 shutdown() 才能获得优雅关停能力。

使用示例

在 ASGI 应用的启动与关闭钩子中调用 start() 与 shutdown()：

feishu_asgi.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from veadk.extensions import FeishuChannelExtension
from feishu_bot import runner  # 复用上方的 runner 定义

channel = FeishuChannelExtension(
    runner=runner,
    channel_kwargs={"transport": "ws"},
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    channel.start()  # 在运行中的事件循环内调用
    try:
        yield
    finally:
        await channel.shutdown()  # 关闭连接并排空在途消息


app = FastAPI(lifespan=lifespan)

方法

方法	类型	说明
`start(loop)`	同步	在运行中的应用事件循环内调用，捕获该循环并启动后台重连循环。可重复调用，已在运行时再次调用不会重复启动。`loop` 省略时使用当前运行的事件循环。
`await shutdown(*, drain_timeout)`	异步	关闭 WebSocket 并等待已接收消息处理完成，适合作为 ASGI 关停钩子。可重复调用，仅首次调用执行关停。`drain_timeout` 省略时使用 `TOOL_FEISHU_CHANNEL_DRAIN_TIMEOUT`。
`await drain(timeout)`	异步	等待所有在途消息任务完成，超过 `timeout` 秒后取消未完成任务。默认 `300.0`。
`is_draining`	属性	布尔值，表示是否已进入关停流程。重连循环据此停止重连。

shutdown() 会等待在途消息处理完成。若消息处理耗时较长且超过 drain_timeout，未完成的任务将被取消。生产环境应根据智能体平均响应时间调整 TOOL_FEISHU_CHANNEL_DRAIN_TIMEOUT，避免正常回复被中断。

环境变量

环境变量	默认值	说明
`TOOL_FEISHU_CHANNEL_RECONNECT_INTERVAL`	`3`	WebSocket 断开后重连的间隔秒数。
`TOOL_FEISHU_CHANNEL_DRAIN_TIMEOUT`	`300`	优雅关停时等待已接收消息处理完成的最大秒数。

参数

参数	类型	默认值	说明
`runner`	`Runner`	无	必填。处理飞书入站消息的 VeADK 执行器。
`app_id`	`str \| None`	`TOOL_FEISHU_CHANNEL_APP_ID`	飞书应用 ID；兼容回退到 `TOOL_LARK_ENDPOINT`。
`app_secret`	`str \| None`	`TOOL_FEISHU_CHANNEL_APP_SECRET`	飞书应用密钥；兼容回退到 `TOOL_LARK_API_KEY`。应通过环境变量或密钥服务提供。
`channel`	`Any \| None`	`None`	已创建的飞书 Channel 客户端；提供后不再自动创建客户端。
`session_id_factory`	`Callable[[Any], str] \| None`	话题、回复、会话或消息标识	自定义飞书消息到 VeADK `session_id` 的映射。
`user_id_factory`	`Callable[[Any], str] \| None`	发送者 `union_id`、`open_id` 或 `user_id`	自定义飞书发送者到 VeADK `user_id` 的映射。
`message_handler`	`Callable \| None`	`None`	自定义消息处理函数；未设置时调用 `runner`。可返回字符串、异步字符串或 `None`。
`response_formatter`	`Callable[[str], dict[str, str]] \| None`	`{"text": text}`	把智能体文本转换为飞书发送载荷。
`reply_in_thread`	`bool`	`True`	是否回复原消息，使输出保持在当前消息线程中。
`ignore_empty_messages`	`bool`	`True`	是否忽略无法提取出文本的消息。
`channel_kwargs`	`dict[str, Any] \| None`	`None`	创建 Channel 客户端时传入的选项；`transport` 默认读取 `TOOL_FEISHU_CHANNEL_TRANSPORT`，未设置时为 `ws`。
`streaming`	`bool`	`False`	是否使用流式回复；`TOOL_FEISHU_CHANNEL_STREAMING=true` 也会启用。
`reactions`	`bool`	`False`	收到消息时是否添加“收到”表情；`TOOL_FEISHU_CHANNEL_REACTIONS=true` 也会启用。
`include_parent_message`	`bool`	`True`	收到回复消息时是否获取并附带被引用的父消息内容。
`include_thread_history`	`bool`	`True`	收到话题消息时是否获取并附带该话题的历史消息。
`thread_history_limit`	`int`	`20`	获取话题历史消息的最大条数；小于 `1` 的值按 `1` 处理。

话题历史与引用消息

当消息来自飞书话题（thread）或是一条回复消息时，扩展会在把消息转发给智能体之前，自动获取相关上下文并拼接到用户消息文本之前：

话题历史：消息属于话题时，扩展获取该话题内按时间正序排列的历史消息（最多 thread_history_limit 条，不包含当前消息），将其作为话题上下文拼接到用户消息前。
引用消息：消息是一条回复但未归属话题时，扩展获取被引用的父消息；若父消息无法获取，则回退获取该回复链的根消息，将其作为引用上下文拼接到用户消息前。

话题历史优先于引用消息：当消息同时存在话题标识时，仅附加话题历史，不再单独获取父消息。获取到的消息内容会被还原为可读文本，覆盖文本消息、富文本（post）、互动卡片与合并转发消息。

获取话题历史与引用消息需要通过飞书 OpenAPI 调用消息接口，因此必须配置 app_id 与 app_secret（构造参数，或 TOOL_FEISHU_CHANNEL_APP_ID / TOOL_FEISHU_CHANNEL_APP_SECRET 环境变量，回退到 TOOL_LARK_ENDPOINT / TOOL_LARK_API_KEY）。未配置凭证时该能力会被跳过，不会报错。

使用示例

先保存并配置上方的 feishu_bot.py，再在同一目录创建以下文件：

feishu_bot_thread.py

from veadk.extensions import FeishuChannelExtension
from feishu_bot import runner

channel = FeishuChannelExtension(
    runner=runner,
    include_thread_history=True,
    thread_history_limit=30,
    include_parent_message=True,
)

关闭该能力，使扩展仅转发当前消息原文：

feishu_bot_plain.py

from veadk.extensions import FeishuChannelExtension
from feishu_bot import runner

channel = FeishuChannelExtension(
    runner=runner,
    include_parent_message=False,
    include_thread_history=False,
)

额外说明

默认使用飞书的 WebSocket 长连接模式，因此只要机器人已订阅消息事件，即可直接启动连接。
默认以回复原消息的形式发送，使 VeADK 的输出挂在当前飞书消息线程下。
可以通过 session_id_factory 与 user_id_factory 覆盖默认的身份映射逻辑。
connect() 与 disconnect() 同时兼容异步客户端和提供 start() / stop() 的同步 Channel SDK。
开启流式输出时，回复内容按增量片段直接拼接发送，不再因去重启发式而丢失或截断字符。

启动与验证

将第一个示例保存为 feishu_bot.py，运行 python feishu_bot.py 后保持进程运行。在测试会话中向机器人发送一条文本，确认收到回复。使用 ASGI 示例时，将 feishu_asgi.py 放在同一目录，安装 fastapi 和 uvicorn 后运行 uvicorn feishu_asgi:app --host 127.0.0.1 --port 8000；两种启动方式选择一种后面的历史消息开关示例仅替换 channel 配置，仍需启动连接。机器人无响应时，依次检查应用发布范围、事件订阅、机器人是否在会话中、消息权限以及模型配置；只有当前消息可用而历史缺失时，再检查历史读取权限与两个上下文开关

概览

快速开始

核心组件

可观测

扩展

前端

安全

部署

飞书

功能说明

环境变量与前提

使用方法

优雅关停

使用示例

方法

环境变量

参数

话题历史与引用消息

使用示例

额外说明

启动与验证

​功能说明

​环境变量与前提

​使用方法

​优雅关停

​使用示例

​方法

​环境变量

​参数

​话题历史与引用消息

​使用示例

​额外说明

​启动与验证

功能说明

环境变量与前提

使用方法

优雅关停

使用示例

方法

环境变量

参数

话题历史与引用消息

使用示例

额外说明

启动与验证