文档中心

AgentKit

最佳实践

使用 TRAE 快速集成飞书机器人并部署至 AgentKit

TRAE 深度融合 AI 能力，是一名能够理解需求、调用工具并独立完成各类开发任务的“AI 开发工程师”，可帮助开发者高效推进每一个项目。本文档介绍如何利用 TRAE 的 AI 编程能力，将飞书机器人快速集成至 Agent 并部署至 AgentKit 智能体运行时。

项目结构

本教程项目核心结构如下：

前提条件

已开通 AgentKit 服务。若未开通，请登录 AgentKit 控制台，根据系统引导开通产品。

已安装 TRAE。

已开通方舟模型服务，并获取 API Key。

登录平台控制台获取 AK/SK，进入 API 访问密钥页面，创建并获取 Access Key ID 和 Secret Access Key。

步骤一：创建飞书机器人并获取配置

创建企业自建应用。

登录飞书开发者平台，单击“创建企业自建应用”按钮。

需配置应用名、描述及图标。

单击“创建”按钮，完成操作。

添加机器人能力。

在左侧目录树选择“应用能力 > 添加应用能力”。

选择“按能力添加”页签，单击“机器人”能力卡片的“添加”按钮。

配置权限。

在左侧目录树选择“开发配置 > 权限配置”，单击“批量导入/导出权限”按钮。

在“导入”页签中，将如下权限替换原有示例，单击“下一步，确认新增权限”按钮。

{
"scopes": {
"tenant": [
"im:chat:read",
"im:chat:update",
"im:message.group_at_msg:readonly",
"im:message.p2p_msg:readonly",
"im:message.pins:read",
"im:message.pins:write_only",
"im:message.reactions:read",
"im:message.reactions:write_only",
"im:message:readonly",
"im:message:recall",
"im:message:send_as_bot",
"im:message:send_multi_users",
"im:message:send_sys_msg",
"im:message:update",
"im:resource",
"contact:contact.base:readonly"
],
"user": []
}
}

在弹窗中确认权限无误后，单击“申请开通”按钮，完成操作。

说明

相关权限的具体含义可查看飞书 API 权限列表。

获取配置信息。

说明

获取的应用凭证和加密策略将用于配置 Agent 环境变量，请妥善保存。

获取应用凭证。

在左侧目录树选择“基础信息 > 凭证与基础信息”。

在“应用凭证”模块中，获取并记录 App ID 与 App Secret 信息。

（可选）若您期望提升飞书机器人数据传输的安全性，需获取加密策略。否则，请跳过本步骤。

在左侧目录树选择“开发配置 > 事件与回调”。

选择“加密策略”页签，获取并记录 Encrypt Key 和 Verification Token。

如果未设置 Encrypt Key，单击 Encrypt Key 旁的编辑按钮或重置按钮，配置 Encrypt Key。

发布应用。

单击顶部的“创建版本”按钮。

按需配置应用版本号、默认能力及更新说明等信息。了解更多

单击页面底部的“保存”按钮，创建版本。

单击页面右上角的“确认发布”按钮，完成应用发布。

步骤二：本地开发环境准备

推荐使用uv进行依赖管理和调试，速度更快且更稳定。

安装uv工具。

使用 curl 下载并安装

curl -LsSf https://astral.sh/uv/install.sh | sh

或使用 wget 下载并安装

wget -qO- https://astral.sh/uv/install.sh | sh

创建虚拟环境。

uv venv --python 3.12

激活虚拟环境。

环境激活后，您可以看到您的命令行开头会有括号标注的虚拟环境名称（通常为您的所在目录名称）。

source .venv/bin/activate

安装依赖。

uv pip install veadk-python agentkit-sdk-python lark-oapi python-dotenv

配置火山引擎 AK/SK 和飞书应用环境变量。

注意

由于 AK/SK 属于高敏感凭据，直接写入.env等静态配置文件存在泄露风险，在本地开发环境中配置火山引擎 AK/SK 时，推荐通过终端会话级环境变量导出的方式管理。

