跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

运行 Workflows

最后更新 查看 MarkdownAgent 设置

将 Cloudflare Workflows 与 Agent 集成，实现持久的多步骤后台处理，同时由 Agent 处理实时通信。

快速入门

1. 定义 Workflow

扩展 AgentWorkflow 以带类型地访问发起方 Agent：

import { AgentWorkflow } from "agents/workflows";

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

		const result = await step.do("process-data", async () => {
			return processData(params.data);
		});

		// Non-durable: progress reporting (may repeat on retry)
		await this.reportProgress({
			step: "process",
			status: "complete",
			percent: 0.5,
		});

		// Broadcast to connected WebSocket clients
		this.broadcastToClients({ type: "update", taskId: params.taskId });

		await step.do("save-results", async () => {
			// Call Agent methods via RPC
			await this.agent.saveResult(params.taskId, result);
		});

		// Durable: idempotent, won't repeat on retry
		await step.reportComplete(result);
		return result;
	}
}
import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";
import type { MyAgent } from "./agent";

type TaskParams = { taskId: string; data: string };

export class ProcessingWorkflow extends AgentWorkflow<MyAgent, TaskParams> {
	async run(event: AgentWorkflowEvent<TaskParams>, step: AgentWorkflowStep) {
		const params = event.payload;

		const result = await step.do("process-data", async () => {
			return processData(params.data);
		});

		// Non-durable: progress reporting (may repeat on retry)
		await this.reportProgress({
			step: "process",
			status: "complete",
			percent: 0.5,
		});

		// Broadcast to connected WebSocket clients
		this.broadcastToClients({ type: "update", taskId: params.taskId });

		await step.do("save-results", async () => {
			// Call Agent methods via RPC
			await this.agent.saveResult(params.taskId, result);
		});

		// Durable: idempotent, won't repeat on retry
		await step.reportComplete(result);
		return result;
	}
}

2. 从 Agent 启动 Workflow

使用 runWorkflow() 启动并追踪 workflow：

import { Agent } from "agents";

export class MyAgent extends Agent {
	async startTask(taskId, data) {
		const instanceId = await this.runWorkflow("PROCESSING_WORKFLOW", {
			taskId,
			data,
		});
		return { instanceId };
	}

	async onWorkflowProgress(workflowName, instanceId, progress) {
		this.broadcast(JSON.stringify({ type: "workflow-progress", progress }));
	}

	async onWorkflowComplete(workflowName, instanceId, result) {
		console.log(`Workflow completed:`, result);
	}

	async saveResult(taskId, result) {
		this
			.sql`INSERT INTO results (task_id, data) VALUES (${taskId}, ${JSON.stringify(result)})`;
	}
}
import { Agent } from "agents";

export class MyAgent extends Agent {
	async startTask(taskId: string, data: string) {
		const instanceId = await this.runWorkflow("PROCESSING_WORKFLOW", {
			taskId,
			data,
		});
		return { instanceId };
	}

	async onWorkflowProgress(
		workflowName: string,
		instanceId: string,
		progress: unknown,
	) {
		this.broadcast(JSON.stringify({ type: "workflow-progress", progress }));
	}

	async onWorkflowComplete(
		workflowName: string,
		instanceId: string,
		result?: unknown,
	) {
		console.log(`Workflow completed:`, result);
	}

	async saveResult(taskId: string, result: unknown) {
		this
			.sql`INSERT INTO results (task_id, data) VALUES (${taskId}, ${JSON.stringify(result)})`;
	}
}

3. 配置 Wrangler

{
	"name": "my-app",
	"main": "src/index.ts",
	// Set this to today's date
	"compatibility_date": "2026-08-17",
	"durable_objects": {
		"bindings": [{ "name": "MY_AGENT", "class_name": "MyAgent" }],
	},
	"workflows": [
		{
			"name": "processing-workflow",
			"binding": "PROCESSING_WORKFLOW",
			"class_name": "ProcessingWorkflow",
		},
	],
	"migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyAgent"] }],
}
name = "my-app"
main = "src/index.ts"
# Set this to today's date
compatibility_date = "2026-08-17"

[[durable_objects.bindings]]
name = "MY_AGENT"
class_name = "MyAgent"

