Scheduled Workflows

Schedule workflows to run automatically using cron expressions.

Basic Scheduling

Add a schedule parameter to your workflow decorator:

from bifrost import workflow

@workflow(schedule="0 9 * * *")
async def daily_report():
    """Generate daily report at 9 AM UTC."""
    return {"status": "report_sent"}

Cron Syntax

Standard 5-field cron format: minute hour day month weekday

Field	Values	Description
Minute	0-59	Minute of the hour
Hour	0-23	Hour of the day (UTC)
Day	1-31	Day of the month
Month	1-12	Month of the year
Weekday	0-6	Day of week (0=Sunday)

Common Patterns

# Every hour at minute 0
@workflow(schedule="0 * * * *")

# Every day at 9 AM UTC
@workflow(schedule="0 9 * * *")

# Every Monday at 8 AM UTC
@workflow(schedule="0 8 * * 1")

# First day of every month at midnight
@workflow(schedule="0 0 1 * *")

# Every 15 minutes
@workflow(schedule="*/15 * * * *")

# Weekdays at 6 PM UTC
@workflow(schedule="0 18 * * 1-5")

Scheduled Workflow Example

from bifrost import workflow, config
import httpx
import logging

logger = logging.getLogger(__name__)

@workflow(
    schedule="0 8 * * 1-5",  # Weekdays at 8 AM UTC
    timeout_seconds=300,
    category="Reports"
)
async def send_daily_summary():
    """Send daily summary email to team."""
    logger.info("Starting daily summary")

    # Gather metrics
    metrics = await gather_metrics()

    # Send notification
    webhook_url = await config.get("slack_webhook")
    async with httpx.AsyncClient() as client:
        await client.post(webhook_url, json={
            "text": f"Daily Summary: {metrics['total_tasks']} tasks completed"
        })

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

How Scheduling Works

1. Discovery: Scheduler finds workflows with schedule parameter
2. Registration: Jobs are registered with next run time
3. Execution: At scheduled time, workflow runs in global scope
4. Logging: Results stored in execution history

Note

Scheduled workflows run in global scope (no specific organization). Access all orgs or specify one explicitly if needed.

Execution Context

Scheduled workflows have limited context:

from bifrost import workflow, context

@workflow(schedule="0 0 * * *")
async def scheduled_task():
    # context.org_id is None (global scope)
    # context.user_id is "system" or scheduler user

    # To process all orgs:
    for org in await get_all_orgs():
        await process_org(org.id)

Monitoring Scheduled Jobs

• View scheduled jobs in Workflows → filter by scheduled
• Execution history shows scheduled runs
• Failed runs appear in execution logs

Disabling a Schedule

Remove or comment the schedule parameter:

# Temporarily disabled
@workflow(
    # schedule="0 9 * * *",  # Disabled
    category="Reports"
)
async def daily_report():
    ...

Or redeploy without the schedule.

Tips

• Use UTC: All schedules are in UTC timezone
• Avoid overlaps: Don’t schedule faster than execution time
• Set timeouts: Prevent long-running scheduled jobs
• Log progress: Track execution for debugging

See Also

• Decorators Reference - All decorator options
• Writing Workflows - Workflow basics