botmux:让AI Agent通过飞书机器人实现移动端授权审批
2026/8/29 7:19:22 网站建设 项目流程

之前做本地 AI Agent 时,最折磨人的不是模型效果不好,也不是流程写不出来,而是每次 Agent 要执行敏感操作,都会在终端弹一个授权确认,等着我回到电脑前点一下。开会、通勤、午休的时候,任务卡住,白白浪费一串时间。后来我把飞书机器人接进来,用一套轻量级消息网关 botmux 把授权请求搬到手机端,才真正解决了这个“人机协同”的断点。

这篇教程会从背景、系统设计、飞书应用配置、botmux 代码实现、Agent 接入、运行验证和排错几个方面完整展开。核心思路是:让 Agent 在需要授权时自动发起请求,botmux 把请求包装成飞书卡片推给用户,用户点击“允许/拒绝”后,botmux 再把结果回传给 Agent。无论你是做 Agent 开发,还是想把飞书机器人和现有自动化流程打通,都可以直接参考这套方案。

1. 为什么需要 botmux 打通飞书与 Agent

1.1 AI Agent 执行链路中的授权场景

很多团队在落地 AI Agent 时,不会让 Agent 拥有所有权限,而是给关键步骤加一道人工确认。比如删除数据库记录、发送对外邮件、调用外部付费 API、修改生产配置等,都需要在真正执行前确认一次。这个设计是符合安全预期的,但问题出在“确认”这个动作发生在哪里。

如果你只在终端里实现确认,那么 Agent 部署在哪台机器,人就得在哪台机器附近。一旦 Agent 是 7x24 小时运行的,你就得随时盯着电脑,否则任务会在授权点无限等待。更麻烦的是,Agent 的授权状态往往是有时间窗口的,错过之后任务重启,前面浪费的时间又得重来。飞书这类即时通讯工具天然适合承担“移动审批”的工作,于是把授权请求推到飞书就成了一个很自然的需求。

1.2 botmux 是什么

botmux 不是一个庞大复杂的平台,而是一个轻量级消息网关。名字是 Bot Multiplexer 的缩写,意思是“机器人消息多路复用器”。它一般部署在 Agent 服务与飞书开放平台之间,负责做三件事:接收 Agent 的授权请求、把请求组装成飞书消息卡片、接收飞书按钮回调并转回给 Agent。

你可以把 botmux 看成一层适配器。它不需要关心 Agent 内部用什么框架、怎么写业务逻辑,只要 Agent 按约定调用 botmux 的 HTTP 接口,botmux 就能把授权流程变成飞书卡片事件。这样做的最大好处是解耦:Agent 不需要知道飞书 API 细节,飞书也不直接对接 Agent,所有消息互通都集中在 botmux 这一层。

用飞书多维表格记录审计日志时,botmux 还可以作为日志写入的统一入口。授权请求、用户点击结果、过期时间都能落表,方便事后追溯,这在生产环境里非常实用。

1.3 botmux 的工作流程

先看一下整体链路,下面的文本流程可以帮你理解每一步发生了什么。

Agent 发起任务 -> 遇到授权点 -> 调用 botmux /v1/notify -> botmux 生成 request_id 并保存状态 -> 调用飞书 API 发送消息卡片 -> 用户手机端收到卡片 -> 用户点击“允许”或“拒绝” -> 飞书把按钮回调发送给 botmux -> botmux 校验签名并更新授权状态 -> botmux 调用 Agent 的 callback_url -> Agent 根据结果继续执行或终止任务

这个流程看起来简单,但真正落地时还有几个关键点。第一个是超时处理,授权卡片不能永远有效,所以 botmux 要保存过期时间,Agent 也要有超时轮询机制。第二个是回调安全,飞书发来的 webhook 请求必须校验 token,否则任何人都可以伪造按钮回调。第三个是状态存储,生产环境不能用进程内存保存授权状态,必须用 Redis 这类外部存储,否则 botmux 重启一次,所有待授权的任务就全部失效了。

2. 环境准备与系统设计

2.1 技术选型与版本说明

