尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

Agent与飞书集成:从任务闭环到工程落地

Agent与飞书集成:从任务闭环到工程落地 “豆包工作”Agent 产品发布后AI 办公助手的形态出现了一次明显变化它不再是一个独立网页里来回对话的聊天机器人而是与飞书深度打通直接出现在群聊、文档、多维表格和审批流旁边。对开发者和企业系统负责人来说这个产品背后真正值得研究的是 Agent 与协同办公平台之间的集成链路Agent 如何拿到飞书里的上下文如何调用工具完成任务又如何把结果写回业务系统。这里不讨论产品发布会的演示效果而是从工程落地角度拆解办公 Agent 与飞书打通时的核心机制。你会看到 Agent 的基本工作原理、飞书开放平台提供的接入点、一个最小的飞书消息 Agent 示例以及权限、安全、排错和生产部署需要注意的细节。即使不直接使用“豆包工作”这套分析同样适用于自研 Agent 接入飞书或其他协同办公平台的场景。1. 先理解“豆包工作”Agent 与飞书打通的产品逻辑1.1 Agent 的核心不是“对话”而是“任务闭环”很多团队把聊天机器人误当成 Agent实际上两者的差异非常明显。聊天机器人的核心是“回复”用户问一句模型答一句交互结束。Agent 的核心是“任务闭环”用户提出一个目标Agent 需要自己拆解步骤、选择工具、执行动作、观察结果再决定继续执行还是向用户汇报。用一句话概括Agent 大模型 规划能力 工具调用 记忆 反馈循环。大模型负责理解和生成自然语言工具调用负责让模型突破“只会说话”的边界记忆负责保存上下文和历史状态反馈循环负责根据执行结果修正下一步动作。办公场景里任务闭环的价值会被明显放大。以“帮我把项目 A 的进度整理成周报”为例聊天机器人只能基于已有文字生成一段周报草稿Agent 则可能需要先访问飞书多维表格里的任务记录读取每个成员的状态再结合文档模板生成周报最后把结果发送到指定群聊或文档。这里每一步都是一次工具调用而不是一次单纯的自然语言生成。维度聊天机器人办公 Agent目标生成回复完成任务数据来源用户消息和模型知识办公系统、业务 API、外部服务是否需要工具通常不需要必须调用工具执行路径单一问答多步规划与反馈失败处理重新问一次观察结果、修正计划、重试或上报这也是“豆包工作”Agent 与飞书深度打通后值得关注的原因办公 Agent 只有真正接入团队日常使用的系统才能完成从“理解意图”到“改变业务数据”的闭环。1.2 飞书是 Agent 最重要的办公上下文来源办公场景里最有价值的数据往往不在外部互联网而在企业内部的协同软件中。飞书里沉淀了组织架构、群聊消息、日历日程、在线文档、多维表格、审批流程等大量结构化和非结构化数据。Agent 如果只靠用户一句自然语言很难知道项目进度、负责人、截止日期和审批状态。飞书深度打通意味着 Agent 可以主动读取这些数据。比如用户问“这个需求下周能上线吗”Agent 可以读取多维表格中的排期记录查看相关任务的负责人和状态再结合当前日期判断风险。这里的“打通”不是把飞书当作一个临时消息通道而是让飞书成为 Agent 的上下文仓库和动作执行环境。从技术角度看飞书开放平台提供了应用、机器人、事件订阅、云文档、多维表格、审批、日历等一系列能力。这些能力组合起来就像一个“办公操作系统”的接口层。Agent 通过这些接口读取数据、发起动作、接收反馈最终把自然语言指令翻译成真正的业务操作。结合“豆包工作”Agent 的定位来看办公场景的下一步竞争不只是模型能力更是“Agent 能操作多少个企业系统”。飞书深度打通解决的就是这个操作系统连接问题。1.3 办公 Agent 的典型能力边界根据目前办公类 Agent 产品的常见形态能力通常集中在以下几类信息查询查文档、查表格、查日程、查审批状态。内容生成写周报、写会议纪要、写邮件草稿、生成总结。数据操作新增多维表格记录、更新任务状态、发起审批。协同动作拉群、发消息、安排会议、提醒事项。知识问答基于团队知识库回答业务问题。需要注意不同产品在“数据操作”和“协同动作”上的放开程度不同。写入类操作一旦放开风险也成倍增加比如误改数据、重复创建任务、发送错误消息。后续章节会专门讨论授权、权限和审计问题。本文后面的示例不追求复刻“豆包工作”完整产品能力而是实现一个最小闭环用户通过飞书机器人输入指令Agent 调用大模型判断意图查询或写入多维表格再把结果发回飞书。把这个链路跑通后其他能力基本都是在工具列表里做加法。2. 飞书开放平台为 Agent 提供了哪些接入点2.1 自建应用与机器人形态是首选接入方式飞书开放平台支持多种应用形态自建应用是最适合企业内 Agent 的接入方式。自建应用可以配置机器人、网页应用、权限点、事件订阅和 API 调用能力而且发布范围可以限定在企业内部灵活度最高。接入 Agent 时建议优先使用“企业自建应用 机器人”的组合。机器人负责接收用户在私聊、群聊中的消息Agent 服务负责处理消息并调用飞书 API 返回结果。如果需要给用户提供配置页面再添加一个网页应用入口。创建自建应用的核心流程登录飞书开放平台进入开发者后台。创建企业自建应用。在“应用能力”中添加机器人。配置权限点和事件订阅。创建版本并发布等待企业管理员审核。整个过程不需要写代码但权限点和事件订阅的配置会影响后面的开发需要提前规划好。2.2 消息链路事件订阅、消息接收与消息发送飞书 Agent 最常用的交互方式是消息。用户私聊机器人或者在群聊中 机器人飞书都会向应用配置的事件订阅地址推送一条事件常见事件类型是im.message.receive_v1。事件订阅地址必须是一个公网可访问的 HTTPS 接口。飞书在配置时会发送一次 URL 验证请求校验通过后才会正式推送业务事件。事件内容默认是 JSON 格式如果配置了加密事件内容会用 Encrypt Key 加密后再推送服务端需要先解密才能读取。Agent 处理完消息后通过飞书发送消息接口把结果返回给用户。消息类型可以是文本、富文本、卡片等。卡片消息适合展示结构化信息比如查询结果、审批状态、多步骤执行进度。链路节点作用常见实现机器人接收用户输入飞书应用能力中开启事件订阅推送消息事件到 Agent 服务HTTPS 回调或长连接Agent 服务解析意图、调用 LLM、执行工具FastAPI / Spring Boot / Node.js消息发送 API把 Agent 结果返回给用户im/v1/messages业务 API读取和写入业务数据多维表格、文档、审批等如果企业内网不方便暴露公网回调地址飞书也支持长连接模式应用通过 WebSocket 长连接接收事件避免公网端口暴露。选择哪种方式取决于公司网络策略和现有基础设施。2.3 业务数据链路多维表格、文档与审批飞书多维表格是 Agent 最友好的业务数据载体之一。它像数据库一样支持行、列、字段类型又能通过图形界面让非技术人员维护数据。Agent 可以通过多维表格 API 读取记录、筛选记录、新增记录、更新记录非常适合做任务管理、需求跟踪、知识整理等场景。云文档 API 可以让 Agent 读取和写入文档内容适合生成周报、会议纪要、项目方案。审批 API 则让 Agent 发起和查询审批实例比如请假、报销、采购审批。日程 API 可以用来查询忙闲、创建日程和会议邀请。接入业务数据时需要先明确一个边界Agent 读取数据通常安全风险可控写入数据必须谨慎。推荐的最小起步方案是“先只读、后写入”查询类功能稳定后再开放有权限校验的写入功能。写入类操作最好带二次确认比如 Agent 在生成新增任务前先向用户展示要写入的内容。3. 搭建一个最小飞书 Agent豆包工作式场景的简化版3.1 示例需求一个部门任务助手用一个具体场景来演示完整链路。假设有一个部门任务表放在飞书多维表格中字段包括任务标题、负责人、截止日期、状态。现在要让飞书机器人成为一个“任务助手”支持两种能力用户说“查一下张三负责的任务”Agent 查询多维表格并返回结果。用户说“新增一个任务完成 API 联调负责人李四截止日期 2025-06-30”Agent 向多维表格写入一条记录。这个需求虽然简单但覆盖了消息接收、LLM 函数调用、多维表格查询和写入、结果回发的完整闭环。跑通后再扩展其他工具会非常容易。3.2 项目结构与依赖示例使用 Python 3.10 和 FastAPI 实现服务端使用 httpx 调用飞书 API使用 OpenAI 兼容接口调用大模型。项目结构保持轻量方便理解。feishu-agent/ ├── app.py # FastAPI 入口和事件回调 ├── agent.py # Agent 调度循环 ├── feishu_client.py # 飞书 token、消息、多维表格 API ├── config.py # 环境变量配置 └── requirements.txt依赖文件如下fastapi uvicorn httpx cryptography openai python-dotenv其中cryptography用于飞书事件内容解密openai用于调用兼容 OpenAI 协议的大模型接口。实际项目中模型服务商可能不同建议把 LLM 调用封装成一个独立模块方便替换。3.3 创建飞书自建应用并完成基础配置在飞书开放平台创建自建应用后需要完成以下配置在“应用能力”中开启机器人。在“权限管理”中添加机器人消息、多维表格相关权限点。在“事件订阅”中配置请求地址选择im.message.receive_v1事件。获取应用的 App ID、App Secret、Encrypt Key 和 Verification Token。配置项建议通过环境变量管理避免写入代码仓库。export FEISHU_APP_IDcli_xxxxxxxx export FEISHU_APP_SECRETyour_app_secret export FEISHU_ENCRYPT_KEYyour_encrypt_key export FEISHU_VERIFICATION_TOKENyour_verification_token export LLM_API_KEYyour_llm_api_key export LLM_BASE_URLhttps://api.xxx.com/v1 export LLM_MODELyour-model-name配置项用途来源App ID标识应用飞书开发者后台App Secret获取 token 时使用飞书开发者后台Encrypt Key解密推送的事件内容飞书事件订阅配置Verification Token校验事件来源飞书事件订阅配置LLM_API_KEY调用大模型模型服务商配置权限点时建议遵循最小权限原则。示例需要读取和写入多维表格因此申请对应权限消息类权限可以更细化为“获取用户发给机器人的消息”和“以机器人身份发送消息”。3.4 实现事件回调服务FastAPI 入口收到飞书推送的事件后先判断是否 URL 验证请求再处理加密事件最后根据事件类型分发到业务逻辑。# app.py import hashlib import base64 import json from fastapi import FastAPI, Request from config import FEISHU_ENCRYPT_KEY, FEISHU_VERIFICATION_TOKEN from agent import handle_message_event app FastAPI() def decrypt_event(encrypt_key: str, encrypt_data: str) - dict: # 飞书事件加密规则在不同 SDK 版本中实现略有差异 # 这里给出一种常见实现落地前要和你使用的 SDK 版本比对。 digest hashlib.md5(encrypt_key.encode(utf-8)).digest() key digest.hex().encode(utf-8) # 32 字节 ASCII key iv key[:16] from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives import padding encrypted base64.b64decode(encrypt_data) cipher Cipher(algorithms.AES(key), modes.CBC(iv)) decryptor cipher.decryptor() padded decryptor.update(encrypted) decryptor.finalize() unpadder padding.PKCS7(128).unpadder() plaintext unpadder.update(padded) unpadder.finalize() return json.loads(plaintext.decode(utf-8)) app.post(/webhook/feishu/event) async def handle_event(request: Request): body await request.json() # URL 验证请求 if body.get(type) url_verification: return {challenge: body.get(challenge)} # 解密事件内容 if body.get(encrypt): body decrypt_event(FEISHU_ENCRYPT_KEY, body[encrypt]) # token 校验 header body.get(header, {}) verify_token header.get(token) or body.get(token) if verify_token ! FEISHU_VERIFICATION_TOKEN: return {code: 1, msg: invalid token} event_type header.get(event_type) or body.get(type) if event_type im.message.receive_v1: event body.get(event, {}) await handle_message_event(event) return {code: 0, msg: success}这里有一个关键点飞书新旧版本事件结构存在差异。新版事件中事件类型放在header.event_typetoken 放在header.token旧版可能直接放在 body 顶层。示例代码做了兼容处理真正接入时还需要根据本地版本和日志确认。事件回调接口应该尽快返回。示例中handle_message_event是异步函数但如果消息处理耗时较长建议在回调内部先做异步化处理而不是阻塞整个请求。3.5 实现 Agent 调度与工具调用Agent 调度层的核心逻辑是把用户消息拼装成对话交给大模型判断是否调用工具如果调用工具执行工具后把结果继续交给模型生成最终回复。# agent.py import json from openai import AsyncOpenAI import config from feishu_client import query_tasks, create_task, send_text SYSTEM_PROMPT ( 你是一个办公任务助手。你可以查询任务列表也可以新增任务。 查询结果要整理成简洁的中文回复。 ) TOOL_SCHEMAS [ { type: function, function: { name: query_tasks, description: 按负责人查询多维表格中的任务列表, parameters: { type: object, properties: { owner: {type: string, description: 负责人姓名} }, required: [owner], }, }, }, { type: function, function: { name: create_task, description: 新增一条任务记录到多维表格, parameters: { type: object, properties: { title: {type: string, description: 任务标题}, owner: {type: string, description: 负责人}, due: {type: string, description: 截止日期格式 YYYY-MM-DD}, }, required: [title, owner], }, }, }, ] client AsyncOpenAI(api_keyconfig.LLM_API_KEY, base_urlconfig.LLM_BASE_URL) TOOLS { query_tasks: query_tasks, create_task: create_task, } async def run_agent(user_text: str) - str: messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_text}, ] for _ in range(3): response await client.chat.completions.create( modelconfig.LLM_MODEL, messagesmessages, toolsTOOL_SCHEMAS, tool_choiceauto, ) message response.choices[0].message if not message.tool_calls: return message.content or 没有找到合适的回复。 messages.append({ role: assistant, content: message.content, tool_calls: [ { id: tool_call.id, type: function, function: tool_call.function.model_dump(), } for tool_call in message.tool_calls ], }) for tool_call in message.tool_calls: name tool_call.function.name args json.loads(tool_call.function.arguments or {}) try: result await TOOLS[name](**args) except Exception as exc: result {error: str(exc)} messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) return 多次尝试后仍未完成任务请检查任务描述或联系管理员。函数调用的关键点是工具 JSON Schema。模型并不知道每个工具怎么实现它只看名称、描述和参数。因此描述必须写得清楚参数名要和实际函数参数保持一致。如果工具描述含糊模型会经常误选工具或填错参数。消息事件处理函数从飞书事件中提取用户输入然后执行 Agent最后把结果发送回当前会话。# agent.py 继续 import json async def handle_message_event(event: dict): message event.get(message, {}) chat_id message.get(chat_id) message_type message.get(message_type) if message_type ! text: await send_text(chat_id, 目前只支持文本消息。) return try: content json.loads(message.get(content, {})) user_text content.get(text, ).strip() except json.JSONDecodeError: await send_text(chat_id, 消息解析失败请稍后再试。) return if not user_text: await send_text(chat_id, 消息内容为空。) return reply await run_agent(user_text) await send_text(chat_id, reply)如果是在群聊中 机器人飞书消息内容里的文本会包含用户昵称或_user_之类的内容实际使用时需要清理无用前缀。这个细节在排错章节会再提到。3.6 实现飞书多维表格查询与写入飞书客户端模块负责三件事获取 tenant_access_token、查询多维表格记录、新增多维表格记录。# feishu_client.py import time import json import httpx import config _token_cache {value: None, expire_at: 0} APP_TOKEN your_app_token TABLE_ID your_table_id async def get_tenant_access_token() - str: now time.time() if _token_cache[value] and _token_cache[expire_at] now 60: return _token_cache[value] async with httpx.AsyncClient() as client: resp await client.post( https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal, json{app_id: config.FEISHU_APP_ID, app_secret: config.FEISHU_APP_SECRET}, ) data resp.json() if data.get(code) ! 0: raise RuntimeError(f获取 token 失败: {data.get(msg)}) _token_cache[value] data[tenant_access_token] _token_cache[expire_at] now data.get(expire, 7200) return _token_cache[value] async def query_tasks(owner: str) - list[dict]: token await get_tenant_access_token() url fhttps://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records headers {Authorization: fBearer {token}} # filter 语法以飞书官方文档为准不同版本可能不同 params
返回列表