[[workflows]]
name = "processing-workflow"
binding = "PROCESSING_WORKFLOW"
class_name = "ProcessingWorkflow"

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

AgentWorkflow 类

与 Agents 集成的 Workflows 基类。

类型参数

参数 描述
AgentType 用于带类型 RPC 的 Agent 类类型
Params 传给 Workflow 的参数
ProgressType 进度报告类型（默认为 DefaultProgress）
Env 环境类型（默认为 Cloudflare.Env）

属性

属性 类型 描述
agent Stub 用于调用 Agent 方法的带类型 stub。对于从子 Agent 启动的 Workflow，这是回到发起方面的仅 RPC stub；HTTP 或 WebSocket fetch() 流量请使用子 Agent 路由
instanceId string Workflow 实例 ID
workflowName string Workflow 绑定名称
env Env 环境绑定

实例方法（非持久）

这些方法可能在重试时重复执行。用于轻量、高频更新。

reportProgress(progress)

向 Agent 报告进度。触发 onWorkflowProgress 回调。

await this.reportProgress({
	step: "processing",
	status: "running",
	percent: 0.5,
});
await this.reportProgress({
	step: "processing",
	status: "running",
	percent: 0.5,
});

broadcastToClients(message)

向连接到 Agent 的所有 WebSocket 客户端广播消息。

this.broadcastToClients({ type: "update", data: result });
this.broadcastToClients({ type: "update", data: result });

waitForApproval(step, options?)

等待审批事件。若被拒绝则抛出 WorkflowRejectedError。

const approval = await this.waitForApproval(step, {
	timeout: "7 days",
});
const approval = await this.waitForApproval<{ approvedBy: string }>(step, {
	timeout: "7 days",
});

Step 方法（durable）

这些方法具有幂等性，重试时不会重复执行。用于必须持久化的状态变更。

方法 描述
step.reportComplete(result?) 报告成功完成
step.reportError(error) 报告错误
step.sendEvent(event) 向 Agent 发送自定义事件
step.updateAgentState(state) 替换 Agent 状态（广播到客户端）
step.mergeAgentState(partial) 合并到 Agent 状态（广播到客户端）
step.resetAgentState() 将 Agent 状态重置为 initialState

DefaultProgress 类型

type DefaultProgress = {
	step?: string;
	status?: "pending" | "running" | "complete" | "error";
	message?: string;
	percent?: number;
	[key: string]: unknown;
};

Agent Workflow 方法

Agent 类上可用于 Workflow 管理的方法。

runWorkflow(workflowName, params, options?)

启动 workflow 实例并在 Agent 数据库中追踪。

参数：

参数 类型 描述
workflowName string env 中的 Workflow 绑定名称
params object 传给 Workflow 的参数
options.id string 自定义 Workflow ID（未提供则自动生成）
options.metadata object 用于查询的元数据（不会传给 Workflow）
options.agentBinding string Agent 绑定名称（未提供则自动检测）。从子 Agent 调用时，这是根 Agent 绑定名称

返回值： Promise<string> — Workflow 实例 ID

const instanceId = await this.runWorkflow(
	"MY_WORKFLOW",
	{ taskId: "123" },
	{
		metadata: { userId: "user-456", priority: "high" },
	},
);
const instanceId = await this.runWorkflow(
	"MY_WORKFLOW",
	{ taskId: "123" },
	{
		metadata: { userId: "user-456", priority: "high" },
	},
);

从子 Agent 启动 Workflow

子 Agent 可直接调用 this.runWorkflow()。Workflow 在发起方子 Agent 的 SQLite 数据库中跟踪，AgentWorkflow 内的 this.agent 将 RPC、回调、状态更新与广播路由回同一子 Agent。

父 Agent 不会自动列出或控制子 Agent 启动的 Workflow。SubAgentStub<T> 仅暴露用户定义的方法，而非继承的 Agent 方法（如 approveWorkflow() 或 getWorkflow()）。要从父 Agent 控制子 Agent 启动的 Workflow，请在子 Agent 上定义小型包装方法，并通过子 Agent stub 调用这些包装。

export class ParentAgent extends Agent {
	async startChildWorkflow(childName, task) {
		const child = await this.subAgent(ChildAgent, childName);
		return child.startWorkflow(task);
	}

