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

# Queue System

> Schedule tasks for asynchronous background processing with automatic retries and FIFO ordering

## Overview

The Agents SDK provides a built-in queue system that allows you to schedule tasks for asynchronous execution. This is particularly useful for background processing, delayed operations, and managing workloads that don't need immediate execution.

The queue system is built into the base `Agent` class. Tasks are stored in a SQLite table and processed automatically in FIFO (First In, First Out) order.

## QueueItem Type

```typescript theme={null}
export type QueueItem<T = string> = {
  id: string; // Unique identifier for the queued task
  payload: T; // Data to pass to the callback function
  callback: keyof Agent<unknown>; // Name of the method to call
  created_at: number; // Timestamp when the task was created
  retry?: RetryOptions; // Retry options (if configured)
};
```

## Core Methods

### queue()

Adds a task to the queue for future execution.

```typescript theme={null}
async queue<T = unknown>(
  callback: keyof this,
  payload: T,
  options?: { retry?: RetryOptions }
): Promise<string>
```

**Parameters:**

* `callback`: The name of the method to call when processing the task
* `payload`: Data to pass to the callback method
* `options.retry`: Optional retry configuration. See [Retries](/background/retries) for details.

**Returns:** The unique ID of the queued task

**Example:**

```typescript theme={null}
class MyAgent extends Agent {
  async processEmail(data: { email: string; subject: string }) {
    // Process the email
    console.log(`Processing email: ${data.subject}`);
  }

  async onMessage(message: string) {
    // Queue an email processing task
    const taskId = await this.queue("processEmail", {
      email: "user@example.com",
      subject: "Welcome!"
    });

    console.log(`Queued task with ID: ${taskId}`);
  }
}
```

### dequeue()

Removes a specific task from the queue by ID.

```typescript theme={null}
dequeue(id: string): void
```

**Parameters:**

* `id`: The ID of the task to remove

**Example:**

```typescript theme={null}
// Remove a specific task
agent.dequeue("abc123def");
```

### dequeueAll()

Removes all tasks from the queue.

```typescript theme={null}
dequeueAll(): void
```

**Example:**

```typescript theme={null}
// Clear the entire queue
agent.dequeueAll();
```

### dequeueAllByCallback()

Removes all tasks that match a specific callback method.

```typescript theme={null}
dequeueAllByCallback(callback: string): void
```

**Parameters:**

* `callback`: Name of the callback method

**Example:**

```typescript theme={null}
// Remove all email processing tasks
agent.dequeueAllByCallback("processEmail");
```

### getQueue()

Retrieves a specific queued task by ID.

```typescript theme={null}
getQueue(id: string): QueueItem<string> | undefined
```

**Parameters:**

* `id`: The ID of the task to retrieve

**Returns:** The QueueItem with parsed payload or undefined if not found

<Note>The payload is automatically parsed from JSON before being returned</Note>

**Example:**

```typescript theme={null}
const task = agent.getQueue("abc123def");
if (task) {
  console.log(`Task callback: ${task.callback}`);
  console.log(`Task payload:`, task.payload);
}
```

### getQueues()

Retrieves all queued tasks that match a specific key-value pair in their payload.

```typescript theme={null}
getQueues(key: string, value: string): QueueItem<string>[]
```

**Parameters:**

* `key`: The key to filter by in the payload
* `value`: The value to match

**Returns:** Array of matching QueueItem objects

<Note>This method fetches all queue items and filters them in memory by parsing each payload and checking if the specified key matches the value</Note>

**Example:**

```typescript theme={null}
// Find all tasks for a specific user
const userTasks = await agent.getQueues("userId", "12345");
```

## How Queue Processing Works

<Steps>
  <Step title="Validation">
    When calling `queue()`, the method validates that the callback exists as a function on the agent
  </Step>

  <Step title="Automatic Processing">
    After queuing, the system automatically attempts to flush the queue
  </Step>

  <Step title="FIFO Order">
    Tasks are processed in the order they were created (`created_at` timestamp)
  </Step>

  <Step title="Context Preservation">
    Each queued task runs with the same agent context (connection, request, email)
  </Step>

  <Step title="Automatic Retries">
    If a callback fails, it is retried with exponential backoff (configurable per task)
  </Step>

  <Step title="Automatic Dequeue">
    Tasks are removed from the queue after successful execution or after all retry attempts are exhausted
  </Step>

  <Step title="Error Handling">
    If a callback method does not exist at execution time, an error is logged and the task is skipped
  </Step>

  <Step title="Persistence">
    Tasks are stored in the `cf_agents_queues` table and survive agent restarts
  </Step>
