Run Workflows·续

migrateWorkflowBinding(oldName, newName)

Migrate tracked workflows after renaming a workflow binding.

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

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

Lifecycle callbacks

Override these methods in your Agent to handle workflow events:

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 tracking

Workflows started with runWorkflow() are automatically tracked in the originating Agent's internal database. You can query, filter, and manage workflows using the methods described above (getWorkflow(), getWorkflows(), deleteWorkflow(), etc.).

Sub-agent scoping

getWorkflows() and getWorkflowById() only see workflows tracked in this Agent's storage. If a sub-agent starts the workflow, the row lives in that sub-agent's own cf_agents_workflows table, not in the parent's. To build a combined view, expose a wrapper method on each child (for example, listMyWorkflows()) and aggregate the results across your sub-agents yourself.

Status values

Use the metadata option in runWorkflow() to store queryable information (like user IDs or task types) that you can filter on later with getWorkflows().

Examples

Human-in-the-loop approval

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

Retry with backoff

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

State synchronization

Workflows can update Agent state durably via step, which automatically broadcasts to all connected clients:

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

Custom progress types

Define custom progress types for domain-specific reporting:

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

Cleanup strategy

The internal cf_agents_workflows table can grow unbounded, so implement a retention policy:

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
}

Bidirectional communication

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

Best practices

1. Keep workflows focused — One workflow per logical task
2. Use meaningful step names — Helps with debugging and observability
3. Report progress regularly — Keeps users informed
4. Handle errors gracefully — Use reportError() before throwing
5. Clean up completed workflows — Implement a retention policy for the tracking table
6. Handle workflow binding renames — Use migrateWorkflowBinding() when renaming workflow bindings in wrangler.jsonc

Limitations

Workflows cannot open WebSocket connections directly. Use broadcastToClients() to communicate with connected clients through the Agent.

Related resources

Workflows documentation

Learn about Cloudflare Workflows fundamentals.

Store and sync state

Persist and synchronize agent state.

Schedule tasks

Time-based task execution.

Human-in-the-loop

Approval flows and manual intervention patterns.

Was this helpful?

YesNo

On this page

Docs

{"@context":"https://schema.org","@type":"TechArticle","@id":"https://developers.cloudflare.com/agents/runtime/execution/run-workflows/#page","headline":"Run Workflows","description":"Integrate Cloudflare Workflows with Agents for durable, multi-step background processing and failure recovery.","url":"https://developers.cloudflare.com/agents/runtime/execution/run-workflows/","inLanguage":"en","image":"https://developers.cloudflare.com/agents/runtime/execution/run-workflows/og.png?v=611bc41a222bbe40","dateModified":"2026-10-12","publisher":{"@type":"Organization","name":"Cloudflare","description":"One platform for your apps, agents, and workforce. Build, secure, and scale without managing infrastructure","url":"https://www.cloudflare.com/","sameAs":["https://github.com/cloudflare","https://www.linkedin.com/company/cloudflare","https://x.com/cloudflare"],"logo":{"@type":"ImageObject","url":"https://developers.cloudflare.com/logo.svg"},"address":{"@type":"PostalAddress","streetAddress":"101 Townsend St","addressLocality":"San Francisco","addressRegion":"CA","postalCode":"94107","addressCountry":"US"},"contactPoint":[{"@type":"ContactPoint","contactType":"Customer Support","url":"https://support.cloudflare.com/","availableLanguage":["English"]},{"@type":"ContactPoint","contactType":"Sales","url":"https://www.cloudflare.com/contact/","availableLanguage":["English"]}]},"isPartOf":{"@type":"WebSite","@id":"https://developers.cloudflare.com/#website","name":"Cloudflare Docs","url":"https://developers.cloudflare.com/"}}

description: Schedule delayed, date-based, cron, and interval tasks on Agents with persistent SQLite-backed execution. title: Schedule tasks image: https://developers.cloudflare.com/agents/runtime/execution/schedule-tasks/og.png?v=dc9be7068b6b2c24

Skip to content

Documentation Index
Fetch the complete documentation index at: https://developers.cloudflare.com/agents/llms.txt
Use this file to discover all available pages before exploring further.