Defining tasks

A task Task An async Python function decorated with @task that does one unit of work — an API call, a query, a transformation. Tasks are the unit of retry, cache, and resumability. Full definition → is an async (or plain sync) function decorated with @task. It is the smallest unit of durable work in Flux: every time a workflow awaits a task, the result is recorded to the execution log. If the process restarts, Flux replays from the log and skips re-running completed tasks.

The minimal task

Import task and apply the decorator:

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


@task
async def shout(text: str) -> str:
    return text.upper()


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


if __name__ == "__main__":
    ctx = echo_loud.run("hello")
    print(ctx.output)         # HELLO
    print(ctx.has_succeeded)  # True

@task wraps the function in a task object. Call it with await from inside a workflow, just like any coroutine. Flux records the return value after the first successful run and returns the recorded value on replay without calling the function again. @task works on both async def and plain def functions. Sync functions are run as-is; Flux wraps them with maybe_awaitable internally.

Configuring a task with task.with_options()

Use @task.with_options(...) to attach configuration to a task without touching its body. Declare options once, close to the function definition; they apply every time the task runs:

@task.with_options(
    name="fetch-external-data",
    retry_max_attempts=3,
    retry_delay=1,
    retry_backoff=2,
    timeout=30,
)
async def fetch_data(url: str) -> dict:
    ...

task.with_options(...) is a descriptor: called on the class (task.with_options(...)) it returns a decorator; called on an instance (existing_task.with_options(...)) it returns a new task with the changed options merged over the existing ones.

The full options surface

Task with retries

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


@task.with_options(
    retry_max_attempts=3,
    retry_delay=1,
    retry_backoff=2,
)
async def flaky_fetch(url: str) -> str:
    # Raises on transient errors; Flux retries up to 3 times
    # with an exponentially growing wait before each retry
    ...


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

The retry delay grows exponentially: with retry_delay=1 and retry_backoff=2, the waits are 1 s → 2 s → 4 s, capped at 600 s. After all retries are exhausted, Flux raises an ExecutionError wrapping the last exception (or calls fallback if one is configured).

Task with caching

Set cache=True when the same inputs always produce the same output and the computation is expensive:

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


@task.with_options(cache=True)
async def expensive_lookup(key: str) -> str:
    # Expensive computation or external call
    return f"result-for-{key}"


@workflow
async def double_lookup(ctx: ExecutionContext[str]):
    first  = await expensive_lookup(ctx.input)
    second = await expensive_lookup(ctx.input)  # served from disk cache
    return {"first": first, "second": second}

The cache is keyed by the task’s internal ID, which is derived from the task name and its call arguments. A second call with the same arguments within the same execution returns the cached value without invoking the function.

Task with secret_requests

Provide secret_requests to pull credentials from the Flux secret store. The secrets are fetched before the task runs and injected as a secrets dict keyword argument:

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


@task.with_options(
    secret_requests=["STRIPE_SECRET_KEY"],
)
async def charge_customer(amount_cents: int, secrets: dict = {}) -> str:
    api_key = secrets["STRIPE_SECRET_KEY"]
    # Use api_key to call the Stripe API ...
    return f"charged {amount_cents} cents"


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

The task function declares secrets: dict = {} as a keyword argument with a default value. Flux overwrites it with the fetched secrets at runtime; the default is never used in production but satisfies type checkers and makes the dependency explicit in the signature.

Secrets must be registered in the store before the workflow runs. Use the CLI to add them:

flux secrets set STRIPE_SECRET_KEY sk-live-...

Dynamic task names

The name option supports Python format-string syntax against the task’s call arguments. Use this when many tasks share a template but need distinct names for observability:

@task.with_options(name="process-{item_id}")
async def process_item(item_id: str) -> str:
    ...

Each call to process_item("order-42") is recorded as process-order-42 in the execution log, making execution traces readable without extra instrumentation.

What’s next

• Reliability: errors and retries — deeper coverage of retry strategies, fallback patterns, and rollback logic.
• Reference: @task — complete API reference for the task class and all with_options parameters.