</Steps>

## Queue Callback Methods

When defining callback methods for queued tasks, they must follow this signature:

```typescript theme={null}
async callbackMethod(payload: unknown, queueItem: QueueItem<string>): Promise<void>
```

**Example:**

```typescript theme={null}
class MyAgent extends Agent {
  async sendNotification(
    payload: { userId: string; message: string },
    queueItem: QueueItem<string>
  ) {
    console.log(`Processing task ${queueItem.id}`);
    console.log(
      `Sending notification to user ${payload.userId}: ${payload.message}`
    );

    // Your notification logic here
    await this.notificationService.send(payload.userId, payload.message);
  }

  async onUserSignup(userData: any) {
    // Queue a welcome notification
    await this.queue("sendNotification", {
      userId: userData.id,
      message: "Welcome to our platform!"
    });
  }
}
```

## Use Cases

### Background Processing

```typescript theme={null}
class DataProcessor extends Agent {
  async processLargeDataset(data: { datasetId: string; userId: string }) {
    const results = await this.heavyComputation(data.datasetId);
    await this.notifyUser(data.userId, results);
  }

  async onDataUpload(uploadData: any) {
    // Queue the processing instead of doing it synchronously
    await this.queue("processLargeDataset", {
      datasetId: uploadData.id,
      userId: uploadData.userId
    });

    return { message: "Data upload received, processing started" };
  }
}
```

### Delayed Operations

```typescript theme={null}
class ReminderAgent extends Agent {
  async sendReminder(data: { userId: string; message: string }) {
    await this.emailService.send(data.userId, data.message);
  }

  async scheduleReminder(userId: string, message: string, delayMs: number) {
    // Note: For true delayed execution, combine with the scheduling system
    // This example shows queueing for later processing
    await this.queue("sendReminder", { userId, message });
  }
}
```

### Batch Operations

```typescript theme={null}
class BatchProcessor extends Agent {
  async processBatch(data: { items: any[]; batchId: string }) {
    for (const item of data.items) {
      await this.processItem(item);
    }
    console.log(`Completed batch ${data.batchId}`);
  }

  async onLargeRequest(items: any[]) {
    // Split large requests into smaller batches
    const batchSize = 10;
    for (let i = 0; i < items.length; i += batchSize) {
      const batch = items.slice(i, i + batchSize);
      await this.queue("processBatch", {
        items: batch,
        batchId: `batch-${i / batchSize + 1}`
      });
    }
  }
}
```

## Best Practices

<Card title="Keep Payloads Small" icon="weight">
  Payloads are JSON-serialized and stored in the database
</Card>

<Card title="Idempotent Operations" icon="rotate">
  Design callback methods to be safe to retry
</Card>

<Card title="Error Handling" icon="shield">
  Include proper error handling in callback methods
</Card>

<Card title="Monitoring" icon="chart-line">
  Use logging to track queue processing
</Card>

<Card title="Cleanup" icon="broom">
  Regularly clean up completed or failed tasks if needed
</Card>

## Error Handling and Retries

Queued tasks are automatically retried on failure with exponential backoff. The default is 3 attempts. You can customize this per task:

```typescript theme={null}
// Retry up to 5 times with custom backoff
await this.queue("reliableTask", payload, {
  retry: { maxAttempts: 5, baseDelayMs: 500 }
});
```

If you need custom error handling in the callback:

```typescript theme={null}
class RobustAgent extends Agent {
  async reliableTask(payload: { data: string }, queueItem: QueueItem<string>) {
    try {
      await this.doSomethingRisky(payload);
    } catch (error) {
      console.error(`Task ${queueItem.id} failed:`, error);
      // The retry system will catch this error and retry automatically
      throw error;
    }
  }
}
```

See [Retries](/background/retries) for full documentation on retry options and patterns.

## Integration with Other Features

The queue system works seamlessly with other Agent SDK features:

* **State Management**: Access agent state within queued callbacks
* **Scheduling**: Combine with `schedule()` for time-based queue processing
* **Retries**: Built-in retry with exponential backoff. See [Retries](/background/retries).
* **Context**: Queued tasks maintain the original request context
* **Database**: Uses the same database as other agent data

## Limitations

<Warning>
  * Tasks are processed sequentially, not in parallel
  * No priority system (FIFO only)
  * Queue processing happens during agent execution, not as separate background jobs
  * Failed tasks are dequeued after all retry attempts are exhausted (no dead-letter queue)
</Warning>