Run Workflows

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;
	}
}

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)})`;
	}
}

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

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

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

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

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

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

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" },
	},
);

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);
	}
}

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

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

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 }

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

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

// 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,
	});
}

// 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"],
});

await this.terminateWorkflow(instanceId);

await this.terminateWorkflow(instanceId);

await this.pauseWorkflow(instanceId);

await this.pauseWorkflow(instanceId);

await this.resumeWorkflow(instanceId);

await this.resumeWorkflow(instanceId);

// 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 });

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

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

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

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

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

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

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);
	}
}

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;
	}
}

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}`);
	}
}

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
}

// 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 });

// 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" });