Your first workflow

Write a Flux workflow from scratch — tasks, composition, and durability.

You ran a workflow in the Quickstart. This page builds one from scratch: single task, then composed, then parallel, then with retries. It covers the full core surface: @task, @workflow, ExecutionContext, and how Flux turns ordinary async functions into durable executions.

The smallest unit: a task

A task is an async function decorated with @task. Each call is recorded to the event log; if the workflow restarts, completed tasks don’t re-execute.

from flux.task import task

@task
async def double(n: int) -> int:
    return n * 2

Tasks are typed via Python type hints and can do anything: fetch from a database, call an HTTP service, run a CPU-heavy computation.

Wrap tasks in a workflow

A workflow is a function decorated with @workflow. It receives an ExecutionContext (ctx) and calls one or more tasks.

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow

@task
async def double(n: int) -> int:
    return n * 2

@workflow
async def double_it(ctx: ExecutionContext[int]):
    return await double(ctx.input)

ctx.input is whatever you pass to .run(). The workflow’s return value is captured as ctx.output after completion.

Run it:

if __name__ == "__main__":
    ctx = double_it.run(21)
    print(ctx.output)  # 42

Compose tasks sequentially

Multiple await calls in a workflow run in order, with each task’s result available to the next.

@task
async def add_one(n: int) -> int:
    return n + 1

@workflow
async def double_then_add(ctx: ExecutionContext[int]):
    doubled = await double(ctx.input)
    return await add_one(doubled)

If the workflow crashes after double completes but before add_one starts, replay re-executes from add_one. double’s result is read from the event log, not recomputed.

Run tasks in parallel

For independent work, use parallel:

from flux.tasks import parallel

@task
async def fetch_user(user_id: int) -> dict:
    return await db.get(user_id)

@workflow
async def fetch_many(ctx: ExecutionContext[list[int]]):
    user_ids = ctx.input
    users = await parallel(*[fetch_user(uid) for uid in user_ids])
    return users

parallel(*tasks) returns a list of results in the same order as the input tasks, after all tasks complete. An exception in any task propagates from parallel and the workflow fails. Other tasks that were already in flight may still complete in the background.

Add retries

Configure retry behavior per task with task.with_options:

@task.with_options(
    retry_max_attempts=3,
    retry_delay=1,
    retry_backoff=2,
    timeout=30,
)
async def flaky_api_call(url: str) -> dict:
    return await http.get(url)

Flux retries on any unhandled exception. The wait between attempts grows exponentially: with the configuration above the first retry waits retry_delay (1 s), the second waits retry_delay × retry_backoff (2 s), the third retry_delay × retry_backoff² (4 s), capped at 600 s. See Reliability → errors and retries for the full backoff behaviour.

To handle specific failure cases, use fallback (called when retries are exhausted) or rollback (called for cleanup after failure):

async def my_fallback(*args, **kwargs):
    return {"status": "degraded", "reason": "API unreachable"}

@task.with_options(
    retry_max_attempts=3,
    fallback=my_fallback,
)
async def fetch_with_fallback(url: str) -> dict:
    return await http.get(url)

What to remember

The non-obvious parts: parallel(*tasks) preserves input order in the result list, and retry backoff compounds exponentially — each retry waits retry_delay × retry_backoff more than the last, capped at 600 s. Everything else follows from the fact that Flux writes each task result to the event log before moving on.

Next