Workflows Example

Build multi-step workflows with human-in-the-loop approval, real-time progress tracking, and durable state management

Demonstrates the workflow integration features of Cloudflare Agents - multiple concurrent workflows, real-time progress updates, human-in-the-loop approval gates, and paginated workflow management.

What it demonstrates

• Multiple concurrent workflows - Start and track many tasks simultaneously
• Real-time progress updates via WebSocket
• Human-in-the-loop approval gate per workflow
• Paginated workflow list with status tracking via getWorkflows()
• Per-workflow approve/reject controls in the UI
• Workflow lifecycle callbacks - progress, completion, error handling

Features

Feature	API Used
Start workflow	agent.runWorkflow()
Query workflows	agent.getWorkflows()
Typed progress	workflow.reportProgress({ step, status, percent, message })
Wait for approval	workflow.waitForApproval(step, options)
Approve workflow	agent.approveWorkflow(id, data)
Reject workflow	agent.rejectWorkflow(id, data)
State sync	workflow.mergeAgentState(partial)
Progress callbacks	agent.onWorkflowProgress()
Completion callbacks	agent.onWorkflowComplete()

Server Implementation

Agent

src/server.ts

import { Agent, callable } from "agents";
import { AgentWorkflow } from "agents/workflows";
import type { DefaultProgress, WorkflowInfo } from "agents/workflows";

export class TaskAgent extends Agent<Env, AgentState> {
  initialState: AgentState = {};

  @callable()
  async submitTask(taskName: string): Promise<WorkflowItem> {
    const taskId = crypto.randomUUID();

    // Start the workflow
    const workflowId = await this.runWorkflow(
      "TASK_WORKFLOW",
      { taskId, taskName },
      { metadata: { taskName } }
    );

    // Create workflow item for UI
    const newWorkflow: WorkflowItem = {
      id: workflowId,
      workflowId,
      workflowName: "TASK_WORKFLOW",
      metadata: { taskName },
      error: null,
      createdAt: new Date(),
      updatedAt: new Date(),
      completedAt: null,
      status: "queued",
      taskName,
      progress: { step: "starting", status: "pending", percent: 0 },
      waitingForApproval: false
    };

    // Broadcast to all clients
    this.broadcast(JSON.stringify({ type: "workflow:added", workflow: newWorkflow }));

    return newWorkflow;
  }

  @callable()
  async approve(workflowId: string, reason?: string): Promise<void> {
    await this.approveWorkflow(workflowId, {
      reason: reason || "Approved by user",
      metadata: { approvedAt: Date.now() }
    });

    this.broadcast(JSON.stringify({
      type: "workflow:updated",
      workflowId,
      updates: {
        waitingForApproval: false,
        status: "running",
        progress: {
          step: "approved",
          status: "running",
          percent: 0.6,
          message: "Approval received, continuing..."
        }
      }
    }));
  }

  @callable()
  async reject(workflowId: string, reason?: string): Promise<void> {
    await this.rejectWorkflow(workflowId, {
      reason: reason || "Rejected by user"
    });
  }

  @callable()
  listWorkflows(cursor?: string, limit = 5): WorkflowPage {
    const page = this.getWorkflows({
      workflowName: "TASK_WORKFLOW",
      orderBy: "desc",
      limit,
      cursor
    });

    const workflows = page.workflows.map((info) => this.toWorkflowItem(info));

    return {
      workflows,
      total: page.total,
      nextCursor: page.nextCursor
    };
  }

  // Lifecycle callbacks
  async onWorkflowProgress(
    workflowName: string,
    workflowId: string,
    progress: unknown
  ): Promise<void> {
    const p = progress as DefaultProgress & { waitingForApproval?: boolean };
    console.log(`Progress: ${workflowName}/${workflowId}`, p);

    this.broadcast(JSON.stringify({
      type: "workflow:updated",
      workflowId,
      updates: {
        progress: p,
        waitingForApproval: p.waitingForApproval ?? false,
        status: p.waitingForApproval ? "waiting" : "running"
      }
    }));
  }

  async onWorkflowComplete(
    workflowName: string,
    workflowId: string,
    result?: unknown
  ): Promise<void> {
    console.log(`Complete: ${workflowName}/${workflowId}`, result);

    this.broadcast(JSON.stringify({
      type: "workflow:updated",
      workflowId,
      updates: {
        progress: {
          step: "done",
          status: "complete",
          percent: 1,
          message: "Task completed!"
        },
        status: "complete",
        result,
        waitingForApproval: false
      }
    }));
  }

  async onWorkflowError(
    workflowName: string,
    workflowId: string,
    error: string
  ): Promise<void> {
    console.log(`Error: ${workflowName}/${workflowId}`, error);

    this.broadcast(JSON.stringify({
      type: "workflow:updated",
      workflowId,
      updates: {
        progress: { step: "error", status: "error", percent: 0, message: error },
        status: "errored",
        error: { name: "WorkflowError", message: error },
        waitingForApproval: false
      }
    }));
  }
}

Workflow

src/server.ts

export class TaskProcessingWorkflow extends AgentWorkflow<
  TaskAgent,
  TaskParams
