Defining workflows

A workflow is an async function decorated with @workflow. Flux intercepts every await inside it, records the result to a durable log, and replays from the last checkpoint if the process restarts. That is the entire contract. Everything else follows from it.

The decorator anatomy

Import workflow and ExecutionContext from Flux, then apply @workflow to any async def:

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


@task
async def greet(name: str) -> str:
    return f"Hello, {name}!"


@workflow
async def hello(ctx: ExecutionContext[str]):
    return await greet(ctx.input)


if __name__ == "__main__":
    ctx = hello.run("world")
    print(ctx.output)        # Hello, world!
    print(ctx.has_succeeded) # True

The function signature takes exactly one positional argument: ctx, typed as ExecutionContext[T]. The generic parameter T is the expected type of the input your caller provides. Inside the workflow you call tasks with await; Flux records the outcome of each awaited call so a replay can skip re-executing them. .run() executes the workflow in-process (no server required) and returns the same ExecutionContext object, now populated with .output and status flags.

Typed input

The type parameter on ExecutionContext[T] is for documentation and static analysis; Flux passes whatever you give to .run() as ctx.input. Use a dataclass or Pydantic model when your input has structure:

from dataclasses import dataclass
from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@dataclass
class ReportRequest:
    topic: str
    max_words: int = 500


@task
async def draft_section(topic: str, limit: int) -> str:
    # Replace with real generation logic
    return " ".join([topic] * min(limit, 10))


@workflow
async def generate_report(ctx: ExecutionContext[ReportRequest]):
    req = ctx.input
    return await draft_section(req.topic, req.max_words)


if __name__ == "__main__":
    ctx = generate_report.run(ReportRequest(topic="durable execution", max_words=10))
    print(ctx.output)
    print(ctx.has_succeeded)  # True

Annotating the type parameter lets your editor infer the type of ctx.input correctly. req in the example above resolves to ReportRequest in most type checkers.

Returning a structured result

A workflow can return any picklable value. When the result has multiple fields, return a dataclass, which keeps the call site readable and gives you a typed .output on the context:

from dataclasses import dataclass
from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@dataclass
class PipelineResult:
    raw_count: int
    processed_count: int
    summary: str


@task
async def count_records(data: list[str]) -> int:
    return len(data)


@task
async def process(data: list[str]) -> list[str]:
    return [item.strip().lower() for item in data]


@workflow
async def etl_pipeline(ctx: ExecutionContext[list[str]]) -> PipelineResult:
    data = ctx.input
    raw_count = await count_records(data)
    processed = await process(data)
    processed_count = await count_records(processed)
    return PipelineResult(
        raw_count=raw_count,
        processed_count=processed_count,
        summary=f"Processed {processed_count} of {raw_count} records.",
    )


if __name__ == "__main__":
    ctx = etl_pipeline.run(["  Apple  ", "banana", "  Cherry"])
    result = ctx.output
    print(result.summary)          # Processed 3 of 3 records.
    print(result.processed_count)  # 3

Flux serializes the return value with dill (a pickle-compatible serializer) by default. For large outputs, or when you need the result accessible outside the process, configure output_storage on the workflow — see Reference: @workflow for the full options surface.

Configuring a workflow

Use @workflow.with_options(...) to set metadata or infrastructure requirements without touching the function body:

@workflow.with_options(
    name="etl-pipeline",
    secret_requests=["DATABASE_URL"],
    affinity={"role": "etl-worker"},
)
async def etl_pipeline(ctx: ExecutionContext[list[str]]) -> PipelineResult:
    ...

Options most commonly used:

ExecutionContext properties

The ctx object you receive is more than a carrier for .input. After a workflow finishes (or when you retrieve a past execution), these properties tell you what happened:

You won’t need most of these while writing the workflow body. They are primarily for the call site inspecting a finished execution, or for tests asserting on results.

What’s next

• Composing workflows — call one workflow from another, fan out in parallel, and chain pipelines.
• Reference: @workflow — the complete options surface for @workflow.with_options, serialization rules, and version pinning.