本文的示例代码使用 Python 编写,你不需要有很深 Python 基础,只要会基本的异步编程就能跟上。具体环境如下:

  • 操作系统:Linux / macOS / Windows 均可,推荐 Linux 服务器作为 botmux 部署环境。
  • Python:3.9 及以上版本,示例代码使用 3.10 语法。
  • Web 框架:FastAPI。
  • HTTP 客户端:httpx。
  • 可选组件:Redis,用于生产环境状态存储。
  • Agent 服务:示例中用 FastAPI 模拟了一个 Agent,本身也没有额外依赖。

版本需要根据你的实际项目调整,本文重点是演示配置和代码思路,不是绑定某一套固定版本。飞书开放平台的接口字段偶有更新,你在配置时一定要以飞书官方最新文档为准,尤其是应用权限和事件订阅的入口名称。

2.2 系统结构

botmux 和 Agent 建议分开部署,即便在同一个服务器上,也尽量使用不同端口。这样可以避免相互影响,也方便以后独立扩容。

项目目录建议这样组织:

botmux-demo/ ├── agent/ │ └── agent_server.py # 模拟 Agent,等待授权并继续执行 ├── app/ │ ├── __init__.py │ └── main.py # botmux 主服务,FastAPI 入口 └── requirements.txt

在这个结构里,app/main.py是 botmux 的核心,负责接收飞书 webhook 和 Agent 的授权通知;agent/agent_server.py是业务侧,负责启动任务、调用 botmux、等待授权结果。实际项目中,Agent 可能是 n8n、Dify、AutoGen 或自研的工作流引擎,但接入 botmux 的方式大同小异,核心都是“需要确认时调 botmux,botmux 回传后再继续”。

依赖文件requirements.txt内容如下:

fastapi uvicorn httpx

生产环境再额外加一个redis和一个asyncio-redis相关库,演示阶段不需要。

2.3 飞书应用配置要点

要让 botmux 能发消息、收回调,你得先有一个飞书自建应用。打开飞书开放平台,在开发者后台创建一个企业自建应用,然后做以下几件事。

第一,开启“机器人”能力。这一步通常在应用功能里能找到,开启后应用会自动获得一个与 App ID 绑定的机器人,用户可以在飞书中搜索到它。

第二,配置“事件订阅”。如果你想接收用户发给机器人的消息,需要订阅im.message.receive_v1事件。订阅后,飞书会把用户消息推送到你配置的请求地址,也就是 botmux 的/webhook/feishu路由。

第三,配置“卡片回调”。这个和事件订阅是两套机制,用户点击卡片上的按钮后,飞书会把回调发送到你配置的卡片回调地址。示例代码里对应/webhook/feishu/card

第四,获取三个关键配置项:App ID、App Secret、Verification Token。App ID 和 App Secret 在“凭证与基础信息”里获取,用于换取 tenant_access_token;Verification Token 在“事件订阅”里配置,用于校验飞书发来的请求是否合法。

如果你的服务器没有公网 IP,飞书无法直接访问你的本地服务,可以借助内网穿透工具,把本地的 8000 端口映射到一个公网临时域名。要注意,内网穿透工具只适合开发测试,生产环境一定要把 botmux 部署到具备稳定公网地址的服务器上。

3. 搭建 botmux 消息网关

3.1 创建项目基础代码

先安装依赖。在项目根目录执行:

pip install -r requirements.txt

然后先写一个空的app/__init__.py文件,接着创建app/main.py。botmux 的代码不会太复杂,但我会把关键部分拆开讲解。

3.2 飞书鉴权与消息发送

botmux 要主动给用户发送飞书卡片,必须获取一个tenant_access_token。这个 token 是应用访问飞书开放平台 API 的通行证,有效期通常为 2 小时,需要在有效期内缓存和复用。

