Agent

Base class for creating Agent implementations with state, RPC, scheduling, and email support

Overview

The Agent class is the core building block for creating stateful agents on Cloudflare Workers. It extends PartyServer to provide WebSocket connections, state management, RPC methods, SQL storage, scheduling, email routing, MCP client support, and workflow integration.

import { Agent } from "agents";

class MyAgent extends Agent<Env, State> {
  initialState = { count: 0 };

  @callable()
  async increment() {
    this.setState({ count: this.state.count + 1 });
    return this.state.count;
  }
}

Type Parameters

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

Environment type containing bindings (KV, D1, R2, etc.)

State · unknown · default: "unknown"

State type to store within the Agent

Props · Record<string, unknown> · default: "Record<string, unknown>"

Props type passed to the Agent on creation

Properties

state

state · State · required

Current state of the Agent. Read-only. Use setState() to update.

const count = this.state.count;

initialState

initialState · State

Initial state for the Agent. Override to provide default state values.

class MyAgent extends Agent<Env, { count: number }> {
  initialState = { count: 0 };
}

name

name · string · required

The unique name/ID of this Agent instance (inherited from PartyServer).

env

env · Env · required

The environment bindings for this Agent (KV, D1, R2, etc.).

ctx

ctx · AgentContext · required

The Durable Object context (storage, waitUntil, etc.).

mcp

mcp · MCPClientManager · required

MCP client manager for connecting to external MCP servers.

await this.mcp.registerServer(id, {
  url: "https://mcp-server.example.com",
  name: "My MCP Server"
});

observability

observability · Observability

Observability implementation for emitting events. Defaults to genericObservability.

Static Options

options

options · AgentStaticOptions

Static configuration options for the Agent class. Override in subclasses.

class SecureAgent extends Agent {
  static options = {
    hibernate: true,
    sendIdentityOnConnect: false,
    hungScheduleTimeoutSeconds: 60,
    retry: {
      maxAttempts: 5,
      baseDelayMs: 200,
      maxDelayMs: 5000
    }
  };
}

Show AgentStaticOptions fields

hibernate · boolean · default: "true"

Whether the Agent should hibernate when inactive

sendIdentityOnConnect · boolean · default: "true"

Whether to send identity (name, agent) to clients on connect

hungScheduleTimeoutSeconds · number · default: "30"

Timeout in seconds before a running interval schedule is considered “hung” and force-reset

retry · RetryOptions

Default retry options for schedule(), queue(), and this.retry()

Show RetryOptions fields

maxAttempts · number · default: "3"

Maximum number of retry attempts

baseDelayMs · number · default: "100"

Base delay in milliseconds for exponential backoff

maxDelayMs · number · default: "3000"

Maximum delay cap in milliseconds

Methods

setState()

state · State · required

New state to set

Update the Agent’s state. Persists to storage and broadcasts to all connected clients.

this.setState({ count: this.state.count + 1 });

Throws an error if called from a readonly connection context.

sql()

query · TemplateStringsArray · required

SQL query template strings

values · (string | number | boolean | null)[]

Values to be inserted into the query

Execute SQL queries against the Agent’s database.

const users = this.sql<{ id: number; name: string }>`
  SELECT * FROM users WHERE id = ${userId}
`;

Returns: T[] - Array of query results

Throws: SqlError - If the query fails

schedule()

Schedule a callback to run at a future time or on a recurring interval.

callback · keyof this · required

Name of the method to call

options · ScheduleOptions · required

Scheduling options

Show ScheduleOptions variants

At a specific time:

time · Date · required

Date/time to execute

payload · T

Data to pass to the callback

retry · RetryOptions

Retry options for this specific schedule

After a delay:

delayInSeconds · number · required

Number of seconds to delay

Cron schedule:

cron · string · required

Cron expression (e.g., “0 0 * * *”)

Interval:

intervalSeconds · number · required

Number of seconds between executions

// One-time scheduled task
await this.schedule("sendReminder", {
  time: new Date(Date.now() + 3600000),
  payload: { userId: "123" }
});

