Scheduling Work with libfx
An fx agent keeps each session in a durable log, so a scheduled task can come back to the conversation that created it. The tool stores the session ID in the schedule's payload, and the consumer prompts that session when the schedule fires.
Each tool call receives the session's ID and the context you pass to agent.session(). With idempotent: true, a call interrupted by a crash runs again. executionId stays the same when it does, so use it in the schedule name: the rerun finds the existing schedule instead of creating a second one.
import { createFxAgent } from 'libfx';
import { Schedules, SchedulesApiError } from '@vercel/schedules';
export const agent = createFxAgent({
model: 'anthropic/claude-sonnet-5.5',
tools: [
{
name: 'schedule_task',
description: 'Schedule work to run later in this conversation',
idempotent: true,
inputSchema: {
type: 'object',
properties: {
prompt: { type: 'string', description: 'What to do when the task fires' },
cron: { type: 'string', description: 'Cron expression, for recurring tasks' },
at: { type: 'string', description: 'Local time YYYY-MM-DDTHH:mm, for one-time tasks' },
},
required: ['prompt'],
},
execute: async ({ prompt, cron, at }, { executionId, sessionId, context }) => {
const { userId, timezone } = context as { userId: string; timezone: string };
try {
await Schedules.create({
name: `task-${executionId}`,
namespace: userId,
expression: cron ? { type: 'cron', cron } : { type: 'single', at: at! },
timezone,
target: { topic: 'agent-tasks' },
payload: { sessionId, prompt, userId, timezone },
});
} catch (error) {
// A rerun after a crash finds the schedule the first attempt created.
if (!(error instanceof SchedulesApiError && error.code === 'schedule_already_exists')) {
throw error;
}
}
return 'Scheduled.';
},
},
],
});Open each session with the user's context, so the tool knows who it's scheduling for:
agent.session(sessionId, { context: { userId, timezone } }).prompt(text);When the schedule fires, prompt the session it came from. Pass the run's executionId as the messageId, so a redelivered message reaches the same turn instead of starting another:
import { handleCallback } from '@vercel/queue';
import type { ScheduleQueueMessage } from '@vercel/schedules';
import { agent } from '../lib/agent';
type TaskMessage = ScheduleQueueMessage<{
sessionId: string;
prompt: string;
userId: string;
timezone: string;
}>;
export const POST = handleCallback<TaskMessage>(async ({ executionId, payload }) => {
if (!payload) return;
const { sessionId, prompt, userId, timezone } = payload;
const turn = agent
.session(sessionId, { context: { userId, timezone } })
.prompt(`Scheduled task: ${prompt}`, { messageId: `task-${executionId}` });
await turn.result;
});Subscribe the consumer to the topic, next to libfx's own queue route:
{
"functions": {
"api/agent-tasks.ts": {
"experimentalTriggers": [{ "type": "queue/v2beta", "topic": "agent-tasks" }]
}
}
}The scheduled turn is part of the session, so it appears when the user reads the conversation from session.stream().
Was this helpful?