人机协同模式

人机协同（HITL）模式在 Agent 不同层添加审批或输入。你可以响应 MCP 服务器请求、在持久 Workflow 中挂起应用工作，或在模型生成代码调用工具之前审批连接器调用。

为何需要 human-in-the-loop？

合规：法规可能要求对某些操作进行人工审批
安全：高风险操作（支付、删除、外部通信）需要监督
质量：人工审核可发现 AI 可能遗漏的错误
信任：用户能够审批关键操作时更有信心

常见用例

常见用途包括财务审批、内容审核、批量数据操作、有副作用的 tool 调用前审批，以及访问控制变更。

选择模式

根据谁引入暂停及发生在何处选择模式：

模式	审批层	发起方	典型等待时间	关键 API
MCP elicitation	由 Agent 客户端处理的 MCP 请求	MCP 服务器开发者	数分钟	`configureElicitationHandlers()`
Workflow approval	持久应用任务或工具操作	Agent 应用开发者	数月或数年	`waitForApproval()`
Code Mode approval	模型生成代码中的连接器调用	Code Mode Agent 开发者	直至配置的过期时间	`requiresApproval`、`approve()`、`reject()`

基于 Workflow 的审批

当应用需要挂起任务或工具操作直至审批完成时，使用 Cloudflare Workflows。waitForApproval() 创建由 Cloudflare Workflows 支撑的持久门控，等待可持续数月或更久而无需保持 Agent 运行。

基本模式

import { Agent } from "agents";
import { AgentWorkflow } from "agents/workflows";

export class ExpenseWorkflow extends AgentWorkflow {
	async run(event, step) {
		const expense = event.payload;

		// Step 1: Validate the expense
		const validated = await step.do("validate", async () => {
			if (expense.amount <= 0) {
				throw new Error("Invalid expense amount");
			}
			return { ...expense, validatedAt: Date.now() };
		});

		// Step 2: Report that we are waiting for approval
		await this.reportProgress({
			step: "approval",
			status: "pending",
			message: `Awaiting approval for $${expense.amount}`,
		});

		// Step 3: Wait for human approval (pauses the workflow)
		const approval = await this.waitForApproval(step, {
			timeout: "7 days",
		});

		console.log(`Approved by: ${approval?.approvedBy}`);

		// Step 4: Process the approved expense
		const result = await step.do("process", async () => {
			return { expenseId: crypto.randomUUID(), ...validated };
		});

		await step.reportComplete(result);
		return result;
	}
}

import { Agent } from "agents";
import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";

type ExpenseParams = {
	amount: number;
	description: string;
	requestedBy: string;
};

export class ExpenseWorkflow extends AgentWorkflow<
	ExpenseAgent,
	ExpenseParams
> {
	async run(event: AgentWorkflowEvent<ExpenseParams>, step: AgentWorkflowStep) {
		const expense = event.payload;

		// Step 1: Validate the expense
		const validated = await step.do("validate", async () => {
			if (expense.amount <= 0) {
				throw new Error("Invalid expense amount");
			}
			return { ...expense, validatedAt: Date.now() };
		});

		// Step 2: Report that we are waiting for approval
		await this.reportProgress({
			step: "approval",
			status: "pending",
			message: `Awaiting approval for $${expense.amount}`,
		});

		// Step 3: Wait for human approval (pauses the workflow)
		const approval = await this.waitForApproval<{ approvedBy: string }>(step, {
			timeout: "7 days",
		});

		console.log(`Approved by: ${approval?.approvedBy}`);

		// Step 4: Process the approved expense
		const result = await step.do("process", async () => {
			return { expenseId: crypto.randomUUID(), ...validated };
		});

		await step.reportComplete(result);
		return result;
	}
}

用于审批的 Agent 方法

Agent 提供用于审批或拒绝等待中 workflow 的方法：

import { Agent, callable } from "agents";

export class ExpenseAgent extends Agent {
	initialState = {
		pendingApprovals: [],
	};