# 文件路径:app/main.py import os import time import uuid import json import httpx from typing import Dict from fastapi import FastAPI, Request from pydantic import BaseModel app = FastAPI(title="botmux") # 内存存储,生产环境请替换为 Redis AUTH_STORE: Dict[str, dict] = {} # 飞书配置,建议通过环境变量注入 FEISHU_APP_ID = os.getenv("FEISHU_APP_ID", "") FEISHU_APP_SECRET = os.getenv("FEISHU_APP_SECRET", "") FEISHU_VERIFICATION_TOKEN = os.getenv("FEISHU_VERIFICATION_TOKEN", "") FEISHU_RECEIVE_ID = os.getenv("FEISHU_RECEIVE_ID", "") # token 缓存 _token_cache = { "expire_at": 0, "token": "" } async def get_tenant_access_token() -> str: """获取飞书 tenant_access_token,并做简单缓存。""" if _token_cache["expire_at"] > time.time(): return _token_cache["token"] 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": FEISHU_APP_ID, "app_secret": FEISHU_APP_SECRET } ) data = resp.json() if data.get("code") != 0: raise RuntimeError(f"获取 tenant_access_token 失败: {data}") token = data["tenant_access_token"] # 提前 1 分钟过期,避免边界问题 _token_cache["token"] = token _token_cache["expire_at"] = time.time() + data["expire"] - 60 return token

上面的代码里,_token_cache是一个模块级变量,只适合单进程运行。如果你用 uvicorn 启动多个 worker,token 缓存会被每个 worker 各存一份,问题不大,因为最坏情况只是多换取几次 token。但如果你使用多个副本部署,建议把 token 缓存也放到 Redis 里。

发送消息卡片的函数如下:

async def send_feishu_card(receive_id: str, title: str, description: str, request_id: str): """向指定接收者发送一张包含“允许/拒绝”按钮的消息卡片。""" token = await get_tenant_access_token() card = { "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": title} }, "elements": [ { "tag": "markdown", "content": description }, { "tag": "action", "actions": [ { "tag": "button", "text": {"tag": "plain_text", "content": "✅ 允许"}, "type": "primary", "value": {"request_id": request_id, "action": "approve"} }, { "tag": "button", "text": {"tag": "plain_text", "content": "⛔ 拒绝"}, "type": "danger", "value": {"request_id": request_id, "action": "reject"} } ] } ] } payload = { "receive_id": receive_id, "msg_type": "interactive", "content": json.dumps(card, ensure_ascii=False) } headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json; charset=utf-8" } async with httpx.AsyncClient() as client: resp = await client.post( "https://open.feishu.cn/open-apis/im/v1/messages", params={"receive_id_type": "open_id"}, json=payload, headers=headers ) result = resp.json() if result.get("code") != 0: raise RuntimeError(f"发送飞书卡片失败: {result}")

注意,这里的FEISHU_RECEIVE_ID是接收人的 open_id。你可以通过飞书后台的“用户 ID 查询”功能获取,也可以在用户第一次给机器人发消息时,从消息事件里拿到sender.sender_id.open_id后存储起来。示例里为了简化,直接通过环境变量配置目标用户。

3.3 授权通知接口

Agent 在需要授权时,会向 botmux 发起一个POST /v1/notify请求。这个接口接收 Agent 传来的业务信息,生成唯一的request_id,保存授权记录,然后调用飞书卡片发送函数。

class NotifyRequest(BaseModel): request_id: str = "" title: str description: str = "" agent_callback_url: str expire_seconds: int = 300 @app.post("/v1/notify") async def notify(req: NotifyRequest): """Agent 调用该接口,发起一个授权请求。""" request_id = req.request_id or str(uuid.uuid4()) expire_at = time.time() + req.expire_seconds AUTH_STORE[request_id] = { "title": req.title, "description": req.description, "agent_callback_url": req.agent_callback_url, "expire_at": expire_at, "status": "pending" } try: await send_feishu_card( receive_id=FEISHU_RECEIVE_ID, title=req.title, description=req.description, request_id=request_id ) except Exception as e: # 发送失败时清理状态,避免产生脏数据 AUTH_STORE.pop(request_id, None) return {"code": 1, "msg": str(e)} return {"code": 0, "data": {"request_id": request_id, "expire_at": expire_at}}

