AgentWorkflow

Base class for Workflows that integrate with Agents

Overview

AgentWorkflow extends Cloudflare’s WorkflowEntrypoint to provide seamless access to the Agent that started the workflow, enabling bidirectional communication and typed RPC.

import { AgentWorkflow } 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) {
    // Access the originating Agent
    await this.agent.updateTaskStatus(event.payload.taskId, "processing");

    const result = await step.do("process", async () => {
      // Processing logic
      return { processed: true };
    });

    // Report progress
    await step.reportComplete(result);

    return result;
  }
}

Type Parameters

AgentType · Agent · default: "Agent"

The Agent class type (for typed RPC access)

Params · unknown · default: "unknown"

User-defined params passed to the workflow

ProgressType · DefaultProgress · default: "DefaultProgress"

Type for progress reporting

Env · Cloudflare.Env · default: "Cloudflare.Env"

Environment type

Properties

agent

agent · DurableObjectStub<AgentType> · required

The Agent stub for RPC calls. Provides typed access to the Agent’s methods.

// Call any public method on the Agent
await this.agent.updateStatus("processing");
const data = await this.agent.getData();

workflowId

workflowId · string · required

Workflow instance ID (from Cloudflare Workflows)

workflowName

workflowName · string · required

Workflow binding name (from environment)

Lifecycle

run()

event · AgentWorkflowEvent<Params> · required

Workflow event with user-defined params

step · AgentWorkflowStep · required

Durable step object with Agent communication methods

Main workflow implementation. Override this method with your workflow logic.

async run(
  event: AgentWorkflowEvent<Params>,
  step: AgentWorkflowStep
) {
  // Your workflow logic
  const result = await step.do("step1", async () => {
    return { data: "result" };
  });

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

Returns: Promise<unknown> - Workflow result

AgentWorkflowStep

The step parameter is a standard WorkflowStep extended with Agent communication methods:

step.reportComplete()

Report successful completion to the Agent.

result · T

Result data to send

const result = await step.do("process", async () => {
  return { status: "success", data: "result" };
});

await step.reportComplete(result);

Returns: Promise<void>

step.reportError()

Report an error to the Agent.

error · Error | string · required

Error to report

try {
  await step.do("risky", async () => {
    throw new Error("Something went wrong");
  });
} catch (err) {
  await step.reportError(err);
  throw err; // Re-throw to fail the workflow
}

Returns: Promise<void>

Errors are automatically reported if a workflow throws without explicitly calling reportError().

step.sendEvent()

Send a custom event to the Agent.

event · T · required

Event data to send

await step.sendEvent({
  type: "progress",
  percent: 0.5,
  message: "Processing..."
});

Returns: Promise<void>

step.updateAgentState()

Replace the Agent’s entire state.

state · unknown · required

New state

await step.updateAgentState({
  status: "processing",
  progress: 0.5
});

Returns: Promise<void>

step.mergeAgentState()

Merge partial state into the Agent’s state.

partialState · Record<string, unknown> · required

Partial state to merge

await step.mergeAgentState({
  progress: 0.75
  // Other state fields remain unchanged
});

Returns: Promise<void>

step.resetAgentState()

Reset the Agent’s state to initialState.

await step.resetAgentState();

Returns: Promise<void>

Protected Methods

reportProgress()

Report typed progress to the Agent.

progress · ProgressType · required

Typed progress data

protected async reportProgress(progress: ProgressType): Promise<void>

Example:

type MyProgress = { stage: string; percent: number };

class MyWorkflow extends AgentWorkflow<MyAgent, Params, MyProgress> {
  async run(event, step) {
    await this.reportProgress({ stage: "fetch", percent: 0.25 });
    // ...
    await this.reportProgress({ stage: "process", percent: 0.75 });
  }
}

broadcastToClients()

Broadcast a message to all connected WebSocket clients via the Agent.

message · unknown · required

Message to broadcast (will be JSON-stringified)

protected broadcastToClients(message: unknown): void

Example:

this.broadcastToClients({
  type: "workflow-progress",
  workflowId: this.workflowId,
  percent: 0.5
});

broadcastToClients() is non-durable and may repeat on workflow retry. Use step.sendEvent() for durable messages.

waitForApproval()

Wait for approval from the Agent.

step · AgentWorkflowStep · required

Step object

options · WaitForApprovalOptions

timeout · string

Timeout duration (e.g., “7 days”, “1 hour”)

eventType · string · default: "approval"

Event type to wait for

stepName · string · default: "wait-for-approval"

Step name for the workflow

protected async waitForApproval<T>(step, options?): Promise<T>

Example:

type ApprovalMetadata = { approvedBy: string; notes: string };

class ApprovalWorkflow extends AgentWorkflow {
  async run(event, step) {
    // Report progress before waiting
    await this.reportProgress({ stage: "awaiting-approval" });

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

      console.log(`Approved by ${approval.approvedBy}`);
      // Continue workflow
    } catch (err) {
      if (err instanceof WorkflowRejectedError) {
        console.log("Workflow rejected:", err.reason);
        throw err;
      }
    }
  }
}