// Recurring cron task
await this.schedule("dailyBackup", {
  cron: "0 0 * * *",
  payload: { type: "full" }
});

// Interval task
await this.schedule("healthCheck", {
  intervalSeconds: 300
});

Returns: Promise<string> - Schedule ID

queue()

Queue a callback for asynchronous execution.

callback · keyof this · required

Name of the method to call

payload · T

Data to pass to the callback

retry · RetryOptions

Retry options for this specific queue item

await this.queue("processUpload", {
  fileId: "abc123",
  userId: "user-456"
});

Returns: Promise<void>

retry()

Retry an async operation with exponential backoff and jitter.

fn · (attempt: number) => Promise<T> · required

The async function to retry. Receives the current attempt number (1-indexed).

options · RetryOptions

Retry configuration (falls back to static options)

shouldRetry · (err: unknown, nextAttempt: number) => boolean

Predicate to determine if an error should be retried. Return false to stop immediately.

const result = await this.retry(
  async (attempt) => {
    return await fetchExternalAPI();
  },
  {
    maxAttempts: 5,
    shouldRetry: (err) => err instanceof NetworkError
  }
);

Returns: Promise<T> - The result of fn on success

Throws: The last error if all attempts fail or shouldRetry returns false

replyToEmail()

Reply to an email received via routeAgentEmail().

email · AgentEmail · required

The email to reply to

options · ReplyOptions · required

fromName · string · required

Sender name

subject · string

Email subject (defaults to “Re: original subject”)

body · string · required

Email body

contentType · string · default: "text/plain"

MIME content type

headers · Record<string, string>

Additional headers

secret · string | null

Secret for signing agent headers (enables secure reply routing). Required if the email was routed via createSecureReplyEmailResolver.

await this.replyToEmail(email, {
  fromName: "Support Team",
  body: "Thank you for your message!",
  secret: this.env.EMAIL_SECRET
});

runWorkflow()

Run a Workflow and track its execution.

workflowName · string · required

Name of the Workflow binding in env

params · Params · required

Parameters to pass to the workflow

options · RunWorkflowOptions

id · string

Unique workflow instance ID (auto-generated if not provided)

metadata · Record<string, unknown>

Custom metadata to store with the workflow

const instanceId = await this.runWorkflow("ProcessingWorkflow", {
  taskId: "task-123",
  data: "input data"
});

Returns: Promise<string> - Workflow instance ID

getWorkflows()

Query tracked workflows.

criteria · WorkflowQueryCriteria

workflowName · string

Filter by workflow binding name

status · WorkflowStatus

Filter by status (“queued”, “running”, “complete”, “errored”, etc.)

limit · number · default: "100"

Maximum number of results

offset · number · default: "0"

Number of results to skip

const page = await this.getWorkflows({
  status: "running",
  limit: 10
});

Returns: Promise<WorkflowPage>

approveWorkflow()

Approve a workflow waiting for approval.

workflowId · string · required

Workflow instance ID

metadata · T

Metadata to pass to the workflow

await this.approveWorkflow(instanceId, { approvedBy: "admin" });

rejectWorkflow()

Reject a workflow waiting for approval.

workflowId · string · required

Workflow instance ID

reason · string

Reason for rejection

await this.rejectWorkflow(instanceId, "Insufficient permissions");

Lifecycle Hooks

onConnect()

connection · Connection · required

The new WebSocket connection

ctx · ConnectionContext · required

Connection context (includes the upgrade request)

Called when a new WebSocket connection is established.

async onConnect(connection: Connection, ctx: ConnectionContext) {
  const userId = new URL(ctx.request.url).searchParams.get("user");
  connection.setState({ userId });
}

onMessage()

connection · Connection · required

The connection that sent the message

message · string | ArrayBuffer · required

The message data

Called when a WebSocket message is received.

async onMessage(connection: Connection, message: string | ArrayBuffer) {
  if (typeof message === "string") {
    const data = JSON.parse(message);
    // Handle custom message
  }
}

onClose()

connection · Connection · required