	async approveChildWorkflow(childName, workflowId) {
		const child = await this.subAgent(ChildAgent, childName);
		return child.approveChildWorkflow(workflowId);
	}
}

export class ChildAgent extends Agent {
	async startWorkflow(task) {
		return this.runWorkflow("CHILD_WORKFLOW", { task });
	}

	async approveChildWorkflow(workflowId) {
		return this.approveWorkflow(workflowId);
	}

	async getChildWorkflow(workflowId) {
		return this.getWorkflow(workflowId);
	}
}
export class ParentAgent extends Agent {
	async startChildWorkflow(childName: string, task: string) {
		const child = await this.subAgent(ChildAgent, childName);
		return child.startWorkflow(task);
	}

	async approveChildWorkflow(childName: string, workflowId: string) {
		const child = await this.subAgent(ChildAgent, childName);
		return child.approveChildWorkflow(workflowId);
	}
}

export class ChildAgent extends Agent {
	async startWorkflow(task: string) {
		return this.runWorkflow("CHILD_WORKFLOW", { task });
	}

	async approveChildWorkflow(workflowId: string) {
		return this.approveWorkflow(workflowId);
	}

	async getChildWorkflow(workflowId: string) {
		return this.getWorkflow(workflowId);
	}
}

对于子 Agent 来源，AgentWorkflow.agent 是仅 RPC 的 stub。用它调用 Agent 方法，但外部 HTTP 或 WebSocket 路由请使用 routeSubAgentRequest() 或 /agents/{parent}/{name}/sub/{child}/{name} URL 形状，而非 this.agent.fetch()。

路由约束

由于发起方身份会持久化在 Workflow 参数中，并在每次回调时重放，以下约束适用于所有 Workflow（子 Agent 与顶层相同）：

  • 回调按名称解析 Agent。 运行时使用 getAgentByName(...) 重新解析发起方 Agent。若用原始 Durable Object ID（idFromString / get(id)）而非名称寻址 Agent，回调会落到不同实例。请从按名称寻址的 Agent 启动 Workflow。
  • 类名须在打包后保留。 发起路径以 constructor.name 为键。配置打包器保留类名（esbuild keepNames: true），以便进度、完成与 this.agent RPC 能路由回正确 facet。
  • agentBinding 是根绑定。 从子 Agent 传入 options.agentBinding 时，使用 根 Agent 的 Durable Object 绑定名称，而非子绑定。

sendWorkflowEvent(workflowName, instanceId, event)

向运行中的 workflow 发送事件。

await this.sendWorkflowEvent("MY_WORKFLOW", instanceId, {
	type: "custom-event",
	payload: { action: "proceed" },
});
await this.sendWorkflowEvent("MY_WORKFLOW", instanceId, {
	type: "custom-event",
	payload: { action: "proceed" },
});

getWorkflowStatus(workflowName, instanceId)

获取 workflow 状态并更新追踪记录。

const status = await this.getWorkflowStatus("MY_WORKFLOW", instanceId);
// { status: 'running', output: null, error: null }
const status = await this.getWorkflowStatus("MY_WORKFLOW", instanceId);
// { status: 'running', output: null, error: null }

getWorkflow(instanceId)

按 ID 获取已追踪的 workflow。

const workflow = this.getWorkflow(instanceId);
// { instanceId, workflowName, status, metadata, error, createdAt, ... }
const workflow = this.getWorkflow(instanceId);
// { instanceId, workflowName, status, metadata, error, createdAt, ... }

getWorkflows(criteria?)

使用基于 cursor 的分页查询已追踪 workflow。返回包含 workflow、总数与下一页 cursor 的 WorkflowPage。

// Get running workflows (default limit is 50, max is 100)
const { workflows, total } = this.getWorkflows({ status: "running" });

// Filter by metadata
const { workflows: userWorkflows } = this.getWorkflows({
	metadata: { userId: "user-456" },
});

// Pagination with cursor
const page1 = this.getWorkflows({
	status: ["complete", "errored"],
	limit: 20,
	orderBy: "desc",
});

console.log(`Showing ${page1.workflows.length} of ${page1.total} workflows`);

