Scheduled Workflows

Create a schedule event source to run workflows automatically using cron expressions.

Basic Scheduling

Configure the schedule on the event source, then subscribe the workflow to it:

from bifrost import workflow

# The workflow stays decorator-light; the schedule lives on the event source.

@workflow(category="Reports")
async def daily_report():
    """Generate a daily report when the subscribed schedule fires."""
    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 in the source’s configured timezone
Day	1-31	Day of the month
Month	1-12	Month of the year
Weekday	0-6	Day of week (0=Sunday)

Common Patterns

Schedule	Cron expression
Every hour at minute 0	0 * * * *
Every day at 9 AM	0 9 * * *
Every Monday at 8 AM	0 8 * * 1
First day of every month at midnight	0 0 1 * *
Every 15 minutes	*/15 * * * *
Weekdays at 6 PM	0 18 * * 1-5

Scheduled Workflow Example

from bifrost import workflow, config
import httpx
import logging

logger = logging.getLogger(__name__)

@workflow(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. Source: A schedule event source stores the cron expression, timezone, organization scope, and overlap policy.
2. Subscription: A subscription binds the source to the workflow you want to run.
3. Execution: At the scheduled time, the source emits an event and the subscription queues the workflow in the source’s organization scope.
4. Logging: Results are stored in execution history.

Note

An organization-scoped schedule runs its subscriptions in that organization. A schedule source explicitly created with no organization is global.

Execution Context

Scheduled workflows have limited context:

from bifrost import workflow, context

@workflow
async def scheduled_task():
    # context.org_id follows the schedule source's organization scope.
    # It is None only for an explicitly global source.
    return {"organization_id": context.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

Disable the schedule event source or remove the subscription:

Pause the event source or delete the subscription in the UI/API.

Tips

• Choose a timezone: The source defaults to UTC and accepts IANA names such as America/New_York
• Avoid overlaps: Don’t schedule faster than execution time
• Set timeouts: Configure workflow timeouts in the workflow settings
• Log progress: Track execution for debugging

See Also

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