这里有几个设计决策需要解释。request_id允许 Agent 传入,也可以由 botmux 生成。如果 Agent 自己的工作流里已经有一个任务 ID,建议直接复用,这样关联日志时更容易对应。expire_seconds是授权请求的过期时间,默认 300 秒,Agent 可以根据业务场景自行调整,比如删除数据的操作可以要求 60 秒内授权,发邮件可以放宽到 10 分钟。

3.4 飞书事件与卡片回调

接下来是飞书 webhook 入口。第一个入口接收事件订阅,需要处理飞书的url_verification验证,第二个入口接收卡片按钮回调。

@app.post("/webhook/feishu") async def feishu_event_webhook(request: Request): """飞书事件订阅回调。""" data = await request.json() # 首次配置事件订阅时,飞书会发送 url_verification 请求 if data.get("type") == "url_verification": return {"challenge": data.get("challenge")} # 校验 Verification Token if data.get("token") != FEISHU_VERIFICATION_TOKEN: return {"code": 1, "msg": "invalid token"} # 这里只处理用户发给机器人的消息事件,便于后续获取 open_id event = data.get("event", {}) if event.get("type") == "im.message.receive_v1": sender = event.get("sender", {}) sender_id = sender.get("sender_id", {}).get("open_id") if sender_id: print(f"收到用户 open_id: {sender_id}") return {"code": 0, "msg": "success"}

卡片回调入口稍微不一样,飞书在用户点击按钮后会发送一段 JSON,里面包含action字段,我们需要读取里面的value,然后根据用户选择调用 Agent 的回调接口。

@app.post("/webhook/feishu/card") async def feishu_card_webhook(request: Request): """飞书消息卡片按钮回调。""" data = await request.json() # 飞书卡片回调和事件订阅的校验方式不同,生产环境建议校验签名 action = data.get("action", {}) value = action.get("value", {}) request_id = value.get("request_id") decision = value.get("action") open_id = data.get("operator", {}).get("open_id") if not request_id or decision not in ("approve", "reject"): return {"code": 1, "msg": "invalid request"} auth_info = AUTH_STORE.get(request_id) if not auth_info: return {"code": 1, "msg": "request not found"} if auth_info["expire_at"] < time.time(): return {"code": 1, "msg": "request expired"} # 更新授权状态 auth_info["status"] = "approved" if decision == "approve" else "rejected" auth_info["operator"] = open_id auth_info["handled_at"] = time.time() # 回调 Agent callback_payload = { "request_id": request_id, "approved": decision == "approve", "operator": open_id, "handled_at": auth_info["handled_at"] } try: async with httpx.AsyncClient() as client: await client.post(auth_info["agent_callback_url"], json=callback_payload, timeout=5) except Exception as e: print(f"回调 Agent 失败: {e}") # 给飞书一个明确的成功响应,否则飞书会认为回调失败并重试 return {"code": 0, "msg": "success"}

飞书卡片回调如果要启用签名校验,通常需要从请求头里读取X-Lark-Signature等字段,用 Encrypt Key 做 HMAC-SHA256 签名。示例代码为了便于理解,只做了业务字段校验。生产环境一定不能省掉签名校验,否则攻击者可以伪造一个“允许”回调,让 Agent 执行危险操作。

4. 让 Agent 接入 botmux

4.1 Agent 侧授权状态存储

现在我们来模拟 Agent 服务。实际项目中 Agent 可能是一个复杂的执行引擎,但接入 botmux 的核心逻辑很简单,只有两步:启动任务时先请求授权,授权通过后再继续执行。为了演示,我写一个独立进程,端口设为 8080。

