Scheduling workflows

A scheduled workflow runs automatically, without an external cron daemon or orchestration glue. You declare the schedule on the workflow itself, and Flux creates and maintains it whenever the workflow is registered with a server.

The three schedule types

Flux ships three schedule factories, all importable from flux:

Attaching a schedule

Pass a schedule object to workflow.with_options(schedule=...). The workflow still runs normally when called directly; the schedule only activates in a distributed deployment.

Cron schedule

Run a workflow at 9 AM UTC on weekdays:

from flux import ExecutionContext, cron
from flux.task import task
from flux.workflow import workflow


@task
async def generate_report(data: str) -> str:
    return f"Report: {data}"


@workflow.with_options(name="daily_report", schedule=cron("0 9 * * MON-FRI", timezone="UTC"))
async def daily_report_workflow(ctx: ExecutionContext[str]):
    data = ctx.input or "Daily metrics"
    return await generate_report(data)

cron() accepts any five-field cron expression. The optional timezone parameter defaults to "UTC". Use explicit IANA timezone names ("America/New_York", "Europe/Berlin"); the string "local" is rejected because it is not portable across workers.

Interval schedule

Run a workflow every six hours:

from flux import ExecutionContext, interval
from flux.task import task
from flux.workflow import workflow


@task
async def sync_data(source: str, target: str) -> str:
    return f"Synced {source} → {target}"


@workflow.with_options(name="sync_data", schedule=interval(hours=6, timezone="UTC"))
async def data_sync_workflow(ctx: ExecutionContext[dict]):
    config = ctx.input or {"source": "database", "target": "warehouse"}
    return await sync_data(config["source"], config["target"])

interval() accepts seconds, minutes, hours, days, and weeks. Mix them freely; the total interval must be positive. Two optional boundary parameters let you scope the schedule:

from datetime import datetime, timezone

schedule=interval(
    hours=6,
    timezone="UTC",
    start_time=datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc),
    end_time=datetime(2026, 12, 31, 23, 59, tzinfo=timezone.utc),
)

When start_time is set, the first execution waits until that time. When end_time is reached, the schedule stops producing runs.

One-time schedule

Run a workflow exactly once at a specific moment:

from datetime import datetime, timezone

from flux import ExecutionContext, once
from flux.task import task
from flux.workflow import workflow


@task
async def run_migration() -> str:
    return "Migration complete"


@workflow.with_options(
    name="one_time_migration",
    schedule=once(datetime(2026, 12, 25, 9, 0, 0, tzinfo=timezone.utc)),
)
async def migration_workflow(ctx: ExecutionContext[str]):
    return await run_migration()

After the workflow fires once, the schedule is marked executed and never triggers again. Use this for one-off data migrations, post-deploy jobs, or anything that should run at a precise future time without manual intervention.

Auto-registration on workflow register

When you register a workflow that carries a schedule, Flux automatically creates a named schedule record in the server’s schedule store. No additional CLI step is required.

The auto-created schedule is named <workflow-name>_auto. If the schedule already exists (for example, after a code redeploy), Flux updates it in place rather than creating a duplicate.

Register a workflow with the CLI:

flux workflow register path/to/workflow.py

Verify the schedule was created:

flux schedule list

NAME                    WORKFLOW        TYPE      NEXT RUN
daily_report_auto       daily_report    cron      2026-07-06 09:00:00 UTC
sync_data_auto          sync_data       interval  2026-07-05 15:00:00 UTC

Providing default input

Scheduled workflows receive no caller-supplied input. Use a fallback in the workflow body to provide defaults:

@workflow.with_options(name="daily_report", schedule=cron("0 9 * * MON-FRI"))
async def daily_report_workflow(ctx: ExecutionContext[str]):
    data = ctx.input or "Daily metrics"   # safe default for scheduled runs
    return await generate_report(data)

In Flux 0.56.0 there is no flux schedule update command. The available subcommands are create, list, show, pause, resume, delete, and history. To change a scheduled run’s input (or any other field), delete the existing auto-schedule and re-create it, or use the REST API or Python SDK directly:

# flux schedule delete prompts for confirmation; pass --yes to skip
flux schedule delete daily_report_auto --yes
flux schedule create daily_report daily_report_auto \
  --cron "0 9 * * MON-FRI" \
  --input '{"region": "us-east-1"}'

flux schedule create takes two positional arguments — the workflow name followed by the schedule name. Reusing the original <workflow>_auto schedule name preserves the convention used by auto-registration.

The fallback-in-the-workflow-body pattern shown above is the lower-friction option: it keeps the schedule untouched and lets callers override ctx.input per run when invoking the workflow directly.

Inspecting the schedule at runtime

The schedule object is accessible on the workflow itself, which is useful for logging or testing:

from flux import cron
from flux.workflow import workflow

@workflow.with_options(name="daily_report", schedule=cron("0 9 * * MON-FRI"))
async def daily_report_workflow(ctx):
    ...

print(daily_report_workflow.schedule)          # <CronSchedule ...>
print(daily_report_workflow.schedule.cron_expression)  # "0 9 * * MON-FRI"
print(daily_report_workflow.schedule.next_run_time())  # next scheduled datetime

What comes next

Declaring a schedule is the development-time step. For the production side (pausing, resuming, and auditing schedules after deployment), see Operate → Scheduling → schedule management.