	// Approve a waiting workflow
	@callable()
	async approve(workflowId, approvedBy) {
		await this.approveWorkflow(workflowId, {
			reason: "Expense approved",
			metadata: { approvedBy, approvedAt: Date.now() },
		});

		// Update state to reflect approval
		this.setState({
			...this.state,
			pendingApprovals: this.state.pendingApprovals.filter(
				(p) => p.workflowId !== workflowId,
			),
		});
	}

	// Reject a waiting workflow
	@callable()
	async reject(workflowId, reason) {
		await this.rejectWorkflow(workflowId, { reason });

		this.setState({
			...this.state,
			pendingApprovals: this.state.pendingApprovals.filter(
				(p) => p.workflowId !== workflowId,
			),
		});
	}

	// Track workflow progress to update pending approvals
	async onWorkflowProgress(workflowName, workflowId, progress) {
		const p = progress;

		if (p.step === "approval" && p.status === "pending") {
			// Add to pending approvals list for UI display
			this.setState({
				...this.state,
				pendingApprovals: [
					...this.state.pendingApprovals,
					{
						workflowId,
						amount: 0, // Would come from workflow params
						description: p.message || "",
						requestedBy: "user",
						requestedAt: Date.now(),
					},
				],
			});
		}
	}
}

import { Agent, callable } from "agents";

type PendingApproval = {
	workflowId: string;
	amount: number;
	description: string;
	requestedBy: string;
	requestedAt: number;
};

type ExpenseState = {
	pendingApprovals: PendingApproval[];
};

export class ExpenseAgent extends Agent<Env, ExpenseState> {
	initialState: ExpenseState = {
		pendingApprovals: [],
	};

	// Approve a waiting workflow
	@callable()
	async approve(workflowId: string, approvedBy: string): Promise<void> {
		await this.approveWorkflow(workflowId, {
			reason: "Expense approved",
			metadata: { approvedBy, approvedAt: Date.now() },
		});

		// Update state to reflect approval
		this.setState({
			...this.state,
			pendingApprovals: this.state.pendingApprovals.filter(
				(p) => p.workflowId !== workflowId,
			),
		});
	}

	// Reject a waiting workflow
	@callable()
	async reject(workflowId: string, reason: string): Promise<void> {
		await this.rejectWorkflow(workflowId, { reason });

		this.setState({
			...this.state,
			pendingApprovals: this.state.pendingApprovals.filter(
				(p) => p.workflowId !== workflowId,
			),
		});
	}

	// Track workflow progress to update pending approvals
	async onWorkflowProgress(
		workflowName: string,
		workflowId: string,
		progress: unknown,
	): Promise<void> {
		const p = progress as { step: string; status: string; message?: string };

		if (p.step === "approval" && p.status === "pending") {
			// Add to pending approvals list for UI display
			this.setState({
				...this.state,
				pendingApprovals: [
					...this.state.pendingApprovals,
					{
						workflowId,
						amount: 0, // Would come from workflow params
						description: p.message || "",
						requestedBy: "user",
						requestedAt: Date.now(),
					},
				],
			});
		}
	}
}

超时处理

设置超时以防止 workflow 无限期等待：

const approval = await this.waitForApproval(step, {
	timeout: "7 days", // Also supports: "1 hour", "30 minutes", etc.
});

if (!approval) {
	// Timeout expired - escalate or auto-reject
	await step.reportError("Approval timeout - escalating to manager");
	throw new Error("Approval timeout");
}

const approval = await this.waitForApproval<{ approvedBy: string }>(step, {
	timeout: "7 days", // Also supports: "1 hour", "30 minutes", etc.
});

if (!approval) {
	// Timeout expired - escalate or auto-reject
	await step.reportError("Approval timeout - escalating to manager");
	throw new Error("Approval timeout");
}

通过调度升级

使用 schedule() 设置升级提醒：

import { Agent, callable } from "agents";