export VOLCENGINE_ACCESS_KEY=<your_access_key>
export VOLCENGINE_SECRET_KEY=<your_secret_key>
export FEISHU_APP_ID=cli_a917fd3c13f9**** # 飞书应用的 App ID
export FEISHU_APP_SECRET=LaOArqmCT9Z3sGHmZAEKCbZGgaWM**** # 飞书应用的 App Secret
# Encrypt Key 和 Verification Token 环境变量为飞书应用加密策略环境变量，配置后可提升数据传输的安全性，请按需配置。本示例已启用加密策略。
export FEISHU_ENCRYPT_KEY=oFEyNoLg62bG6HTC2fjj6dLcgSn0**** # 飞书应用的 Encrypt Key
export FEISHU_VERIFICATION_TOKEN=8A8ghhTlDi0ommEKSjLknb2VvQtJ**** # 飞书应用的 Verification Token

（可选）步骤三：创建 Agent

说明

如果您已经开发好 Agent 代码，可跳过本步骤，直接执行下一个步骤。

项目初始化。

初始化 Agent。

执行以下命令，生成一个预制好的 Agent 项目模板。

veadk create

您将被提示输入如下信息：

项目名称（同时也将会成为您的本地目录名称，如feishu_bot_agent）

方舟平台 API Key（可跳过，支持后续手动填写到环境变量文件中）

输入完毕后，您本地的项目结构如下：

feishu_bot_agent
├── __init__.py # 模块导出文件
├── .env # 环境变量文件，需要在这里配置您所需的环境变量信息
└── agent.py # Agent 定义文件

初始化待部署项目。

说明

请将 feishu_bot_agent 替换为实际 Agent 项目名称。

请将 agent.py 替换为实际 Agent 文件名称。

cd feishu_bot_agent
agentkit init --from-agent ./agent.py

初始化会自动生成以下文件：

agentkit-agent.py ：Agent 应用入口文件。将自动将您的智能体封装成符合 AgentKit 标准的 HTTP 服务，默认将封装成 AgentkitSimpleApp 应用框架，如希望使用其他应用框架可参考 SDK 概述。

requirements.txt：Python 依赖。

agentkit.yaml ：配置文件。

.dockerignore：构建 Docker 时需要忽略的文件或目录。

配置 AgentKit 部署参数。

配置向导会引导您完成设置，按 Enter 键表示使用默认配置。详细参数说明请参见创建 Agent。

agentkit config

配置完成后，会将配置内容写入 agentkit.yaml 配置文件。您可以执行 cat agentkit.yaml 命令进入配置文件，查看 Agent 源数据信息。

在 .env 文件中配置方舟模型 API Key 和飞书应用的环境变量。

MODEL_AGENT_API_KEY=07ee790b-2278-4f13-a2aa-6f7b460c**** # 方舟模型 API Key
FEISHU_APP_ID=cli_a9MODEL_AGENT_API_KEY17fd3c13f9**** # 飞书应用的 App ID
FEISHU_APP_SECRET=LaOArqmCT9Z3sGHmZAEKCbZGgaWM**** # 飞书应用的 App Secret
# Encrypt Key 和 Verification Token 环境变量为飞书应用加密策略环境变量，配置后可提升数据传输的安全性，请按需配置。本示例已启用加密策略。
FEISHU_ENCRYPT_KEY=oFEyNoLg62bG6HTC2fjj6dLcgSn0**** #飞书应用的 Encrypt Key
FEISHU_VERIFICATION_TOKEN=8A8ghhTlDi0ommEKSjLknb2VvQtJ**** # 飞书应用的 Verification Token

步骤四：使用 TRAE 为 Agent 接入飞书应用

本文以步骤三创建的 agentkit-agent.py 为例，展示使用 TRAE 的 AI 能力，将复杂的对接代码转化为简单的自然语言指令，生成包含飞书功能的Agent。

打开目标 Agent 项目（例如 feishu_bot_agent）。

使用 Vibe Coding 方式，让 AI 帮您编写代码。

说明

本文 Prompt 仅供参考，请根据实际业务需求自行调整与编写。

例如，将以下 Prompt 发送给 TRAE：

