> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt # Agent 审批 Agent 在调用处理敏感操作(例如删除资源或运行长时间进程)的 Tool 时,有时也需要 Workflow 中所使用的[人工介入](https://mastra.zisheng.pro/docs/workflows/human-in-the-loop)监督。通过 Agent 审批,你可以在 Tool 调用执行前将其挂起,以便由人工批准或拒绝;也可以让 Tool 自行挂起,向用户请求更多上下文。 ## 何时使用 Agent 审批 - **破坏性或不可逆操作**,例如删除记录、发送电子邮件或处理付款。 - **成本高昂的操作**,例如调用昂贵的第三方 API,并且希望先验证参数。 - **条件式确认**:Tool 开始执行后发现,需要用户确认或提供更多数据才能完成。 ## 快速入门 为 Tool 标记 `requireApproval: true`,然后检查流中的 `tool-call-approval` 分块并进行批准或拒绝: ```typescript import { Agent } from '@mastra/core/agent' import { createTool } from '@mastra/core/tools' import { z } from 'zod' const deleteTool = createTool({ id: 'delete-record', description: 'Delete a record by ID', inputSchema: z.object({ id: z.string() }), outputSchema: z.object({ deleted: z.boolean() }), requireApproval: true, execute: async ({ id }) => { await db.delete(id) return { deleted: true } }, }) const agent = new Agent({ id: 'my-agent', name: 'My Agent', model: 'openai/gpt-5-mini', tools: { deleteTool }, }) const stream = await agent.stream('Delete record abc-123') for await (const chunk of stream.fullStream) { if (chunk.type === 'tool-call-approval') { const approved = await agent.approveToolCall({ runId: stream.runId }) for await (const c of approved.textStream) process.stdout.write(c) } } ``` > **备注:** Agent 审批使用快照捕获请求状态。请在 Mastra 实例上配置 [Storage Provider](https://mastra.zisheng.pro/docs/storage/overview),否则会出现 “snapshot not found” 错误。 > > Agent 运行快照是最小化的恢复制品:只保留恢复挂起运行所需的内容,并在运行结束后删除。使用 [Tracing](https://mastra.zisheng.pro/docs/observability/overview) 保存执行记录,使用 [Memory](https://mastra.zisheng.pro/docs/memory/overview) 保存对话历史记录。 ## 审批的工作原理 Mastra 提供两种不同的暂停 Tool 调用机制:**执行前审批**和**运行时挂起**。 ### 执行前审批 执行前审批会在 Tool 调用的 `execute` 函数运行\_之前\_将其暂停。LLM 仍会决定调用哪个 Tool 并提供参数,但在你明确批准前,`execute` 不会运行。 以下标志通过 OR 逻辑共同控制此行为。只要\_任意一个\_为 `true`,调用就会暂停: | 标志 | 设置位置 | 作用域 | | --------------------------- | ---------------------------- | --------------------- | | `requireToolApproval: true` | `stream()` / `generate()` 选项 | 暂停该请求的**每一次** Tool 调用 | | `requireApproval: true` | `createTool()` 定义 | 暂停对**该特定 Tool** 的调用 | 流会发出 `tool-call-approval` 分块,其中包含 `toolCallId`、`toolName` 和 `args`。调用 `approveToolCall()` 或 `declineToolCall()`,并传入流的 `runId` 以继续: ```typescript const stream = await agent.stream("What's the weather in London?", { requireToolApproval: true, }) for await (const chunk of stream.fullStream) { if (chunk.type === 'tool-call-approval') { console.log('Tool:', chunk.payload.toolName) console.log('Args:', chunk.payload.args) // Approve const approved = await agent.approveToolCall({ runId: stream.runId }) for await (const c of approved.textStream) process.stdout.write(c) // Or decline const declined = await agent.declineToolCall({ runId: stream.runId }) for await (const c of declined.textStream) process.stdout.write(c) } } ``` #### 使用函数进行条件审批 `requireToolApproval` 除了接受布尔值,还可以接受一个针对每次 Tool 调用做出判断的函数。该函数接收 `toolName`、模型传入的 `args`、`requestContext` 和 `workspace`。返回 `true` 要求批准该调用,返回 `false` 则允许调用。这样可以在运行时设置审批条件,例如仅审批名称与某种模式匹配的 Tool: ```typescript const stream = await agent.stream('Clean up old records', { requireToolApproval: ({ toolName }) => /^delete_/.test(toolName), }) ``` Tool 自身的 `requireApproval` 设置优先于上述函数,由 Tool 自身的规则决定是否需要审批。如果函数抛出错误,出于故障安全考虑,该调用将需要审批。 > **备注:** 基于函数的 `requireToolApproval` 仅适用于常规 `stream()` / `generate()` 调用。Durable Agent 和已存储的 Agent 会持久化选项,而函数无法序列化,因此它们只接受布尔值。如果在这些上下文中传入函数,系统会回退为要求审批每一次 Tool 调用。 #### 将审批绑定到确切的 Tool 参数 对于敏感 Tool,应将审批绑定到向审核者展示的确切 Tool 名称和参数。如果参数在执行前发生变化,不应继续沿用旧审批来运行 Tool。 `tool-call-approval` 分块已包含 `toolName`、`toolCallId` 和 `args`。你可以在展示审批请求时为这些字段生成指纹。以下示例使用 JSON 字符串作为指纹,但在生产环境中应使用 Tool 名称和参数的稳定哈希值: ```typescript import { Agent } from '@mastra/core/agent' // For your production usecase, build a stable hash of the tool name and args function actionFingerprint(toolName: string, args: unknown) { const payload = JSON.stringify({ toolName, args }) return `fingerprint-${payload}` } const sensitiveTools = new Set(['issue_refund', 'delete_record']) const approvedFingerprints = new Set() export const approvalBoundAgent = new Agent({ id: 'approval-bound-agent', name: 'Approval Bound Agent', model: 'openai/gpt-5.6-sol', tools: { issueRefundTool, deleteRecordTool }, hooks: { beforeToolCall: ({ toolName, input }) => { if (!sensitiveTools.has(toolName)) return const fingerprint = actionFingerprint(toolName, input) if (!approvedFingerprints.delete(fingerprint)) { return { proceed: false, output: `Tool call blocked: approval did not match ${toolName} arguments.`, } } }, }, }) ``` ```typescript const stream = await approvalBoundAgent.stream('Refund order ord-1042', { requireToolApproval: ({ toolName }) => sensitiveTools.has(toolName), }) async function consumeApprovalStream(currentStream: typeof stream) { for await (const chunk of currentStream.fullStream) { if (chunk.type === 'tool-call-approval') { const { toolName, toolCallId, args } = chunk.payload const fingerprint = actionFingerprint(toolName, args) // Present toolName, args, and fingerprint to your approval UI. const approved = await showApprovalDialog({ toolName, args, fingerprint }) const nextStream = approved ? await approveReviewedToolCall(currentStream.runId, toolCallId, fingerprint) : await approvalBoundAgent.declineToolCall({ runId: currentStream.runId, toolCallId }) await consumeApprovalStream(nextStream) } } } async function approveReviewedToolCall(runId: string, toolCallId: string, fingerprint: string) { approvedFingerprints.add(fingerprint) return approvalBoundAgent.approveToolCall({ runId, toolCallId }) } await consumeApprovalStream(stream) ``` 在生产环境中,应将批准的指纹存储在持久 Storage 中,并将其作用域限定为用户、运行、Tool 调用和策略版本。上面的 `Set` 特意保持简洁,以清晰展示边界:审批只会使用一次,并且仅用于审核过的同一组规范 Tool 参数。 ### 使用 `suspend()` 在运行时挂起 Tool 还可以在其 `execute` 函数执行\_期间\_调用 `suspend()` 进行暂停。当 Tool 开始运行后发现需要更多用户输入或确认才能完成时,这种方式非常有用。 流会发出 `tool-call-suspended` 分块,其中包含由 Tool 的 `suspendSchema` 定义的自定义 payload。要恢复运行,请调用 `resumeStream()`,并传入符合 Tool `resumeSchema` 的数据。 ```typescript const weatherTool = createTool({ id: 'get-weather', inputSchema: z.object({ location: z.string().optional(), }), suspendSchema: z.object({ question: z.string(), }), resumeSchema: z.object({ location: z.string(), }), execute: async ({ location }, context) => { if (!location) { return await context?.agent?.suspend({ question: 'Which city would you like the weather for?', }) } return await fetchWeather(location) }, }) ``` > **备注:** `suspend()` 不会抛出错误,调用后应立即返回(例如 `return await suspend({ ... })`)。在 Tool 暂停前,`await suspend(...)` 之后的代码仍会继续运行。 ## 在 `generate()` 中进行 Tool 审批 Tool 审批也适用于非流式的 `generate()` 用例。当 Tool 需要审批时,`generate()` 会立即返回,其中包含 `finishReason: 'suspended'`、`suspendPayload`(包含 Tool 调用详情 `toolCallId`、`toolName` 和 `args`),以及 `runId`: ```typescript const output = await agent.generate('Find user John', { requireToolApproval: true, }) if (output.finishReason === 'suspended') { console.log('Tool requires approval:', output.suspendPayload.toolName) // Approve const result = await agent.approveToolCallGenerate({ runId: output.runId, toolCallId: output.suspendPayload.toolCallId, }) console.log('Final result:', result.text) // Or decline const result = await agent.declineToolCallGenerate({ runId: output.runId, toolCallId: output.suspendPayload.toolCallId, }) } ``` ### Stream 与 generate 对比 | 方面 | `stream()` | `generate()` | | ---- | ---------------------------- | ------------------------------------------------ | | 响应类型 | 流式分块 | 完整响应 | | 审批检测 | `tool-call-approval` 分块 | `finishReason: 'suspended'` | | 批准方法 | `approveToolCall({ runId })` | `approveToolCallGenerate({ runId, toolCallId })` | | 拒绝方法 | `declineToolCall({ runId })` | `declineToolCallGenerate({ runId, toolCallId })` | | 结果 | 可迭代的流 | 完整输出对象 | > **备注:** `toolCallId` 在上述四个方法中均为可选参数。当可能存在多个待处理的 Tool 调用时(在 Supervisor Agent 中很常见),请传入此参数。如果省略,Agent 会恢复最近挂起的 Tool 调用。 ## Tool 级审批 你可以将单个 Tool 标记为需要审批,而不必在 Agent 级别暂停每一次 Tool 调用。这样可以实现细粒度控制:只有特定 Tool 会暂停,其他 Tool 则立即执行。 ### 使用 `requireApproval` 进行审批 在 Tool 定义中设置 `requireApproval: true`。无论是否在 Agent 上设置 `requireToolApproval`,该 Tool 都会在执行前暂停: ```typescript export const testTool = createTool({ id: 'test-tool', description: 'Fetches weather for a location', inputSchema: z.object({ location: z.string(), }), outputSchema: z.object({ weather: z.string(), }), resumeSchema: z.object({ approved: z.boolean(), }), execute: async inputData => { const response = await fetch(`https://wttr.in/${inputData.location}?format=3`) const weather = await response.text() return { weather } }, requireApproval: true, }) ``` 当 `requireApproval` 为 `true` 时,流会像 Agent 级审批一样发出 `tool-call-approval` 分块。使用 `approveToolCall()` 或 `declineToolCall()` 继续: ```typescript const stream = await agent.stream("What's the weather in London?") for await (const chunk of stream.fullStream) { if (chunk.type === 'tool-call-approval') { console.log('Approval required for:', chunk.payload.toolName) } } const handleApproval = async () => { const approvedStream = await agent.approveToolCall({ runId: stream.runId }) for await (const chunk of approvedStream.textStream) { process.stdout.write(chunk) } process.stdout.write('\n') } ``` ### 使用 `suspend()` 进行审批 这种方式中,Agent 和 Tool 都不使用 `requireApproval`,而是由 Tool 的 `execute` 函数调用 `suspend()` 在特定位置暂停,并向用户返回上下文或确认提示。当审批取决于运行时条件,而不是无条件进行时,这种方式很有用。 ```typescript export const testToolB = createTool({ id: 'test-tool-b', description: 'Fetches weather for a location', inputSchema: z.object({ location: z.string(), }), outputSchema: z.object({ weather: z.string(), }), resumeSchema: z.object({ approved: z.boolean(), }), suspendSchema: z.object({ reason: z.string(), }), execute: async (inputData, context) => { const { resumeData: { approved } = {}, suspend } = context?.agent ?? {} if (!approved) { return suspend?.({ reason: 'Approval required.' }) } const response = await fetch(`https://wttr.in/${inputData.location}?format=3`) const weather = await response.text() return { weather } }, }) ``` 采用这种方式时,流会包含一个 `tool-call-suspended` 分块,`suspendPayload` 中包含 `reason`,它由 Tool 的 `suspendSchema` 定义。调用 `resumeStream`,并传入符合 `resumeSchema` 的数据和 `runId` 以继续: ```typescript const stream = await agent.stream("What's the weather in London?") for await (const chunk of stream.fullStream) { if (chunk.type === 'tool-call-suspended') { console.log(chunk.payload.suspendPayload) } } const handleResume = async () => { const resumedStream = await agent.resumeStream({ approved: true }, { runId: stream.runId }) for await (const chunk of resumedStream.textStream) { process.stdout.write(chunk) } process.stdout.write('\n') } ``` ## 自动恢复 Tool 使用调用 `suspend()` 的 Tool 时,可以启用自动恢复,让 Agent 根据用户的下一条消息恢复挂起的 Tool。在 Agent 的默认选项中或针对单次请求将 `autoResumeSuspendedTools` 设置为 `true`: ```typescript import { Agent } from '@mastra/core/agent' import { Memory } from '@mastra/memory' const agent = new Agent({ id: 'my-agent', name: 'My Agent', instructions: 'You are a helpful assistant', model: 'openai/gpt-5-mini', tools: { weatherTool }, memory: new Memory(), defaultOptions: { autoResumeSuspendedTools: true, }, }) ``` 启用后,Agent 会在收到下一条用户消息时,从消息历史记录中检测挂起的 Tool。它会提取 `resumeData`,其结构基于 Tool 的 `resumeSchema`,然后自动恢复 Tool。 以下示例展示了完整的对话流程: ```typescript import { createTool } from '@mastra/core/tools' import { z } from 'zod' const weatherTool = createTool({ id: 'weather-tool', description: 'Fetches weather for a city', inputSchema: z.object({ city: z.string(), }), outputSchema: z.object({ weather: z.string(), }), suspendSchema: z.object({ message: z.string(), }), resumeSchema: z.object({ city: z.string(), }), execute: async (inputData, context) => { const { resumeData, suspend } = context?.agent ?? {} // If no city provided, ask the user if (!inputData.city && !resumeData?.city) { return suspend?.({ message: 'What city do you want to know the weather for?' }) } const city = resumeData?.city ?? inputData.city const response = await fetch(`https://wttr.in/${city}?format=3`) const weather = await response.text() return { weather: `${city}: ${weather}` } }, }) ``` ```typescript const stream = await agent.stream("What's the weather like?") for await (const chunk of stream.fullStream) { if (chunk.type === 'tool-call-suspended') { console.log(chunk.payload.suspendPayload) } } // User sends follow-up on the same thread const resumedStream = await agent.stream('San Francisco') for await (const chunk of resumedStream.textStream) { process.stdout.write(chunk) } ``` ```text User: "What's the weather like?" Agent: "What city do you want to know the weather for?" User: "San Francisco" Agent: "The weather in San Francisco is: San Francisco: ☀️ +72°F" ``` 第二条消息会自动恢复挂起的 Tool。Agent 从用户消息中提取 `{ city: "San Francisco" }`,并将其作为 `resumeData` 传入。 ### 要求 要使 Tool 自动恢复正常工作,需要满足以下条件: - **已配置 Memory**:Agent 需要使用 Memory 跨消息跟踪挂起的 Tool - **相同 thread**:后续消息必须使用相同的 Memory thread 和 resource 标识符 - **已定义 `resumeSchema`**:Tool 必须定义 `resumeSchema`,让 Agent 知道要从用户消息中提取怎样的数据结构 ### 手动恢复与自动恢复 | 方式 | 用例 | | ------------------------------ | ------------------------ | | 手动(`resumeStream()`) | 编程式控制、Webhook、点击按钮、外部触发器 | | 自动(`autoResumeSuspendedTools`) | 用户以自然语言提供恢复数据的对话流程 | 两种方式可使用相同的 Tool 定义。只有当消息历史记录中存在挂起的 Tool,并且用户在相同 thread 上发送新消息时,才会触发自动恢复。 ## 重启后恢复 上述示例会在挂起和审批之间一直保留 `stream.runId`。只要进程保持运行,这种方式就有效;但在生产环境中,审批通常会在页面刷新、Server 重启后才到达,或由负载均衡器后的另一个 Server 实例处理。 使用 [`listSuspendedRuns()`](https://mastra.zisheng.pro/reference/agents/listSuspendedRuns) 可以从 Storage 中重新发现某个对话的待处理运行: ```typescript // In the request handler that receives the user's decision const { runs } = await agent.listSuspendedRuns({ threadId: 'thread-123', resourceId: 'user-456', }) const run = runs[0] const toolCall = run?.toolCalls[0] if (run && toolCall) { let stream if (toolCall.requiresApproval) { // Suspended by requireApproval — approve or decline the tool call stream = await agent.approveToolCall({ runId: run.runId, toolCallId: toolCall.toolCallId }) } else { // Suspended by suspend() — resume with the data the tool asked for console.log('Tool asked:', toolCall.suspendPayload) stream = await agent.resumeStream({ name: 'San Francisco' }, { runId: run.runId }) } for await (const chunk of stream.textStream) process.stdout.write(chunk) } ``` 返回的每个运行都包含挂起的 Tool 调用(`toolCallId`、`toolName`、`args` 和 `requiresApproval`)。对于审批挂起(`requiresApproval: true`),使用 `approveToolCall()` / `declineToolCall()` 进行响应;对于基于 `suspend()` 的挂起,响应会携带相应的 `suspendPayload`,并需要使用恢复数据调用 `resumeStream()`。因此,无需在内存中保留任何状态,即可为任一流程重新构建正确的 UI。 `sendToolApproval()` 会自动使用相同的、由 Storage 支持的发现机制:如果未在内存中找到该 thread 的活跃运行,它会先在 Storage 中查找挂起的运行,然后才会失败。如果多个挂起的运行与该 thread 匹配,请传入 `toolCallId` 以消除歧义。 同样的发现功能也可通过 HTTP 上的 `GET /agents/:agentId/suspended-runs` 和客户端 SDK 中的 [`agent.listSuspendedRuns()`](https://mastra.zisheng.pro/reference/client-js/agents) 使用,因此基于浏览器的审批 UI 可以直接重新发现待处理的运行。 > **备注:** 只有在 Mastra 实例配置了持久化 [Storage Provider](https://mastra.zisheng.pro/docs/storage/overview) 时,挂起的运行才能在重启后继续存在。默认的内存存储会在进程退出时丢失快照。 ## Tool 审批:Supervisor Agent [Supervisor Agent](https://mastra.zisheng.pro/docs/capabilities/subagents) 使用 `.stream()` 或 `.generate()` 协调多个子 Agent。当某个子 Agent 调用需要审批的 Tool 时,请求会沿委派链向上传播,并出现在 Supervisor 层: 1. Supervisor 将任务委派给子 Agent。 2. 子 Agent 调用设置了 `requireApproval: true` 或使用 `suspend()` 的 Tool。 3. 审批请求向上传播至 Supervisor。 4. 你在 Supervisor 层批准或拒绝。 5. 决定向下传播回子 Agent。 Tool 审批也可以跨多层委派传播。如果 Supervisor 将任务委派给子 Agent A,而子 Agent A 又将任务委派给子 Agent B,后者有一个设置了 `requireApproval: true` 的 Tool,审批请求仍会出现在顶层 Supervisor。 ### 在 Supervisor Agent 中批准和拒绝 以下示例创建了一个子 Agent,其中有一个需要审批的 Tool。当该 Tool 触发审批请求时,请求会作为 `tool-call-approval` 分块出现在 Supervisor 的流中: ```typescript import { Agent } from '@mastra/core/agent' import { createTool } from '@mastra/core/tools' import { Memory } from '@mastra/memory' import { z } from 'zod' const findUserTool = createTool({ id: 'find-user', description: 'Finds user by ID in the database', inputSchema: z.object({ userId: z.string(), }), outputSchema: z.object({ user: z.object({ id: z.string(), name: z.string(), email: z.string(), }), }), requireApproval: true, execute: async input => { const user = await database.findUser(input.userId) return { user } }, }) const dataAgent = new Agent({ id: 'data-agent', name: 'Data Agent', description: 'Handles database queries and user data retrieval', model: 'openai/gpt-5-mini', tools: { findUserTool }, }) const supervisorAgent = new Agent({ id: 'supervisor', name: 'Supervisor Agent', instructions: `You coordinate data retrieval tasks. Delegate to data-agent for user lookups.`, model: 'openai/gpt-5.6-sol', agents: { dataAgent }, memory: new Memory(), }) const stream = await supervisorAgent.stream('Find user with ID 12345') for await (const chunk of stream.fullStream) { if (chunk.type === 'tool-call-approval') { console.log('Tool requires approval:', chunk.payload.toolName) console.log('Arguments:', chunk.payload.args) // Approve the tool call const resumeStream = await supervisorAgent.approveToolCall({ runId: stream.runId, toolCallId: chunk.payload.toolCallId, }) for await (const resumeChunk of resumeStream.textStream) { process.stdout.write(resumeChunk) } // To decline instead, use: const declineStream = await supervisorAgent.declineToolCall({ runId: stream.runId, toolCallId: chunk.payload.toolCallId, }) } } ``` ### 在 Supervisor Agent 中使用 `suspend()` Tool 也可以使用 [`suspend()`](#approval-using-suspend) 暂停执行,并向用户返回上下文。这种方式与 `requireApproval` 一样,能够穿过 Supervisor 委派链:挂起状态会出现在 Supervisor 层。 ```typescript const conditionalTool = createTool({ id: 'conditional-operation', description: 'Performs an operation that may require confirmation', inputSchema: z.object({ operation: z.string(), }), suspendSchema: z.object({ message: z.string(), }), resumeSchema: z.object({ confirmed: z.boolean(), }), execute: async (input, context) => { const { resumeData } = context?.agent ?? {} if (!resumeData?.confirmed) { return context?.agent?.suspend({ message: `Confirm: ${input.operation}?`, }) } // Proceed with operation return await performOperation(input.operation) }, }) ``` ```typescript // When using this tool through a subagent in supervisor agents for await (const chunk of stream.fullStream) { if (chunk.type === 'tool-call-suspended') { console.log('Tool suspended:', chunk.payload.suspendPayload.message) // Resume with confirmation const resumeStream = await supervisorAgent.resumeStream( { confirmed: true }, { runId: stream.runId }, ) for await (const resumeChunk of resumeStream.textStream) { process.stdout.write(resumeChunk) } } } ``` ### 在 Supervisor 中使用 `generate()` 进行审批 Tool 审批也可以通过 Supervisor Agent 中的 `generate()` 进行传播: ```typescript const output = await supervisorAgent.generate('Find user with ID 12345', { maxSteps: 10, }) if (output.finishReason === 'suspended') { console.log('Tool requires approval:', output.suspendPayload.toolName) // Approve const result = await supervisorAgent.approveToolCallGenerate({ runId: output.runId, toolCallId: output.suspendPayload.toolCallId, }) console.log('Final result:', result.text) } ``` ## 相关内容 - [Tool](https://mastra.zisheng.pro/docs/agents/using-tools) - [Agent 概览](https://mastra.zisheng.pro/docs/agents/overview) - [MCP 概览](https://mastra.zisheng.pro/docs/mcp/overview) - [Memory](https://mastra.zisheng.pro/docs/memory/overview) - [请求上下文](https://mastra.zisheng.pro/docs/server/request-context)