class ExpenseAgent extends Agent {
	@callable()
	async submitForApproval(expense) {
		// Start the approval workflow
		const workflowId = await this.runWorkflow("EXPENSE_WORKFLOW", expense);

		// Schedule reminder after 4 hours
		await this.schedule(Date.now() + 4 * 60 * 60 * 1000, "sendReminder", {
			workflowId,
		});

		// Schedule escalation after 24 hours
		await this.schedule(Date.now() + 24 * 60 * 60 * 1000, "escalateApproval", {
			workflowId,
		});

		return workflowId;
	}

	async sendReminder(payload) {
		const workflow = this.getWorkflow(payload.workflowId);
		if (workflow?.status === "waiting") {
			// Send reminder notification
			console.log("Reminder: approval still pending");
		}
	}

	async escalateApproval(payload) {
		const workflow = this.getWorkflow(payload.workflowId);
		if (workflow?.status === "waiting") {
			// Escalate to manager
			console.log("Escalating to manager");
		}
	}
}

import { Agent, callable } from "agents";

class ExpenseAgent extends Agent<Env, ExpenseState> {
	@callable()
	async submitForApproval(expense: ExpenseParams): Promise<string> {
		// Start the approval workflow
		const workflowId = await this.runWorkflow("EXPENSE_WORKFLOW", expense);

		// Schedule reminder after 4 hours
		await this.schedule(Date.now() + 4 * 60 * 60 * 1000, "sendReminder", {
			workflowId,
		});

		// Schedule escalation after 24 hours
		await this.schedule(Date.now() + 24 * 60 * 60 * 1000, "escalateApproval", {
			workflowId,
		});

		return workflowId;
	}

	async sendReminder(payload: { workflowId: string }) {
		const workflow = this.getWorkflow(payload.workflowId);
		if (workflow?.status === "waiting") {
			// Send reminder notification
			console.log("Reminder: approval still pending");
		}
	}

	async escalateApproval(payload: { workflowId: string }) {
		const workflow = this.getWorkflow(payload.workflowId);
		if (workflow?.status === "waiting") {
			// Escalate to manager
			console.log("Escalating to manager");
		}
	}
}

使用 SQL 的审计追踪

使用 this.sql 维护不可变的审计追踪：

import { Agent, callable } from "agents";

class ExpenseAgent extends Agent {
	async onStart() {
		// Create audit table
		this.sql`
      CREATE TABLE IF NOT EXISTS approval_audit (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        workflow_id TEXT NOT NULL,
        decision TEXT NOT NULL CHECK(decision IN ('approved', 'rejected')),
        decided_by TEXT NOT NULL,
        decided_at INTEGER NOT NULL,
        reason TEXT
      )
    `;
	}

	@callable()
	async approve(workflowId, userId, reason) {
		// Record the decision in SQL (immutable audit log)
		this.sql`
      INSERT INTO approval_audit (workflow_id, decision, decided_by, decided_at, reason)
      VALUES (${workflowId}, 'approved', ${userId}, ${Date.now()}, ${reason || null})
    `;

		// Process the approval
		await this.approveWorkflow(workflowId, {
			reason: reason || "Approved",
			metadata: { approvedBy: userId },
		});
	}
}

import { Agent, callable } from "agents";

class ExpenseAgent extends Agent<Env, ExpenseState> {
	async onStart() {
		// Create audit table
		this.sql`
      CREATE TABLE IF NOT EXISTS approval_audit (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        workflow_id TEXT NOT NULL,
        decision TEXT NOT NULL CHECK(decision IN ('approved', 'rejected')),
        decided_by TEXT NOT NULL,
        decided_at INTEGER NOT NULL,
        reason TEXT
      )
    `;
	}

	@callable()
	async approve(
		workflowId: string,
		userId: string,
		reason?: string,
	): Promise<void> {
		// Record the decision in SQL (immutable audit log)
		this.sql`
      INSERT INTO approval_audit (workflow_id, decision, decided_by, decided_at, reason)
      VALUES (${workflowId}, 'approved', ${userId}, ${Date.now()}, ${reason || null})
    `;

		// Process the approval
		await this.approveWorkflow(workflowId, {
			reason: reason || "Approved",
			metadata: { approvedBy: userId },
		});
	}
}

