Scheduled workflows

Scheduled workflows run automatically at specified times using cron expressions. Perfect for recurring tasks like daily reports, data syncs, or cleanup jobs.

Defining scheduled workflows

There are two ways to schedule workflows:

1. Inline cron string

Use a cron expression directly in the decorator:

from polos import workflow, WorkflowContext

@workflow(
    id="daily-report",
    schedule="0 9 * * *"  # Every day at 9:00 AM UTC
)
async def daily_report(ctx: WorkflowContext, payload: SchedulePayload):
    print(f"Report generated at: {payload.timestamp}")

    # Generate and send report
    data = await ctx.step.run("fetch_data", fetch_daily_data)
    report = await ctx.step.run("generate", create_report, data)
    await ctx.step.run("send", email_report, report)

    return {"status": "completed"}

import { defineWorkflow } from '@polos/sdk';
import type { SchedulePayload } from '@polos/sdk';

const dailyReport = defineWorkflow<SchedulePayload, {}, { status: string }>(
  {
    id: 'daily-report',
    schedule: '0 9 * * *', // Every day at 9:00 AM UTC
  },
  async (ctx, payload) => {
    console.log(`Report generated at: ${payload.timestamp}`);

    // Generate and send report
    const data = await ctx.step.run('fetch_data', () => fetchDailyData());
    const report = await ctx.step.run('generate', () => createReport(data));
    await ctx.step.run('send', () => emailReport(report));

    return { status: 'completed' };
  }
);

You can also specify timezone in the cron schedule:

@workflow(
    id="end-of-day-sync",
    schedule={
        "cron": "0 17 * * *",  # 5:00 PM
        "timezone": "America/New_York"
    }
)
async def end_of_day_sync(ctx: WorkflowContext, payload: SchedulePayload):
    print(f"Syncing at {payload.timestamp} Eastern Time")

    await ctx.step.run("sync_data", sync_to_warehouse)
    return {"status": "synced"}

const endOfDaySync = defineWorkflow<SchedulePayload, {}, { status: string }>(
  {
    id: 'end-of-day-sync',
    schedule: { cron: '0 17 * * *', timezone: 'America/New_York' }, // 5:00 PM
  },
  async (ctx, payload) => {
    console.log(`Syncing at ${payload.timestamp} Eastern Time`);

    await ctx.step.run('sync_data', () => syncToWarehouse());
    return { status: 'synced' };
  }
);

2. Create schedule via API

Mark a workflow as schedulable, then create one or more schedules for it via API: This is useful for creating separate schedules per user or entity.

# Define workflow as schedulable
@workflow(
    id="user-reminder",
    schedule=True  # Can be scheduled via API
)
async def user_reminder(ctx: WorkflowContext, payload: SchedulePayload):
    user_id = payload.key  # Schedule key (e.g., user ID)
    print(f"Reminder for user {user_id} at {payload.timestamp}")

    await ctx.step.run("send_reminder", send_user_reminder, user_id)
    return {"status": "sent"}

// Define workflow as schedulable
const userReminder = defineWorkflow<SchedulePayload, {}, { status: string }>(
  {
    id: 'user-reminder',
    schedule: true, // Can be scheduled via API
  },
  async (ctx, payload) => {
    const userId = payload.key; // Schedule key (e.g., user ID)
    console.log(`Reminder for user ${userId} at ${payload.timestamp}`);

    await ctx.step.run('send_reminder', () => sendUserReminder(userId));
    return { status: 'sent' };
  }
);

Create schedule via API:

import asyncio
from polos import schedules, PolosClient

async def schedule_user_reminders(client):
    # Schedule per-user reminder
    schedule_id = await schedules.create(
        client,
        workflow="user-reminder",
        cron="0 8 * * *",  # 8:00 AM daily
        timezone="America/Los_Angeles",
        key="user_123"  # Per-user schedule
    )

    print(f"Created schedule: {schedule_id}")

client = PolosClient()
asyncio.run(schedule_user_reminders(client))

import { PolosClient } from '@polos/sdk';

const client = PolosClient.fromEnv();

// Schedule per-user reminder
const scheduleId = await client.schedules.create(
  'user-reminder',
  '0 8 * * *',          // 8:00 AM daily
  'America/Los_Angeles',
  'user_123',           // Per-user schedule
);

console.log(`Created schedule: ${scheduleId}`);

Schedule payload

Scheduled workflows receive a special SchedulePayload with timing information:

from datetime import datetime
from pydantic import BaseModel