Returns: Promise<T> - Approval metadata

Throws: WorkflowRejectedError if rejected

Running Workflows from Agents

runWorkflow()

Start a workflow from an Agent.

class MyAgent extends Agent {
  @callable()
  async startProcessing(taskId: string) {
    const instanceId = await this.runWorkflow("ProcessingWorkflow", {
      taskId,
      data: "input"
    });

    return { workflowId: instanceId };
  }
}

See Agent.runWorkflow() for full documentation.

approveWorkflow()

Approve a waiting workflow.

@callable()
async approve(workflowId: string) {
  await this.approveWorkflow(workflowId, {
    approvedBy: "admin",
    notes: "Looks good!"
  });
}

rejectWorkflow()

Reject a waiting workflow.

@callable()
async reject(workflowId: string, reason: string) {
  await this.rejectWorkflow(workflowId, reason);
}

Agent Callbacks

onWorkflowProgress()

Called when a workflow reports progress.

class MyAgent extends Agent {
  async onWorkflowProgress(event: WorkflowProgressCallback) {
    console.log(`Workflow ${event.workflowId} progress:`, event.progress);

    // Update UI
    this.broadcast(JSON.stringify({
      type: "workflow-progress",
      workflowId: event.workflowId,
      progress: event.progress
    }));
  }
}

onWorkflowComplete()

Called when a workflow completes successfully.

async onWorkflowComplete(event: WorkflowCompleteCallback) {
  console.log(`Workflow ${event.workflowId} completed:`, event.result);

  // Update state
  this.setState({
    ...this.state,
    lastWorkflowResult: event.result
  });
}

onWorkflowError()

Called when a workflow errors.

async onWorkflowError(event: WorkflowErrorCallback) {
  console.error(`Workflow ${event.workflowId} failed:`, event.error);

  // Notify user
  this.broadcast(JSON.stringify({
    type: "workflow-error",
    workflowId: event.workflowId,
    error: event.error
  }));
}

Full Example

// workflow.ts
import { AgentWorkflow } from "agents/workflows";
import type { TaskAgent } from "./agent";

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

type TaskProgress = {
  stage: "fetch" | "process" | "complete";
  percent: number;
};

export class TaskWorkflow extends AgentWorkflow<
  TaskAgent,
  TaskParams,
  TaskProgress
> {
  async run(
    event: AgentWorkflowEvent<TaskParams>,
    step: AgentWorkflowStep
  ) {
    const { taskId, input } = event.payload;

    // Fetch data
    await this.reportProgress({ stage: "fetch", percent: 0.25 });
    const data = await step.do("fetch", async () => {
      return await this.agent.fetchData(taskId);
    });

    // Process data
    await this.reportProgress({ stage: "process", percent: 0.5 });
    const result = await step.do("process", async () => {
      return await processData(data, input);
    });

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

    // Save result
    await this.reportProgress({ stage: "complete", percent: 1.0 });
    await step.do("save", async () => {
      await this.agent.saveResult(taskId, result);
    });

    await step.reportComplete({ taskId, result });
    return result;
  }
}

// agent.ts
import { Agent, callable } from "agents";
import type { TaskWorkflow } from "./workflow";

class TaskAgent extends Agent {
  @callable()
  async startTask(taskId: string, input: string) {
    const instanceId = await this.runWorkflow<typeof TaskWorkflow>("TaskWorkflow", {
      taskId,
      input
    });
    return { workflowId: instanceId };
  }

  async onWorkflowProgress(event: WorkflowProgressCallback) {
    console.log(`Task ${event.workflowId}:`, event.progress);
    this.broadcast(JSON.stringify({ type: "task-progress", ...event }));
  }

  async onWorkflowComplete(event: WorkflowCompleteCallback) {
    console.log(`Task ${event.workflowId} complete!`);
  }
}

Related

• Agent Class - Agent workflow methods
• Cloudflare Workflows - Workflows documentation