# 文件路径:agent/agent_server.py import asyncio import uuid import httpx from fastapi import FastAPI, Request app = FastAPI(title="demo-agent") # 保存授权结果,key 是 request_id AUTH_RESULTS = {} # botmux 服务地址 BOTMUX_BASE_URL = "http://localhost:8000" @app.post("/agent/start") async def start_task(): """启动一个模拟任务,任务会在授权通过后继续执行。""" request_id = str(uuid.uuid4()) AUTH_RESULTS[request_id] = None # 使用后台任务方式执行 asyncio.create_task(run_task(request_id)) return {"request_id": request_id, "status": "started"} async def run_task(request_id: str): """模拟 Agent 执行流程。""" print(f"任务 {request_id} 启动,先执行一些普通逻辑...") await asyncio.sleep(2) # 需要授权,调用 botmux print(f"任务 {request_id} 需要授权,通知 botmux...") async with httpx.AsyncClient() as client: await client.post(f"{BOTMUX_BASE_URL}/v1/notify", json={ "request_id": request_id, "title": "删除 3 条测试订单", "description": "订单号:20250101-001、20250101-002、20250101-003。该操作不可回滚。", "agent_callback_url": "http://localhost:8080/agent/authorize/callback", "expire_seconds": 120 }) # 等待授权结果,最多等 60 秒 for _ in range(60): result = AUTH_RESULTS.get(request_id) if result is not None: if result.get("approved"): print(f"任务 {request_id} 已获得授权,继续执行剩余步骤。") else: print(f"任务 {request_id} 被用户拒绝,任务终止。") return await asyncio.sleep(1) print(f"任务 {request_id} 授权超时,任务终止。")

这段代码里的AUTH_RESULTS只是一个内存字典,同样只适合演示。真实生产环境中,如果 Agent 是分布式部署,授权结果应该存 Redis,并通过 Redis 的键过期或发布订阅机制通知执行协程。还有一点需要注意,asyncio.create_task创建的后台任务在 FastAPI 的进程关闭时会立刻被取消,生产环境建议使用 Celery 或 Arq 这类任务队列框架来管理长任务。

4.2 Agent 接收授权结果

Agent 需要提供一个回调接口,botmux 会在这个接口里推送用户的授权结果。接口逻辑很简单,把结果写入 AUTH_RESULTS 即可。

@app.post("/agent/authorize/callback") async def authorize_callback(request: Request): """botmux 回调,告诉 Agent 用户点击了允许还是拒绝。""" data = await request.json() request_id = data.get("request_id") approved = data.get("approved") operator = data.get("operator") if request_id not in AUTH_RESULTS: return {"code": 1, "msg": "request not found"} AUTH_RESULTS[request_id] = { "approved": approved, "operator": operator, "handled_at": data.get("handled_at") } print(f"收到授权结果: request_id={request_id}, approved={approved}, operator={operator}") return {"code": 0, "msg": "success"}

到这里,Agent 接入 botmux 的最小流程就闭环了。Agent 启动任务、请求授权、等待结果、继续执行四个环节已经全部打通。实际业务中,你完全可以在run_task里把授权点替换成真实的危险操作,比如调用 SQL 删除语句之前检查一次授权结果。

4.3 与真实 Agent 框架对接思路

如果你用的是成熟的 Agent 框架,比如 LangChain、AutoGen、Dify,不要试图把 botmux 强塞进框架内部,而是把它封装成一个 Tool 或者大模型可以调用的工具函数。当 Agent 的计划里出现“执行删除”“发送外部请求”这类节点时,先调用一个名为request_authorization的工具,传入标题和描述,然后轮询结果。框架层面只需要保证这个工具会阻塞当前节点直到得到结果。

这样设计的好处是,你不需要改动框架的调度逻辑,只用一个标准工具就完成了人工确认。后续如果你想换成钉钉、企业微信,也只需要扩展 botmux 的消息渠道,Agent 侧完全不用动。

5. 运行一个完整示例

5.1 启动服务

完成以上代码后,在项目根目录打开两个终端。

终端一启动 botmux:

export FEISHU_APP_ID="你的 App ID" export FEISHU_APP_SECRET="你的 App Secret" export FEISHU_VERIFICATION_TOKEN="你的 Verification Token" export FEISHU_RECEIVE_ID="你的 open_id" uvicorn app.main:app --host 0.0.0.0 --port 8000

终端二启动 Agent:

uvicorn agent.agent_server:app --host 0.0.0.0 --port 8080