配置

{
	"name": "expense-approval",
	"main": "src/index.ts",
	// Set this to today's date
	"compatibility_date": "2026-08-17",
	"compatibility_flags": ["nodejs_compat"],
	"durable_objects": {
		"bindings": [{ "name": "EXPENSE_AGENT", "class_name": "ExpenseAgent" }],
	},
	"workflows": [
		{
			"name": "expense-workflow",
			"binding": "EXPENSE_WORKFLOW",
			"class_name": "ExpenseWorkflow",
		},
	],
	"migrations": [{ "tag": "v1", "new_sqlite_classes": ["ExpenseAgent"] }],
}

name = "expense-approval"
main = "src/index.ts"
# Set this to today's date
compatibility_date = "2026-08-17"
compatibility_flags = [ "nodejs_compat" ]

[[durable_objects.bindings]]
name = "EXPENSE_AGENT"
class_name = "ExpenseAgent"

[[workflows]]
name = "expense-workflow"
binding = "EXPENSE_WORKFLOW"
class_name = "ExpenseWorkflow"

[[migrations]]
tag = "v1"
new_sqlite_classes = [ "ExpenseAgent" ]

MCP 征求

MCP 征求让 MCP 服务器开发者决定工具调用是否需要更多信息或带外交互。你的 Agent 作为 MCP 客户端：向用户展示请求并返回其响应。这些交互通常在数分钟内完成。

Form 模式收集结构化、非敏感输入。URL 模式要求用户打开带外流程，例如第三方授权或支付。

在 onStart() 中配置面向用户的 handler：

import { Agent } from "agents";

export class MyAgent extends Agent {
	onStart() {
		this.mcp.configureElicitationHandlers({
			form: (request, serverId) =>
				this.forwardElicitationToUser(request, serverId),
			url: (request, serverId) =>
				this.forwardElicitationToUser(request, serverId),
		});
	}

	forwardElicitationToUser(request, serverId) {
		// Present the request in your UI and resolve after the user responds.
		throw new Error(
			`Implement elicitation for ${serverId}: ${request.params.message}`,
		);
	}
}

import { Agent } from "agents";
import type { ElicitRequest, ElicitResult } from "agents/mcp";

export class MyAgent extends Agent<Env> {
	onStart() {
		this.mcp.configureElicitationHandlers({
			form: (request, serverId) =>
				this.forwardElicitationToUser(request, serverId),
			url: (request, serverId) =>
				this.forwardElicitationToUser(request, serverId),
		});
	}

	private forwardElicitationToUser(
		request: ElicitRequest,
		serverId: string,
	): Promise<ElicitResult> {
		// Present the request in your UI and resolve after the user responds.
		throw new Error(
			`Implement elicitation for ${serverId}: ${request.params.message}`,
		);
	}
}

完整 form、URL 与 browser 转发模式请参阅 MCP 客户端 elicitation。server 端请求 API 请参阅 McpAgent elicitation。

持久 Code Mode 审批

对编码 agent 及其他使用 Code Mode 模式的 Agent，使用持久 Code Mode 运行时。在连接器方法上标记 requiresApproval: true，可在模型生成代码调用底层工具之前暂停。

以下示例将 GitHub MCP 工具标记为需审批，将连接器添加到持久运行时，并暴露 UI 可用于检查、批准或拒绝待处理操作的方法：

import { AIChatAgent } from "@cloudflare/ai-chat";
import {
	createCodemodeRuntime,
	DynamicWorkerExecutor,
	McpConnector,
} from "@cloudflare/codemode";
import { callable } from "agents";
import { convertToModelMessages, stepCountIs, streamText } from "ai";
import { model } from "./model";

class GitHubConnector extends McpConnector {
	connection;