// Get next page using cursor
if (page1.nextCursor) {
	const page2 = this.getWorkflows({
		status: ["complete", "errored"],
		limit: 20,
		orderBy: "desc",
		cursor: page1.nextCursor,
	});
}
// Get running workflows (default limit is 50, max is 100)
const { workflows, total } = this.getWorkflows({ status: "running" });

// Filter by metadata
const { workflows: userWorkflows } = this.getWorkflows({
	metadata: { userId: "user-456" },
});

// Pagination with cursor
const page1 = this.getWorkflows({
	status: ["complete", "errored"],
	limit: 20,
	orderBy: "desc",
});

console.log(`Showing ${page1.workflows.length} of ${page1.total} workflows`);

// Get next page using cursor
if (page1.nextCursor) {
	const page2 = this.getWorkflows({
		status: ["complete", "errored"],
		limit: 20,
		orderBy: "desc",
		cursor: page1.nextCursor,
	});
}

WorkflowPage 类型：

type WorkflowPage = {
	workflows: WorkflowInfo[];
	total: number; // Total matching workflows
	nextCursor: string | null; // null when no more pages
};

deleteWorkflow(instanceId)

删除单条 workflow 实例追踪记录。删除成功返回 true，未找到返回 false。

deleteWorkflows(criteria?)

删除符合条件的 workflow 实例追踪记录。

// Delete completed workflow instances older than 7 days
this.deleteWorkflows({
	status: "complete",
	createdBefore: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000),
});

// Delete all errored and terminated workflows
this.deleteWorkflows({
	status: ["errored", "terminated"],
});
// Delete completed workflow instances older than 7 days
this.deleteWorkflows({
	status: "complete",
	createdBefore: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000),
});

// Delete all errored and terminated workflows
this.deleteWorkflows({
	status: ["errored", "terminated"],
});

terminateWorkflow(instanceId)

立即终止运行中的 workflow。将 status 设为 "terminated"。

await this.terminateWorkflow(instanceId);
await this.terminateWorkflow(instanceId);

pauseWorkflow(instanceId)

暂停运行中的 workflow。之后可用 resumeWorkflow() 恢复。

await this.pauseWorkflow(instanceId);
await this.pauseWorkflow(instanceId);

resumeWorkflow(instanceId)

恢复已暂停的 workflow。

await this.resumeWorkflow(instanceId);
await this.resumeWorkflow(instanceId);

restartWorkflow(instanceId, options?)

使用相同 ID 从头重启 workflow 实例。

// Reset tracking (default) - clears timestamps and error fields
await this.restartWorkflow(instanceId);

// Preserve original timestamps
await this.restartWorkflow(instanceId, { resetTracking: false });
// Reset tracking (default) - clears timestamps and error fields
await this.restartWorkflow(instanceId);

// Preserve original timestamps
await this.restartWorkflow(instanceId, { resetTracking: false });

approveWorkflow(instanceId, options?)

批准等待中的 workflow。在 workflow 中与 waitForApproval() 配合使用。

await this.approveWorkflow(instanceId, {
	reason: "Approved by admin",
	metadata: { approvedBy: userId },
});
await this.approveWorkflow(instanceId, {
	reason: "Approved by admin",
	metadata: { approvedBy: userId },
});

rejectWorkflow(instanceId, options?)

拒绝等待中的 workflow。会导致 waitForApproval() 抛出 WorkflowRejectedError。

await this.rejectWorkflow(instanceId, { reason: "Request denied" });
await this.rejectWorkflow(instanceId, { reason: "Request denied" });

migrateWorkflowBinding(oldName, newName)

重命名 workflow binding 后迁移已追踪 workflow。

class MyAgent extends Agent {
	async onStart() {
		this.migrateWorkflowBinding("OLD_WORKFLOW", "NEW_WORKFLOW");
	}
}
class MyAgent extends Agent {
	async onStart() {
		this.migrateWorkflowBinding("OLD_WORKFLOW", "NEW_WORKFLOW");
	}
}

生命周期回调

在 Agent 中 override 这些方法以处理 workflow 事件：