class SchedulePayload(BaseModel):
    timestamp: datetime          # When this workflow was scheduled to run
    last_timestamp: datetime | None  # When this schedule last ran (None if first run)
    timezone: str                # Timezone of the schedule
    schedule_id: str             # Unique identifier for this schedule
    key: str                     # User ID or custom identifier
    upcoming: datetime           # Next scheduled run time

@workflow(id="scheduled-task", schedule="0 */6 * * *")
async def scheduled_task(ctx: WorkflowContext, payload: SchedulePayload):
    # Access schedule information
    print(f"Current run: {payload.timestamp}")
    print(f"Last run: {payload.last_timestamp}")
    print(f"Next run: {payload.upcoming}")
    print(f"Schedule ID: {payload.schedule_id}")

    return {"executed_at": payload.timestamp}

const scheduledTask = defineWorkflow<SchedulePayload, {}, { executedAt: Date }>(
  {
    id: 'scheduled-task',
    schedule: '0 */6 * * *',
  },
  async (ctx, payload) => {
    // Access schedule information
    console.log(`Current run: ${payload.timestamp}`);
    console.log(`Last run: ${payload.lastTimestamp}`);
    console.log(`Next run: ${payload.upcoming}`);
    console.log(`Schedule ID: ${payload.scheduleId}`);

    return { executedAt: payload.timestamp };
  }
);

Cron expressions

Cron format: minute hour day month day_of_week Common patterns:

# Every hour
schedule="0 * * * *"

# Every day at 9 AM
schedule="0 9 * * *"

# Every Monday at 8 AM
schedule="0 8 * * 1"

# Every 15 minutes
schedule="*/15 * * * *"

# First day of every month at midnight
schedule="0 0 1 * *"

# Weekdays at 5 PM
schedule="0 17 * * 1-5"

# Every 6 hours
schedule="0 */6 * * *"

// Every hour
{ schedule: '0 * * * *' }

// Every day at 9 AM
{ schedule: '0 9 * * *' }

// Every Monday at 8 AM
{ schedule: '0 8 * * 1' }

// Every 15 minutes
{ schedule: '*/15 * * * *' }

// First day of every month at midnight
{ schedule: '0 0 1 * *' }

// Weekdays at 5 PM
{ schedule: '0 17 * * 1-5' }

// Every 6 hours
{ schedule: '0 */6 * * *' }

Cron syntax:

* - Any value
*/n - Every n units
n-m - Range from n to m
n,m - List of values

Field ranges:

Minute: 0-59
Hour: 0-23
Day of month: 1-31
Month: 1-12
Day of week: 0-6 (Sunday = 0)

Use appropriate cron intervals

# ✅ GOOD: Reasonable intervals
schedule="0 */6 * * *"  # Every 6 hours
schedule="0 9 * * *"    # Daily at 9 AM

# ❌ BAD: Too frequent (use events or queues instead)
schedule="* * * * *"    # Every minute

// ✅ GOOD: Reasonable intervals
{ schedule: '0 */6 * * *' }  // Every 6 hours
{ schedule: '0 9 * * *' }    // Daily at 9 AM

// ❌ BAD: Too frequent (use events or queues instead)
{ schedule: '* * * * *' }    // Every minute

Per-user schedules

Use the key parameter to create separate schedules for each user or entity:

from polos import schedules, PolosClient

async def setup_user_schedule(client, user_id: str, preferences: dict):
    # Create/update per-user schedule
    schedule_id = await schedules.create(
        client,
        workflow="daily-digest",
        cron="0 8 * * *",  # 8 AM
        timezone=preferences["timezone"],
        key=user_id  # Unique per user
    )

    return schedule_id

client = PolosClient()

# Each user gets their own schedule
await setup_user_schedule(client, "user_123", {"timezone": "America/New_York"})
await setup_user_schedule(client, "user_456", {"timezone": "Europe/London"})

import { PolosClient } from '@polos/sdk';

const client = PolosClient.fromEnv();

async function setupUserSchedule(
  client: PolosClient,
  userId: string,
  preferences: { timezone: string }
) {
  // Create/update per-user schedule
  const scheduleId = await client.schedules.create(
    'daily-digest',
    '0 8 * * *',             // 8 AM
    preferences.timezone,
    userId,                   // Unique per user
  );

  return scheduleId;
}

// Each user gets their own schedule
await setupUserSchedule(client, 'user_123', { timezone: 'America/New_York' });
await setupUserSchedule(client, 'user_456', { timezone: 'Europe/London' });

In the workflow:

@workflow(id="daily-digest", schedule=True)
async def daily_digest(ctx: WorkflowContext, payload: SchedulePayload):
    user_id = payload.key  # Get user from schedule key

    # Fetch user-specific data
    user = await ctx.step.run("get_user", get_user, user_id)
    digest = await ctx.step.run("generate", create_digest, user)
    await ctx.step.run("send", send_email, user.email, digest)

    return {"user_id": user_id, "sent": True}

const dailyDigest = defineWorkflow<SchedulePayload, {}, { userId: string; sent: boolean }>(
  {
    id: 'daily-digest',
    schedule: true,
  },
  async (ctx, payload) => {
    const userId = payload.key; // Get user from schedule key

    // Fetch user-specific data
    const user = await ctx.step.run('get_user', () => getUser(userId));
    const digest = await ctx.step.run('generate', () => createDigest(user));
    await ctx.step.run('send', () => sendEmail(user.email, digest));

    return { userId, sent: true };
  }
);

Key behavior:

If a schedule with the same workflow and key exists, it’s updated (not duplicated)
Use key="global" (default) for system-wide schedules
Use key=user_id for per-user schedules

Timezones

Always specify timezones for scheduled workflows to avoid confusion:

@workflow(
    id="morning-report",
    schedule={
        "cron": "0 9 * * *",
        "timezone": "America/New_York"  # Runs at 9 AM Eastern
    }
)
async def morning_report(ctx: WorkflowContext, payload: SchedulePayload):
    # Runs at 9 AM Eastern time, adjusts for DST automatically
    return {"executed_at": payload.timestamp}

const morningReport = defineWorkflow<SchedulePayload, {}, { executedAt: Date }>(
  {
    id: 'morning-report',
    schedule: { cron: '0 9 * * *', timezone: 'America/New_York' }, // Runs at 9 AM Eastern
  },
  async (ctx, payload) => {
    // Runs at 9 AM Eastern time, adjusts for DST automatically
    return { executedAt: payload.timestamp };
  }
);

Common timezones:

"UTC" - Coordinated Universal Time
"America/New_York" - US Eastern
"America/Los_Angeles" - US Pacific
"Europe/London" - UK
"Asia/Tokyo" - Japan
"Australia/Sydney" - Australia

Managing schedules via API

Create or update a schedule

from polos import schedules, PolosClient

client = PolosClient()

# Global schedule (one per workflow)
schedule_id = await schedules.create(
    client,
    workflow="cleanup-task",
    cron="0 3 * * *",  # 3 AM daily
    timezone="UTC",
    key="global"
)

# Per-user schedule
schedule_id = await schedules.create(
    client,
    workflow="user-notification",
    cron="0 12 * * *",  # Noon daily
    timezone="America/Chicago",
    key="user_789"
)

import { PolosClient } from '@polos/sdk';

const client = PolosClient.fromEnv();

// Global schedule (one per workflow)
const globalScheduleId = await client.schedules.create(
  'cleanup-task',
  '0 3 * * *',   // 3 AM daily
  'UTC',
  'global',
);

// Per-user schedule
const userScheduleId = await client.schedules.create(
  'user-notification',
  '0 12 * * *',          // Noon daily
  'America/Chicago',
  'user_789',
);

Update existing schedule:

# Calling create() with the same workflow and key updates it
schedule_id = await schedules.create(
    client,
    workflow="user-notification",
    cron="0 15 * * *",  # Change to 3 PM
    timezone="America/Chicago",
    key="user_789"  # Same key = update
)

// Calling create() with the same workflow and key updates it
const scheduleId = await client.schedules.create(
  'user-notification',
  '0 15 * * *',          // Change to 3 PM
  'America/Chicago',
  'user_789',            // Same key = update
);

Introduction

Getting Started

Fundamentals

Agents

Workflows

Observability

Guides and Examples

Community

Defining scheduled workflows

1. Inline cron string

2. Create schedule via API

Schedule payload

Cron expressions

Use appropriate cron intervals

Per-user schedules

Timezones

Managing schedules via API

Create or update a schedule

​Defining scheduled workflows

​1. Inline cron string

​2. Create schedule via API

​Schedule payload

​Cron expressions

​Use appropriate cron intervals

​Per-user schedules

​Timezones

​Managing schedules via API

​Create or update a schedule

Defining scheduled workflows

1. Inline cron string

2. Create schedule via API

Schedule payload

Cron expressions

Use appropriate cron intervals

Per-user schedules

Timezones

Managing schedules via API

Create or update a schedule