	constructor(ctx, env, connection) {
		super(ctx, env);
		this.connection = connection;
	}

	name() {
		return "github";
	}

	createConnection() {
		return this.connection;
	}

	tool(name, tool) {
		if (name === "create_issue") {
			return { ...tool, requiresApproval: true };
		}
		return tool;
	}
}

export class CodingAgent extends AIChatAgent {
	runtime() {
		const server = this.mcp
			.listServers()
			.find((item) => item.name === "GitHub");
		if (!server) throw new Error("GitHub MCP server is not registered.");

		const connection = this.mcp.mcpConnections[server.id];
		if (!connection) throw new Error("GitHub MCP connection is unavailable.");

		return createCodemodeRuntime({
			ctx: this.ctx,
			executor: new DynamicWorkerExecutor({ loader: this.env.LOADER }),
			connectors: [new GitHubConnector(this.ctx, this.env, connection)],
		});
	}

	async onChatMessage() {
		const result = streamText({
			model,
			messages: await convertToModelMessages(this.messages),
			tools: { codemode: this.runtime().tool() },
			stopWhen: stepCountIs(10),
		});

		return result.toUIMessageStreamResponse();
	}

	@callable()
	async pendingApprovals() {
		return this.runtime().pending();
	}

	@callable()
	async approveExecution(executionId) {
		return this.runtime().approve({ executionId });
	}

	@callable()
	async rejectExecution(executionId, seq) {
		return this.runtime().reject({ executionId, seq });
	}
}

import { AIChatAgent } from "@cloudflare/ai-chat";
import {
	createCodemodeRuntime,
	DynamicWorkerExecutor,
	McpConnector,
	type CodemodeRuntimeHandle,
	type ConnectorTool,
	type McpConnectionLike,
	type PendingAction,
} from "@cloudflare/codemode";
import { callable } from "agents";
import { convertToModelMessages, stepCountIs, streamText } from "ai";
import { model } from "./model";

class GitHubConnector extends McpConnector<Env> {
	private connection: McpConnectionLike;

	constructor(
		ctx: DurableObjectState,
		env: Env,
		connection: McpConnectionLike,
	) {
		super(ctx, env);
		this.connection = connection;
	}

	override name() {
		return "github";
	}

	protected override createConnection() {
		return this.connection;
	}

	protected override tool(name: string, tool: ConnectorTool): ConnectorTool {
		if (name === "create_issue") {
			return { ...tool, requiresApproval: true };
		}
		return tool;
	}
}

export class CodingAgent extends AIChatAgent<Env> {
	private runtime(): CodemodeRuntimeHandle {
		const server = this.mcp
			.listServers()
			.find((item) => item.name === "GitHub");
		if (!server) throw new Error("GitHub MCP server is not registered.");

		const connection = this.mcp.mcpConnections[server.id];
		if (!connection) throw new Error("GitHub MCP connection is unavailable.");

		return createCodemodeRuntime({
			ctx: this.ctx,
			executor: new DynamicWorkerExecutor({ loader: this.env.LOADER }),
			connectors: [new GitHubConnector(this.ctx, this.env, connection)],
		});
	}

	async onChatMessage() {
		const result = streamText({
			model,
			messages: await convertToModelMessages(this.messages),
			tools: { codemode: this.runtime().tool() },
			stopWhen: stepCountIs(10),
		});

		return result.toUIMessageStreamResponse();
	}

	@callable()
	async pendingApprovals(): Promise<PendingAction[]> {
		return this.runtime().pending();
	}

	@callable()
	async approveExecution(executionId: string) {
		return this.runtime().approve({ executionId });
	}

	@callable()
	async rejectExecution(executionId: string, seq: number): Promise<boolean> {
		return this.runtime().reject({ executionId, seq });
	}
}

当生成的代码调用 github.create_issue() 时，运行时会记录待处理的方法与参数，然后在 MCP 工具执行前暂停。审批会使用相同源代码与执行 ID 启动另一次遍历。先前完成的调用从持久日志重放，已批准的调用执行，生成的代码继续运行。