回调 参数 描述
onWorkflowProgress workflowName, instanceId, progress Workflow 报告进度时调用
onWorkflowComplete workflowName, instanceId, result? Workflow 完成时调用
onWorkflowError workflowName, instanceId, error Workflow 出错时调用
onWorkflowEvent workflowName, instanceId, event Workflow 发送事件时调用
onWorkflowCallback callback: WorkflowCallback 所有回调类型时调用 
class MyAgent extends Agent {
	async onWorkflowProgress(workflowName, instanceId, progress) {
		this.broadcast(
			JSON.stringify({ type: "progress", workflowName, instanceId, progress }),
		);
	}

	async onWorkflowComplete(workflowName, instanceId, result) {
		console.log(`${workflowName}/${instanceId} completed`);
	}

	async onWorkflowError(workflowName, instanceId, error) {
		console.error(`${workflowName}/${instanceId} failed:`, error);
	}
}
class MyAgent extends Agent {
	async onWorkflowProgress(
		workflowName: string,
		instanceId: string,
		progress: unknown,
	) {
		this.broadcast(
			JSON.stringify({ type: "progress", workflowName, instanceId, progress }),
		);
	}

	async onWorkflowComplete(
		workflowName: string,
		instanceId: string,
		result?: unknown,
	) {
		console.log(`${workflowName}/${instanceId} completed`);
	}

	async onWorkflowError(
		workflowName: string,
		instanceId: string,
		error: string,
	) {
		console.error(`${workflowName}/${instanceId} failed:`, error);
	}
}

Workflow 追踪

使用 runWorkflow() 启动的 Workflow 会自动追踪在发起方 Agent 的内部数据库中。可使用上文方法（getWorkflow()、getWorkflows()、deleteWorkflow() 等）查询、筛选与管理 Workflow。

状态值

状态 描述
queued 等待启动
running 正在执行
paused 用户暂停
waiting 等待事件
complete 成功完成
errored 出错失败
terminated 手动终止

在 runWorkflow() 中使用 metadata 选项存储可查询信息（如 user ID 或任务类型），以便之后用 getWorkflows() 筛选。

示例

Human-in-the-loop 审批

import { AgentWorkflow } from "agents/workflows";