The connection that closed

code · number · required

WebSocket close code

reason · string · required

Close reason

wasClean · boolean · required

Whether the close was clean

Called when a WebSocket connection closes.

async onClose(connection: Connection, code: number, reason: string) {
  console.log(`Connection ${connection.id} closed: ${reason}`);
}

onRequest()

request · Request · required

The HTTP request

Called when an HTTP request is received.

async onRequest(request: Request) {
  if (request.method === "POST") {
    const data = await request.json();
    return new Response(JSON.stringify({ status: "ok" }));
  }
  return new Response("Method not allowed", { status: 405 });
}

Returns: Response | Promise<Response>

onStart()

props · Props

Props passed to the Agent on creation

Called when the Agent is created or wakes from hibernation.

async onStart(props?: Props) {
  // Initialize resources, restore state, etc.
}

onEmail()

email · AgentEmail · required

The incoming email message

Called when an email is routed to this Agent via routeAgentEmail().

async onEmail(email: AgentEmail) {
  const subject = email.headers.get("subject");
  await this.replyToEmail(email, {
    fromName: "Bot",
    body: `Received: ${subject}`,
    secret: this.env.EMAIL_SECRET
  });
}

onStateChanged()

state · State | undefined · required

The new state

source · Connection | 'server' · required

Source of the state update

Called after state has been persisted and broadcast. This is a notification hook—errors are routed to onError and do not affect persistence.

async onStateChanged(state: State, source: Connection | "server") {
  // Log state changes, trigger side effects, etc.
}

validateStateChange()

nextState · State · required

The proposed new state

source · Connection | 'server' · required

Source of the state update

Called before state is persisted. Throw an error to reject the update. Must be synchronous.

validateStateChange(nextState: State, source: Connection | "server") {
  if (source !== "server" && nextState.adminOnly) {
    throw new Error("Only server can set adminOnly fields");
  }
}

onWorkflowProgress()

event · WorkflowProgressCallback · required

Progress event from the workflow

Called when a tracked workflow reports progress.

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

onWorkflowComplete()

event · WorkflowCompleteCallback · required

Completion event from the workflow

Called when a tracked workflow completes.

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

onWorkflowError()

event · WorkflowErrorCallback · required

Error event from the workflow

Called when a tracked workflow errors.

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

onError()

Called when an error occurs. Override to customize error handling.

async onError(error: unknown) {
  console.error("Agent error:", error);
  // Don't throw to suppress the error, or re-throw to propagate
  throw error;
}

Connection Management

getConnections()

Get all active WebSocket connections.

const connections = this.getConnections();
for (const conn of connections) {
  conn.send("broadcast message");
}

Returns: Iterable<Connection>

broadcast()

message · string | ArrayBuffer · required

Message to broadcast

exclude · string[]

Connection IDs to exclude

Broadcast a message to all connected clients (optionally excluding some).

this.broadcast(JSON.stringify({ event: "update", data }), [sourceConnectionId]);

setConnectionReadonly()

connection · Connection · required

The connection to mark

readonly · boolean · default: "true"

Whether the connection should be readonly

Mark a connection as readonly (cannot call setState).

this.setConnectionReadonly(connection, true);

isConnectionReadonly()

connection · Connection · required

The connection to check

Check if a connection is marked as readonly.

if (this.isConnectionReadonly(connection)) {
  return new Response("Readonly connection", { status: 403 });
}

Returns: boolean

shouldConnectionBeReadonly()

connection · Connection · required

The connection being established

ctx · ConnectionContext · required

Connection context

Override to determine if a connection should be readonly on connect.

shouldConnectionBeReadonly(connection: Connection, ctx: ConnectionContext) {
  const url = new URL(ctx.request.url);
  return url.searchParams.get("readonly") === "true";
}

Returns: boolean

Related

• Routing - Route requests to Agents
• State Management - Manage Agent state
• RPC with @callable - Define callable methods
• Scheduling - Schedule recurring tasks
• Workflows - Integrate with Cloudflare Workflows
• Email Routing - Handle incoming emails