这里的FEISHU_RECEIVE_ID是目标用户的 open_id。如果还没有拿到,可以先不配置,先让飞书事件订阅保持开启,用户给机器人发一条消息,botmux 控制台会打印出 open_id,然后填进去重启服务。

5.2 在飞书中验证

启动成功后,发起一个授权请求。你可以直接调用 Agent 的接口来触发任务:

curl -X POST http://localhost:8080/agent/start

接口会返回类似下面的结果:

{ "request_id": "xxxx-xxxx-xxxx", "status": "started" }

大约 2 秒后,botmux 会调用飞书 API,向你的飞书私聊里推送一张消息卡片。卡片顶部是标题,中间是授权描述,底部有两个按钮:允许和拒绝。你可以在手机上打开飞书,点击“允许”。

点击之后,Agent 控制台会打印获得授权的日志,Botmux 控制台也会打印回调结果。这样一个完整的“移动端授权”流程就成功了。

5.3 预期日志与结果

botmux 控制台预期输出:

收到授权结果: request_id=xxxx-xxxx-xxxx, approved=True, operator=ou_xxxx

Agent 控制台预期输出:

任务 xxxx-xxxx-xxxx 需要授权,通知 botmux... 任务 xxxx-xxxx-xxxx 已获得授权,继续执行剩余步骤。

如果你点击“拒绝”,Agent 会输出任务终止的日志。如果等待 120 秒不点击,botmux 里的授权状态会过期,Agent 等待 60 秒后也会超时终止。

6. 常见问题与排查思路

6.1 飞书事件订阅和回调不通

这是最常见的坑。飞书事件订阅配置 URL 后,一直提示验证失败。通常原因有几个:第一,你开启了 Encrypt Key,但代码里没有实现解密,飞书发送的是加密 JSON,导致你拿不到明文 token 和 challenge。建议开发阶段先不开启 Encrypt Key,跑通后再补上解密逻辑。第二,Verification Token 配错了,注意飞书事件订阅里显示的 token 和应用凭证里的 App Secret 不是一回事,代码里要读取前者。第三,公网地址没有正确穿透,飞书服务器访问不到你的回调地址。

我建议遇到问题时先手动模拟一遍回调。用 curl 往/webhook/feishu发送一个{"type": "url_verification", "challenge": "test"},确认返回里带challenge。再发送一个带错误 token 的请求,确认服务能识别出来。这样能把问题范围锁定在飞书侧还是代码侧。

6.2 授权卡片不展示

如果POST /v1/notify返回成功,但手机端没有收到卡片,优先检查FEISHU_RECEIVE_ID是否正确。Recevice ID 必须是 open_id,不能是手机号或邮箱。还要确认应用是否有im:message发送权限,以及应用是否已经发布。自建应用不发布版本时,机器人只能给应用创建者发送消息,其他用户收不到。

另一种可能是代码里发送卡片用的消息 API 接口字段出问题。如果返回码不是 0,把飞书返回的msg打出来,直接按错误信息处理。

6.3 Agent 执行 Provider 超时

在接入过程中,有人会遇到类似the agent execution provider did not respond in time的报错。这个报错从字面上看是 Agent 的某个执行 Provider 没有及时响应,可能由模型 API 网络抖动、并发数打满或超时时间设置过短导致。

这时要先判断是授权链路问题还是模型调用问题。从日志看位置:如果报错发生在授权回调之后,说明 Agent 已经拿到授权,正准备执行下一步模型调用;如果报错发生在授权之前,说明 Agent 根本还没走到授权点。排查时先用一个简单的测试任务绕过授权,直接调用模型 API,确认模型接口本身是否稳定。如果模型接口稳定,再检查 Agent 框架的 Provider 超时配置,适当增大timeout值,并加上重试逻辑。

6.4 授权状态与安全排查

下表整理了几类常见的授权异常场景,你可以对照排查:

问题现象常见原因解决思路
用户已点击按钮,但 Agent 一直等待Agent 回调地址不可达用 curl 测试 callback_url,检查 Agent 服务日志
同一卡片被点击多次,Agent 执行多次botmux 没有做幂等处理在 AUTH_STORE 中检查 status,已处理过的请求直接忽略
授权记录丢失,Agent 超时botmux 重启导致内存数据丢失使用 Redis 存储,启动后自动恢复
飞书回调数据被篡改缺少签名校验配置 Encrypt Key,实现校验逻辑
用户误点了允许描述信息不够明确在卡片中加入操作影响范围和不可回滚的警告

7. 工程化与最佳实践

7.1 存储设计:从内存到 Redis

示例代码中的内存字典在演示阶段没问题,但生产环境绝对不能直接用。botmux 一旦重启,所有 pending 状态的授权请求都会丢失,Agent 那边就会一直傻等到超时。推荐把 AUTH_STORE 换成 Redis Hash 结构,key 是botmux:auth:{request_id},字段包括 title、description、callback_url、expire_at、status。

读状态时顺便判断expire_at,这样不需要额外定时任务清理过期数据。还可以让 Agent 侧轮询 Redis,而不是等 HTTP 回调。这样即使回调网络失败,Agent 也能通过轮询拿到最终结果。

7.2 安全与审计

移动端授权本质上是给 Agent 开了一个“人不在电脑前也能执行危险操作”的窗口,所以安全设计尤其重要。

第一,飞书 webhook 的请求校验必须完善。事件订阅要校验 Verification Token,卡片回调要做签名校验。第二,授权范围要尽可能小。比如删除订单的授权请求里,明确标识订单号,而不是只写“是否继续执行”。第三,所有授权操作必须有审计日志。最简单的方式是把每次授权请求、用户点击、结果回传都记录到飞书多维表格,作为审计留痕。多维表格里建三列:request_id、操作内容、授权结果,配上执行时间,排查问题时非常方便。

第四,给 botmux 的/v1/notify接口加一个内部调用凭证,防止任何能访问到该端口的外部服务都来调它。最简单的方法是在 Headers 里带一个X-BOTMUX-TOKEN,botmux 校验通过后才接受请求。

7.3 多 Agent 扩展

botmux 的名字叫“多路复用器”,说明它天然适合同时服务多个 Agent。假设你同时跑着数据分析 Agent、自动化测试 Agent、内容生成 Agent,每个 Agent 都可能有授权需求,你需要考虑两个问题。

第一是消息隔离。不同 Agent 的授权请求,在飞书卡片上可以带上agent_name标签,让用户一眼看出是哪个 Agent 发出的。第二是回调地址隔离。每个 Agent 的agent_callback_url不同,botmux 只需要按记录存好,回调时对应着发就行,不需要关心 Agent 的业务细节。这样新增一个 Agent 时,只需要告诉它 botmux 的地址,不需要改造 botmux。

7.4 与飞书多维表格联动

刚才提到的审计日志,用飞书多维表格承载很合适。botmux 在授权状态更新后,调用多维表格的 API,把记录写入表格。你也可以把多维表格当作 Agent 的“人工审批中心”,通过新增记录来发起任务,通过修改记录状态让 Agent 感知任务变化。这样整个链路就从“飞书卡片 -> botmux -> Agent”扩展成了数据闭环。

当然,我不建议在演示阶段就把所有能力都接进来,先跑通最小闭环,再逐步加审计、加多维表格、加多 Agent 管理,这样排查问题时心态会更稳。

8. 总结与后续学习建议

这套 botmux 方案解决的核心问题是:AI Agent 在无人值守场景下,如何通过移动端拿到人工授权。文章里包含了 botmux 的接口设计、飞书机器人配置、卡片消息发送、回调处理和 Agent 接入示例,你可以直接照着搭建一个最小可运行系统。代码是简化的,生产环境至少要补上签名校验、Redis 存储和服务鉴权三个点。

建议你先从一个真实的小场景入手,比如让 Agent 调用一个删除接口前请求确认。跑通整个流程后,再考虑把它接入到现有的 Agent 框架里。当你习惯了在手机上点一下“允许”,Agent 就能继续往下跑的时候,你会发现再也不想回到老是回电脑点授权的日子了。

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

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

立即咨询