待审批项与执行历史在请求完成和 Durable Object 休眠后仍然保留。执行与重放模型请参阅通过中止与重放实现审批。

构建 Workflow 审批 UI

待审批列表

使用 Agent 的 state 在 UI 中展示待审批项：

import { useAgent } from "agents/react";

function PendingApprovals() {
	const { state, agent } = useAgent({
		agent: "expense-agent",
		name: "main",
	});

	if (!state?.pendingApprovals?.length) {
		return <p>No pending approvals</p>;
	}

	return (
		<div className="approval-list">
			{state.pendingApprovals.map((item) => (
				<div key={item.workflowId} className="approval-card">
					<h3>${item.amount}</h3>
					<p>{item.description}</p>
					<p>Requested by {item.requestedBy}</p>

					<div className="actions">
						<button
							onClick={() => agent.stub.approve(item.workflowId, "admin")}
						>
							Approve
						</button>
						<button
							onClick={() => agent.stub.reject(item.workflowId, "Declined")}
						>
							Reject
						</button>
					</div>
				</div>
			))}
		</div>
	);
}

添加多个 Workflow 审批人

对需要多个审批人的敏感操作：

import { Agent, callable } from "agents";

class MultiApprovalAgent extends Agent {
	@callable()
	async approveMulti(workflowId, userId) {
		const approval = this.state.pendingMultiApprovals.find(
			(p) => p.workflowId === workflowId,
		);
		if (!approval) throw new Error("Approval not found");

		// Check if user already approved
		if (approval.currentApprovals.some((a) => a.userId === userId)) {
			throw new Error("Already approved by this user");
		}

		// Add this user's approval
		approval.currentApprovals.push({ userId, approvedAt: Date.now() });

		// Check if we have enough approvals
		if (approval.currentApprovals.length >= approval.requiredApprovals) {
			// Execute the approved action
			await this.approveWorkflow(workflowId, {
				metadata: { approvers: approval.currentApprovals },
			});
			return true;
		}

		this.setState({ ...this.state });
		return false; // Still waiting for more approvals
	}
}

import { Agent, callable } from "agents";

type MultiApproval = {
	workflowId: string;
	requiredApprovals: number;
	currentApprovals: Array<{ userId: string; approvedAt: number }>;
	rejections: Array<{ userId: string; rejectedAt: number; reason: string }>;
};

type State = {
	pendingMultiApprovals: MultiApproval[];
};

class MultiApprovalAgent extends Agent<Env, State> {
	@callable()
	async approveMulti(workflowId: string, userId: string): Promise<boolean> {
		const approval = this.state.pendingMultiApprovals.find(
			(p) => p.workflowId === workflowId,
		);
		if (!approval) throw new Error("Approval not found");

		// Check if user already approved
		if (approval.currentApprovals.some((a) => a.userId === userId)) {
			throw new Error("Already approved by this user");
		}

		// Add this user's approval
		approval.currentApprovals.push({ userId, approvedAt: Date.now() });

		// Check if we have enough approvals
		if (approval.currentApprovals.length >= approval.requiredApprovals) {
			// Execute the approved action
			await this.approveWorkflow(workflowId, {
				metadata: { approvers: approval.currentApprovals },
			});
			return true;
		}

		this.setState({ ...this.state });
		return false; // Still waiting for more approvals
	}
}

最佳实践

定义清晰的审批标准 — 仅对有实质后果的操作（支付、邮件、数据变更）要求确认
提供详细上下文 — 向用户准确展示操作将执行的内容，包括所有参数
实现超时 — 使用 schedule() 在合理期限后升级或自动拒绝
维护审计追踪 — 使用 this.sql 记录所有审批决策以满足合规要求
处理连接断开 — 将待审批项存储在 Agent state 中，使其在断连后仍然保留
优雅降级 — 若审批被拒绝，提供回退行为