参考飞书平台官网文档 (https://open.larkoffice.com/document/server-docs/event-subscription-guide/event-subscription-configure-/request-url-configuration-case)中的 Python 示例，帮我修改 AgentKit 入口文件 agentkit-agent.py，实现通过 WebSocket 长连接对接飞书机器人。

核心目标：同时启动 Flask/FastAPI HTTP 服务（用于 AgentKit 健康检查）和飞书 WebSocket 客户端（用于接收消息），确保两者互不阻塞。

关键实现要求：

并发架构设计：

主线程：运行 AgentKit 的 HTTP 服务 (app.run)，监听 8000/8001 端口，响应 /ping 健康检查。

Agent线程：创建独立的 asyncio 事件循环（ agent_loop），专门运行耗时的 Agent 任务（runner.run），避免阻塞 WebSocket 心跳。

WebSocket 线程：在后台线程中启动飞书客户端（ws_client.start()），处理消息接收。

消息处理流程：

在 WebSocket 回调中，解析 JSON 消息内容。

使用 asyncio.run_coroutine_threadsafe 将任务调度到 Agent 线程执行。

禁止在回调中同步等待结果（不要用future.result()），防止心跳超时导致飞书重试。

兼容性与依赖：

使用 lark-oapi(v1.5.x+)。

正确导入 P2ImMessageReceiveV1 和 ReplyMessageRequest。

兼容 AgentKit 的 Invoke 调试接口，在 HTTP 入口 run() 中判断，如果 Agent 线程已启动，则调度过去，否则在当前线程执行。

配置加载：从环境变量读取 FEISHU_APP_ID，FEISHU_APP_SECRET，FEISHU_ENCRYPT_KEY，FEISHU_VERIFICATION_TOKEN 等配置。

说明

若 .env 文件未配置飞书应用的加密策略环境变量信息，则无需在提示词中要求读取 FEISHU_ENCRYPT_KEY，FEISHU_VERIFICATION_TOKEN。本示例已启用加密策略。

TRAE 生成的代码如下所示：

您也可以直接复制，替换 agentkit-agent.py 内容，但需确保 requirements.txt 中包含以下依赖。

# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

'''
**AgentKit Wrapper for User-Defined Agent**

This file wraps your Agent definition (agent.py) to make it deployable with AgentKit.

Your Agent is imported from: agent
Agent variable: root_agent
'''
import logging
import os
import json
import asyncio
import threading
from concurrent.futures import Future
from dotenv import load_dotenv

# Import user's Agent definition
from agent import root_agent

from veadk import Runner
from agentkit.apps import AgentkitSimpleApp

# Import Lark/Feishu SDK
# 使用 lark_oapi v1.5.x+
import lark_oapi as lark
from lark_oapi.api.im.v1.model.p2_im_message_receive_v1 import P2ImMessageReceiveV1
from lark_oapi.api.im.v1 import ReplyMessageRequest
from lark_oapi.api.im.v1 import ReplyMessageRequestBody

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Load env
load_dotenv()

app = AgentkitSimpleApp()

# Use user's Agent
runner = Runner(agent=root_agent)

# --- Global Variables for Threading ---
# 全局变量用于存储独立线程的 Event Loop，确保 Agent 的异步任务不会阻塞 WebSocket 心跳
agent_loop = None
agent_ready = threading.Event()

@app.entrypoint
async def run(payload: dict, headers: dict) -> str:
"""
Main entrypoint for the Agent.

This wrapper handles the standard AgentKit request/response protocol
and delegates to your Agent for processing.
"""
prompt = payload["prompt"]
user_id = headers["user_id"]
session_id = headers["session_id"]

logger.info(
f"Running agent with prompt: {prompt}, user_id: {user_id}, session_id: {session_id}"
)

# Check if agent_loop is active (Feishu Bot mode)
if agent_loop and agent_loop.is_running():
# Dispatch execution to the dedicated agent thread/loop to avoid loop conflicts
logger.info("Dispatching run request to agent_loop")
future = asyncio.run_coroutine_threadsafe(
runner.run(messages=prompt, user_id=user_id, session_id=session_id),
agent_loop
)
response = await asyncio.wrap_future(future)
else:
# Fallback: Execute directly on the current loop (Standard HTTP mode)
logger.info("Running request on main loop")
response = await runner.run(
messages=prompt,
user_id=user_id,
session_id=session_id
)

logger.info(f"Run response: {response}")
return response

@app.ping
def ping() -> str:
"""Health check endpoint."""
return "pong!"

# --- Agent Execution Thread ---

async def run_agent_async(prompt: str, user_id: str, session_id: str) -> str:
"""Async wrapper for running the agent."""
logger.info(f"Running agent async: user_id={user_id}, session_id={session_id}")
return await runner.run(messages=prompt, user_id=user_id, session_id=session_id)

def start_agent_loop():
"""
启动一个独立的线程来运行 asyncio 事件循环。

原因：
1. 飞书 WebSocket 客户端 (ws_client.start()) 会阻塞主线程。
2. 如果我们在 WebSocket 回调中直接调用 asyncio.run()，会因为当前已在事件循环中（或环境限制）而报错。
3. 因此，我们需要一个专门的线程 (Agent Thread) 来处理 Agent 的异步任务 (runner.run)。
"""
global agent_loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
agent_loop = loop
agent_ready.set()
logger.info("Agent event loop started.")
loop.run_forever()

# --- Feishu Bot Integration ---

def start_feishu_bot():
"""
启动飞书机器人（WebSocket 长连接模式）。
该模式不需要公网 IP，适合本地开发和内网部署。
"""
app_id = os.getenv("FEISHU_APP_ID")
app_secret = os.getenv("FEISHU_APP_SECRET")
encrypt_key = os.getenv("FEISHU_ENCRYPT_KEY", "")
verification_token = os.getenv("FEISHU_VERIFICATION_TOKEN", "")

if not app_id or not app_secret:
logger.warning("FEISHU_APP_ID or FEISHU_APP_SECRET not found. Skipping Feishu Bot startup.")
return

# 1. 启动独立的 Agent 线程
# 使用 daemon=True，这样主程序退出时线程也会自动退出
agent_thread = threading.Thread(target=start_agent_loop, daemon=True)
agent_thread.start()

# 等待 Agent Loop 准备就绪
agent_ready.wait()

# 2. 初始化飞书 API 客户端 (用于发送回复)
client = lark.Client.builder() \
.app_id(app_id) \
.app_secret(app_secret) \
.log_level(lark.LogLevel.DEBUG) \
.build()

# 3. 定义事件处理函数
# 异步处理 Agent 执行和回复，避免阻塞 WebSocket 回调
async def process_message_and_reply(text: str, user_id: str, chat_id: str, message_id: str):
try:
logger.info(f"[Feishu] Processing agent task for message_id: {message_id}")
response_text = await runner.run(messages=text, user_id=user_id, session_id=chat_id)
logger.info(f"[Feishu] Agent response: {response_text}")

# 回复消息给飞书
client.im.v1.message.reply(
ReplyMessageRequest.builder()
.message_id(message_id)
.request_body(ReplyMessageRequestBody.builder()
.content(json.dumps({"text": response_text}))
.msg_type("text")
.build())
.build()
)
except Exception as e:
logger.error(f"[Feishu] Async processing failed: {e}", exc_info=True)

def do_p2_im_message_receive_v1(data: P2ImMessageReceiveV1) -> None:
try:
# 解析消息内容
content_json = json.loads(data.event.message.content)
text = content_json.get("text", "")
chat_id = data.event.message.chat_id
user_id = data.event.sender.sender_id.open_id
message_id = data.event.message.message_id

logger.info(f"[Feishu] Received message: {text} from {user_id}")

if not agent_loop:
logger.error("[Feishu] Agent loop not initialized!")
return

# 4. 将任务提交给独立线程异步执行
# 关键修改：不再使用 future.result() 阻塞等待，而是直接提交任务后立即返回
# 这样 WebSocket 客户端可以立即确认收到消息，避免因处理超时导致飞书重试
asyncio.run_coroutine_threadsafe(
process_message_and_reply(text, user_id, chat_id, message_id),
agent_loop
)

logger.info(f"[Feishu] Task scheduled for message_id: {message_id}")

except Exception as e:
logger.error(f"[Feishu] Error handling message: {e}", exc_info=True)

# 6. 注册事件处理器
event_handler = lark.EventDispatcherHandler.builder(encrypt_key, verification_token) \
.register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \
.build()

# 7. 初始化 WebSocket 客户端
ws_client = lark.ws.Client(
app_id,
app_secret,
event_handler=event_handler,
log_level=lark.LogLevel.DEBUG
)

# 8. 启动 WebSocket 客户端 (在独立线程中，避免阻塞 HTTP 服务)
ws_thread = threading.Thread(target=ws_client.start, daemon=True)
ws_thread.start()
logger.info("Feishu WebSocket Client started in background thread.")

if __name__ == "__main__":
# Check if we want to run Feishu Bot mode or Standard AgentKit Web mode
# For this demo, we can try to run Feishu Bot if env vars are present,
# otherwise default to standard app.run().

if os.getenv("FEISHU_APP_ID") and os.getenv("ENABLE_FEISHU_BOT", "true").lower() == "true":
start_feishu_bot()

# 无论是否启动飞书 Bot，都启动 HTTP 服务以响应 AgentKit 健康检查
app.run(host="0.0.0.0", port=8000)

说明

上述代码已包含异步并发处理和自动回复逻辑。

步骤五：运行与测试

启动本地服务，验证 Agent 与飞书应用连接成功。

说明

建议直接在 TRAE 中启动本地服务，并根据提示完成必要的依赖和修改。

cd feishu_bot_agent
python agentkit-agent.py

如果回显类似 [INFO] connected to wss://msg-frontier.feishu.cn/... 的日志，说明连接成功。

配置飞书应用事件与回调。

登录飞书开发者平台。

选择“企业自建应用”页签，并单击您创建的飞书开放应用。

在左侧目录树选择“开发配置 > 事件与回调”。

事件配置

选择“事件配置”页签，单击“订阅方式”旁的编辑按钮。

选择“使用长连接接收事件”，并单击“保存”按钮。

在“已添加事件”区域，单击“添加事件”按钮。

在添加事件对话框中，选择“应用身份订阅”页签，并勾选“接收消息”及其它您需要订阅的事件，单击“确认添加”按钮。

（可选）若您期望将机器人添加到聊天群组中使用，需开通获取群组消息权限。否则，请跳过本步骤。

在“已添加事件”模块中，单击“接收消息”事件“所需权限”列的“查看其它权限”按钮。

单击“获取群组中所有消息（敏感权限）”。

在弹窗中单击“开通权限”按钮，申请开通。

回调配置

选择“回调配置”页签，单击“订阅方式”旁的编辑按钮。

选择“使用长连接接收回调”，并单击“保存”按钮。

版本发布。

单击顶部的“创建版本”按钮。

按需配置应用版本号、默认能力及更新说明等信息。了解更多

单击页面底部的“保存”按钮，创建版本。

单击页面右上角的“确认发布”按钮，完成应用发布。

至此，您已成功集成飞书机器人，可正式在飞书中使用您的定制化机器人。

例如，发送“你好”，若机器人回复，则表示Agent与飞书机器人集成成功。

步骤六：云端部署

如果需要将服务长期运行，可使用 agentkit 命令将其部署到云端 AgentKit 智能体运行时。

配置环境变量。

部署至 AgentKit 前，您需要将火山引擎AK/SK和飞书应用环境变量同步到 agentkit.yaml 文件的 common.runtime_envs 中，否则云端服务无法读取配置。示例如下：

common:
runtime_envs:
VOLCENGINE_ACCESS_KEY: your_access_key
VOLCENGINE_SECRET_KEY: your_secret_key
MODEL_AGENT_API_KEY: 07ee790b-2278-4f13-a2aa-6f7b460c**** # 方舟模型 API Key
FEISHU_APP_ID: cli_a917fd3c13f9**** # 飞书应用的 App ID
FEISHU_APP_SECRET: LaOArqmCT9Z3sGHmZAEKCbZGgaWM**** # 飞书应用的 App Secret
# Encrypt Key 和 Verification Token 为飞书应用加密策略环境变量，配置后可提升数据传输的安全性，请按需配置。
FEISHU_ENCRYPT_KEY: oFEyNoLg62bG6HTC2fjj6dLcgSn0**** # 飞书应用的 Encrypt Key
FEISHU_VERIFICATION_TOKEN: 8A8ghhTlDi0ommEKSjLknb2VvQtJ**** # 飞书应用的 Verification Token

发布智能体。

agentkit launch

说明

首次部署大约需要 2～3 分钟，请耐心等待。

长连接模式下，云端部署后无需修改飞书后台的请求地址，AgentKit 会主动向飞书建立 WebSocket 连接。

当回显智能体运行时状态为 Ready 时，则表示创建成功。

查看智能体运行时的运行日志确认 WebSocket Client 已启动。

在左侧目录树选择“智能体运行时”，单击刚创建的智能体运行时实例。

选择“日志”页签，在“表格”页签中确认已有以下信息，表示 WebSocket Client 已启动。