Agent与飞书集成:从任务闭环到工程落地
2026/8/29 3:39:09 网站建设 项目流程

“豆包工作”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 返回结果。如果需要给用户提供配置页面,再添加一个网页应用入口。

创建自建应用的核心流程:

  1. 登录飞书开放平台,进入开发者后台。
  2. 创建企业自建应用。
  3. 在“应用能力”中添加机器人。
  4. 配置权限点和事件订阅。
  5. 创建版本并发布,等待企业管理员审核。

整个过程不需要写代码,但权限点和事件订阅的配置会影响后面的开发,需要提前规划好。

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 创建飞书自建应用并完成基础配置

在飞书开放平台创建自建应用后,需要完成以下配置:

  1. 在“应用能力”中开启机器人。
  2. 在“权限管理”中添加机器人消息、多维表格相关权限点。
  3. 在“事件订阅”中配置请求地址,选择im.message.receive_v1事件。
  4. 获取应用的 App ID、App Secret、Encrypt Key 和 Verification Token。

配置项建议通过环境变量管理,避免写入代码仓库。

export FEISHU_APP_ID="cli_xxxxxxxx" export FEISHU_APP_SECRET="your_app_secret" export FEISHU_ENCRYPT_KEY="your_encrypt_key" export FEISHU_VERIFICATION_TOKEN="your_verification_token" export LLM_API_KEY="your_llm_api_key" export LLM_BASE_URL="https://api.xxx.com/v1" export LLM_MODEL="your-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_type,token 放在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_key=config.LLM_API_KEY, base_url=config.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( model=config.LLM_MODEL, messages=messages, tools=TOOL_SCHEMAS, tool_choice="auto", ) 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_ascii=False), }) 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 = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records" headers = {"Authorization": f"Bearer {token}"} # filter 语法以飞书官方文档为准,不同版本可能不同 params

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

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

立即咨询