> {
  async run(event: AgentWorkflowEvent<TaskParams>, step: AgentWorkflowStep) {
    const params = event.payload;
    console.log(`Starting workflow for task: ${params.taskName}`);

    // Step 1: Validate
    await this.reportProgress({
      step: "validate",
      status: "running",
      percent: 0.1,
      message: "Validating task..."
    });

    await step.do("validate", async () => {
      await sleep(1000);
      return { valid: true };
    });

    await this.reportProgress({
      step: "validate",
      status: "complete",
      percent: 0.25,
      message: "Validation complete"
    });

    // Step 2: Process
    await this.reportProgress({
      step: "process",
      status: "running",
      percent: 0.3,
      message: "Processing task..."
    });

    const processResult = await step.do("process", async () => {
      await sleep(1500);
      return {
        processed: true,
        taskId: params.taskId,
        data: `Processed: ${params.taskName}`
      };
    });

    await this.reportProgress({
      step: "process",
      status: "complete",
      percent: 0.5,
      message: "Processing complete - awaiting approval"
    });

    // Step 3: Wait for human approval
    await this.reportProgress({
      step: "approval",
      status: "pending",
      percent: 0.5,
      message: "Waiting for approval...",
      waitingForApproval: true
    });

    // This will throw WorkflowRejectedError if rejected
    const approvalData = await this.waitForApproval<{ approvedAt: number }>(
      step,
      {
        timeout: "1 hour"
      }
    );

    await this.reportProgress({
      step: "approval",
      status: "complete",
      percent: 0.7,
      message: "Approved! Finalizing..."
    });

    // Step 4: Finalize
    await this.reportProgress({
      step: "finalize",
      status: "running",
      percent: 0.8,
      message: "Finalizing task..."
    });

    const finalResult = await step.do("finalize", async () => {
      await sleep(1000);
      return {
        ...processResult,
        finalized: true,
        approvedAt: approvalData?.approvedAt,
        completedAt: Date.now()
      };
    });

    await this.reportProgress({
      step: "finalize",
      status: "complete",
      percent: 1,
      message: "Task completed successfully!"
    });

    await step.reportComplete(finalResult);

    return finalResult;
  }
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

How It Works

1. Submit task

   The client calls agent.submitTask(taskName), which creates a new workflow and returns a WorkflowItem with the initial state.

2. Workflow runs

   The workflow executes through multiple steps: validate, process, approval, finalize. Each step reports progress via reportProgress().

3. Progress updates

   Progress updates trigger onWorkflowProgress(), which broadcasts to all connected clients in real-time.

4. Wait for approval

   The workflow pauses at waitForApproval(). The UI shows approve/reject buttons.

5. User approves/rejects

   The client calls agent.approve(workflowId) or agent.reject(workflowId). The workflow resumes or terminates.

6. Completion

   When the workflow finishes, onWorkflowComplete() fires, broadcasting the final result.

Key Concepts

Multi-Workflow State

The agent maintains workflow state in SQLite via the cf_agents_workflows tracking table:

type WorkflowItem = {
  workflowId: string;
  taskName: string;
  status: "queued" | "running" | "waiting" | "complete" | "errored";
  progress: DefaultProgress | null;
  waitingForApproval: boolean;
  result?: unknown;
  error?: string;
  createdAt: Date;
};

Tracking Table Integration

• getWorkflows() retrieves tracked workflows with pagination (default limit 50)
• Callbacks (onWorkflowProgress, onWorkflowComplete, onWorkflowError) update both the tracking table and broadcast to clients
• Metadata (like taskName) is persisted for display after page refresh

Per-Workflow Approval

Each workflow independently waits for approval:

await this.waitForApproval<{ approvedAt: number }>(step, {
  timeout: "1 hour"
});

The UI renders approve/reject buttons for each waiting workflow:

<button onClick={() => agent.call("approve", [workflowId])}>
  Approve
</button>
<button onClick={() => agent.call("reject", [workflowId, "Reason"])}>
  Reject
</button>

Running the Example

1. Install dependencies

   cd examples/workflows
   npm install
   
2. Generate types

   npm run types
   
3. Start development server

   npm run start
   
4. Try it out

   Open http://localhost:5173 and:

   1. Enter a task name and click “Start Task” (you can start multiple)
   2. Watch progress updates in real-time
   3. When a workflow reaches the approval step, approve or reject it
   4. See the workflow complete and show results
   5. Dismiss completed workflows or clear all completed at once

Workflow Lifecycle

┌─────────────────────────────────┐
│         runWorkflow()            │
└────────────┬────────────────────┘
             │
             v
┌─────────────────────────────────┐
│        QUEUED                   │
│  (in tracking table)           │
└────────────┬────────────────────┘
             │
             v
┌─────────────────────────────────┐
│        RUNNING                  │
│  reportProgress() → onWorkflowProgress() │
└────────────┬────────────────────┘
             │
             v
┌─────────────────────────────────┐
│      WAITING (Approval)          │
│  waitForApproval()             │
└────────────┬────────────────────┘
             │
      (──────┬──────)
      │       │
   approve  reject
      │       │
      v       v
┌─────────  ┌─────────┐
│COMPLETE│  │ ERRORED │
│         │  │(rejected)│
└─────────┘  └─────────┘

Related Examples

AI Chat

AI chat with tool approval (similar pattern)

Email Agent

Process emails with state management

GitHub Webhook

Handle webhooks with event storage

Workflows Guide

In-depth guide to workflow patterns