export class ApprovalWorkflow extends AgentWorkflow {
	async run(event, step) {
		const request = await step.do("prepare", async () => {
			return { ...event.payload, preparedAt: Date.now() };
		});

		await this.reportProgress({
			step: "approval",
			status: "pending",
			message: "Awaiting approval",
		});

		// Throws WorkflowRejectedError if rejected
		const approval = await this.waitForApproval(step, {
			timeout: "7 days",
		});

		console.log("Approved by:", approval?.approvedBy);

		const result = await step.do("execute", async () => {
			return executeRequest(request);
		});

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

class MyAgent extends Agent {
	async handleApproval(instanceId, userId) {
		await this.approveWorkflow(instanceId, {
			reason: "Approved by admin",
			metadata: { approvedBy: userId },
		});
	}

	async handleRejection(instanceId, reason) {
		await this.rejectWorkflow(instanceId, { reason });
	}
}
import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";

export class ApprovalWorkflow extends AgentWorkflow<MyAgent, RequestParams> {
	async run(event: AgentWorkflowEvent<RequestParams>, step: AgentWorkflowStep) {
		const request = await step.do("prepare", async () => {
			return { ...event.payload, preparedAt: Date.now() };
		});

		await this.reportProgress({
			step: "approval",
			status: "pending",
			message: "Awaiting approval",
		});

		// Throws WorkflowRejectedError if rejected
		const approval = await this.waitForApproval<{ approvedBy: string }>(step, {
			timeout: "7 days",
		});

		console.log("Approved by:", approval?.approvedBy);

		const result = await step.do("execute", async () => {
			return executeRequest(request);
		});

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

class MyAgent extends Agent {
	async handleApproval(instanceId: string, userId: string) {
		await this.approveWorkflow(instanceId, {
			reason: "Approved by admin",
			metadata: { approvedBy: userId },
		});
	}

	async handleRejection(instanceId: string, reason: string) {
		await this.rejectWorkflow(instanceId, { reason });
	}
}

带退避的重试

import { AgentWorkflow } from "agents/workflows";

export class ResilientWorkflow extends AgentWorkflow {
	async run(event, step) {
		const result = await step.do(
			"call-api",
			{
				retries: { limit: 5, delay: "10 seconds", backoff: "exponential" },
				timeout: "5 minutes",
			},
			async () => {
				const response = await fetch("https://api.example.com/process", {
					method: "POST",
					body: JSON.stringify(event.payload),
				});
				if (!response.ok) throw new Error(`API error: ${response.status}`);
				return response.json();
			},
		);

		await step.reportComplete(result);
		return result;
	}
}
import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";

export class ResilientWorkflow extends AgentWorkflow<MyAgent, TaskParams> {
	async run(event: AgentWorkflowEvent<TaskParams>, step: AgentWorkflowStep) {
		const result = await step.do(
			"call-api",
			{
				retries: { limit: 5, delay: "10 seconds", backoff: "exponential" },
				timeout: "5 minutes",
			},
			async () => {
				const response = await fetch("https://api.example.com/process", {
					method: "POST",
					body: JSON.stringify(event.payload),
				});
				if (!response.ok) throw new Error(`API error: ${response.status}`);
				return response.json();
			},
		);

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

状态同步

Workflow 可通过 step 持久更新 Agent 状态，并自动广播到所有已连接客户端：

import { AgentWorkflow } from "agents/workflows";

export class StatefulWorkflow extends AgentWorkflow {
	async run(event, step) {
		// Replace entire state (durable, broadcasts to clients)
		await step.updateAgentState({
			currentTask: {
				id: event.payload.taskId,
				status: "processing",
				startedAt: Date.now(),
			},
		});

		const result = await step.do("process", async () =>
			processTask(event.payload),
		);

		// Merge partial state (durable, keeps existing fields)
		await step.mergeAgentState({
			currentTask: { status: "complete", result, completedAt: Date.now() },
		});

		await step.reportComplete(result);
		return result;
	}
}
import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";

export class StatefulWorkflow extends AgentWorkflow<MyAgent, TaskParams> {
	async run(event: AgentWorkflowEvent<TaskParams>, step: AgentWorkflowStep) {
		// Replace entire state (durable, broadcasts to clients)
		await step.updateAgentState({
			currentTask: {
				id: event.payload.taskId,
				status: "processing",
				startedAt: Date.now(),
			},
		});

		const result = await step.do("process", async () =>
			processTask(event.payload),
		);

		// Merge partial state (durable, keeps existing fields)
		await step.mergeAgentState({
			currentTask: { status: "complete", result, completedAt: Date.now() },
		});

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

自定义进度类型

为领域特定报告定义自定义进度类型：

import { AgentWorkflow } from "agents/workflows";

// Custom progress type for data pipeline

// Workflow with custom progress type (3rd type parameter)
export class ETLWorkflow extends AgentWorkflow {
	async run(event, step) {
		await this.reportProgress({
			stage: "extract",
			recordsProcessed: 0,
			totalRecords: 1000,
			currentTable: "users",
		});

		// ... processing
	}
}

// Agent receives typed progress
class MyAgent extends Agent {
	async onWorkflowProgress(workflowName, instanceId, progress) {
		const p = progress;
		console.log(`Stage: ${p.stage}, ${p.recordsProcessed}/${p.totalRecords}`);
	}
}
import { AgentWorkflow } from "agents/workflows";
import type { AgentWorkflowEvent, AgentWorkflowStep } from "agents/workflows";

// Custom progress type for data pipeline
type PipelineProgress = {
	stage: "extract" | "transform" | "load";
	recordsProcessed: number;
	totalRecords: number;
	currentTable?: string;
};

// Workflow with custom progress type (3rd type parameter)
export class ETLWorkflow extends AgentWorkflow<
	MyAgent,
	ETLParams,
	PipelineProgress
> {
	async run(event: AgentWorkflowEvent<ETLParams>, step: AgentWorkflowStep) {
		await this.reportProgress({
			stage: "extract",
			recordsProcessed: 0,
			totalRecords: 1000,
			currentTable: "users",
		});

		// ... processing
	}
}

// Agent receives typed progress
class MyAgent extends Agent {
	async onWorkflowProgress(
		workflowName: string,
		instanceId: string,
		progress: unknown,
	) {
		const p = progress as PipelineProgress;
		console.log(`Stage: ${p.stage}, ${p.recordsProcessed}/${p.totalRecords}`);
	}
}

清理策略

内部 cf_agents_workflows 表可能无限增长，因此应实现保留策略：

class MyAgent extends Agent {
	// Option 1: Delete on completion
	async onWorkflowComplete(workflowName, instanceId, result) {
		// Process result first, then delete
		this.deleteWorkflow(instanceId);
	}

	// Option 2: Scheduled cleanup (keep recent history)
	async cleanupOldWorkflows() {
		this.deleteWorkflows({
			status: ["complete", "errored"],
			createdBefore: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000),
		});
	}

	// Option 3: Keep all history for compliance/auditing
	// Don't call deleteWorkflows() - query historical data as needed
}
class MyAgent extends Agent {
	// Option 1: Delete on completion
	async onWorkflowComplete(
		workflowName: string,
		instanceId: string,
		result?: unknown,
	) {
		// Process result first, then delete
		this.deleteWorkflow(instanceId);
	}

	// Option 2: Scheduled cleanup (keep recent history)
	async cleanupOldWorkflows() {
		this.deleteWorkflows({
			status: ["complete", "errored"],
			createdBefore: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000),
		});
	}

	// Option 3: Keep all history for compliance/auditing
	// Don't call deleteWorkflows() - query historical data as needed
}

双向通信

Workflow 到 Agent

// Direct RPC call (typed)
await this.agent.updateTaskStatus(taskId, "processing");
const data = await this.agent.getData(taskId);

// Non-durable callbacks (may repeat on retry, use for frequent updates)
await this.reportProgress({ step: "process", percent: 0.5 });
this.broadcastToClients({ type: "update", data });

// Durable callbacks via step (idempotent, won't repeat on retry)
await step.reportComplete(result);
await step.reportError("Something went wrong");
await step.sendEvent({ type: "custom", data: {} });

// Durable state synchronization via step (broadcasts to clients)
await step.updateAgentState({ status: "processing" });
await step.mergeAgentState({ progress: 0.5 });
// Direct RPC call (typed)
await this.agent.updateTaskStatus(taskId, "processing");
const data = await this.agent.getData(taskId);

// Non-durable callbacks (may repeat on retry, use for frequent updates)
await this.reportProgress({ step: "process", percent: 0.5 });
this.broadcastToClients({ type: "update", data });

// Durable callbacks via step (idempotent, won't repeat on retry)
await step.reportComplete(result);
await step.reportError("Something went wrong");
await step.sendEvent({ type: "custom", data: {} });

// Durable state synchronization via step (broadcasts to clients)
await step.updateAgentState({ status: "processing" });
await step.mergeAgentState({ progress: 0.5 });

Agent 到 Workflow

// Send event to waiting workflow
await this.sendWorkflowEvent("MY_WORKFLOW", instanceId, {
	type: "custom-event",
	payload: { action: "proceed" },
});

// Approve/reject workflows using convenience methods
await this.approveWorkflow(instanceId, {
	reason: "Approved by admin",
	metadata: { approvedBy: userId },
});

await this.rejectWorkflow(instanceId, { reason: "Request denied" });
// Send event to waiting workflow
await this.sendWorkflowEvent("MY_WORKFLOW", instanceId, {
	type: "custom-event",
	payload: { action: "proceed" },
});

// Approve/reject workflows using convenience methods
await this.approveWorkflow(instanceId, {
	reason: "Approved by admin",
	metadata: { approvedBy: userId },
});

await this.rejectWorkflow(instanceId, { reason: "Request denied" });

最佳实践

  1. 保持 workflow 聚焦 — 每个逻辑任务一个 workflow
  2. 使用有意义的 step 名称 — 有助于调试与可观测性
  3. 定期报告进度 — 让用户了解进展
  4. 优雅处理错误 — 抛出前使用 reportError()
  5. 清理已完成 workflow — 为追踪表实现保留策略
  6. 处理 workflow binding 重命名 — 在 wrangler.jsonc 中重命名 workflow binding 时使用 migrateWorkflowBinding()

限制

约束 限制
最大 step 数 每个 workflow 10,000（默认）/ 最高可配置 25,000
State 大小 每个 workflow 10 MB
事件等待时间 最长 1 年
Step 执行时间 每个 step 30 分钟

Workflow 无法直接打开 WebSocket 连接。请通过 Agent 使用 broadcastToClients() 与已连接客户端通信。

相关资源

这篇文档对您有帮助吗？