Agents SDK 提供内置队列系统，可调度任务异步执行。适用于后台处理、延迟操作与管理无需立即执行的工作负载。
队列系统内置于基础 Agent 类。任务存储在 SQLite 表中，按 FIFO（先进先出）顺序自动处理。
type QueueItem<T> = {
id: string; // Unique identifier for the queued task
payload: T; // Data to pass to the callback function
callback: keyof Agent; // Name of the method to call
created_at: number; // Timestamp when the task was created
retry?: RetryOptions; // Retry options for this task
};向队列添加任务以供后续执行。
async queue<T>(
callback: keyof this,
payload: T,
options?: { retry?: RetryOptions }
): Promise<string>参数：
callback- 处理任务时要调用的方法名payload- 传给回调方法的数据options- 可选配置：retry- 回调执行的重试选项。若回调抛出异常，将按指数退避重试。详见 重试 中的RetryOptions
返回值： 已入队任务的唯一 ID
示例：
class MyAgent extends Agent {
async processEmail(data) {
// Process the email
console.log(`Processing email: ${data.subject}`);
}
async onMessage(message) {
// Queue an email processing task
const taskId = await this.queue("processEmail", {
email: "[email protected]",
subject: "Welcome!",
});
console.log(`Queued task with ID: ${taskId}`);
}
}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: "[email protected]",
subject: "Welcome!",
});
console.log(`Queued task with ID: ${taskId}`);
}
}按 ID 从队列移除指定任务。此方法是同步的。
dequeue(id: string): void参数：
id- 要移除的任务 ID
示例：
// Remove a specific task
agent.dequeue("abc123def");// Remove a specific task
agent.dequeue("abc123def");移除队列中所有任务。此方法是同步的。
dequeueAll(): void示例：
// Clear the entire queue
agent.dequeueAll();// Clear the entire queue
agent.dequeueAll();移除匹配指定回调方法的所有任务。此方法是同步的。
dequeueAllByCallback(callback: string): void参数：
callback- 回调方法名
示例：
// Remove all email processing tasks
agent.dequeueAllByCallback("processEmail");// Remove all email processing tasks
agent.dequeueAllByCallback("processEmail");按 ID 获取指定已入队任务。此方法是同步的。
getQueue<T>(id: string): QueueItem<T> | undefined参数：
id- 要获取的任务 ID
返回值： 解析 payload 后的 QueueItem，未找到则返回 undefined
payload 在返回前会自动从 JSON 解析。
示例：
const task = agent.getQueue("abc123def");
if (task) {
console.log(`Task callback: ${task.callback}`);
console.log(`Task payload:`, task.payload);
}const task = agent.getQueue("abc123def");
if (task) {
console.log(`Task callback: ${task.callback}`);
console.log(`Task payload:`, task.payload);
}获取 payload 中指定键值对匹配的所有已入队任务。此方法是同步的。
getQueues<T>(key: string, value: string): QueueItem<T>[]参数：
key- payload 中用于过滤的键value- 要匹配的值
返回值： 匹配的 QueueItem 数组
此方法获取所有队列项并在内存中解析各 payload，检查指定键是否匹配给定值。
示例：
// Find all tasks for a specific user
const userTasks = agent.getQueues("userId", "12345");// Find all tasks for a specific user
const userTasks = agent.getQueues("userId", "12345");- 校验：调用
queue()时，方法会校验回调在 Agent 上是否存在且为函数。 - 自动处理：入队后系统自动尝试刷新队列。
- FIFO 顺序：任务按创建顺序（
created_at时间戳）处理。 - 上下文保留：每个已入队任务在与发起时相同的 Agent 上下文（connection、request、email）中运行。
- 自动出队：成功执行的任务自动从队列移除。
- 错误处理：若执行时回调方法不存在，会记录错误并跳过该任务。
- 持久化：任务存储在
cf_agents_queuesSQL 表中，Agent 重启后仍保留。
为已入队任务定义回调方法时，须遵循以下签名：
async callbackMethod(payload: unknown, queueItem: QueueItem): Promise<void>示例：
class MyAgent extends Agent {
async sendNotification(payload, queueItem) {
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) {
// Queue a welcome notification
await this.queue("sendNotification", {
userId: userData.id,
message: "Welcome to our platform!",
});
}
}class MyAgent extends Agent {
async sendNotification(
payload: { userId: string; message: string },
queueItem: QueueItem<{ userId: string; message: 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!",
});
}
}class DataProcessor extends Agent {
async processLargeDataset(data) {
const results = await this.heavyComputation(data.datasetId);
await this.notifyUser(data.userId, results);
}
async onDataUpload(uploadData) {
// 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" };
}
}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" };
}
}class BatchProcessor extends Agent {
async processBatch(data) {
for (const item of data.items) {
await this.processItem(item);
}
console.log(`Completed batch ${data.batchId}`);
}
async onLargeRequest(items) {
// 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}`,
});
}
}
}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}`,
});
}
}
}使用内置 retry 选项，而非手动重新入队逻辑。回调抛出异常时，任务将按指数退避自动重试：
class RobustAgent extends Agent {
async reliableTask(payload, queueItem) {
console.log(`Processing task ${queueItem.id}`);
const response = await fetch(payload.url);
if (!response.ok) {
throw new Error(`Request failed: ${response.status}`);
}
}
async onMessage(connection, message) {
await this.queue(
"reliableTask",
{ url: "https://api.example.com/data" },
{
retry: {
maxAttempts: 5,
baseDelayMs: 500,
maxDelayMs: 10_000,
},
},
);
}
}class RobustAgent extends Agent {
async reliableTask(payload: { url: string }, queueItem: QueueItem) {
console.log(`Processing task ${queueItem.id}`);
const response = await fetch(payload.url);
if (!response.ok) {
throw new Error(`Request failed: ${response.status}`);
}
}
async onMessage(connection: Connection, message: WSMessage) {
await this.queue(
"reliableTask",
{ url: "https://api.example.com/data" },
{
retry: {
maxAttempts: 5,
baseDelayMs: 500,
maxDelayMs: 10_000,
},
},
);
}
}若未提供 retry 选项，则使用 static options.retry 的类级默认值（3 次尝试、100ms 基础延迟、3s 最大延迟）。完整说明见 重试。
- 保持 payload 精简：payload 经 JSON 序列化后存入数据库。
- 幂等操作：设计可安全重试的回调方法。
- 错误处理：在回调方法中包含适当的错误处理。
- 监控：使用日志跟踪队列处理。
- 清理：按需定期清理已完成或失败的任务。
Queue 系统可与 Agent SDK 的其他功能配合使用：
- 状态管理：在已入队回调中访问 Agent 状态。
- 调度：与
schedule()结合实现基于时间的队列处理。 - 上下文：已入队任务保留原始请求上下文。
- 数据库：与其他 Agent 数据共用同一数据库。
- 任务按顺序处理，非并行。
- 无优先级系统（仅 FIFO）。
- 队列处理发生在 Agent 执行期间，而非独立后台作业。
需要任务尽快按序执行时使用 queue。需要在特定时间或周期性运行时使用 schedule。
| 特性 | Queue | Schedule |
|---|---|---|
| 执行时机 | 立即（FIFO） | 指定时间或 cron |
| 用例 | 后台处理 | 延迟或周期性任务 |
| 存储 | cf_agents_queues 表 |
cf_agents_schedules 表 |
Agents API
Agents SDK 完整 API 参考。
调度任务
使用 cron 与延迟的基于时间的执行。
运行 Workflows
Durable 多步后台处理。