Built-in tasks

A tour of flux.tasks — the standard library of ready-made tasks for time, randomness, concurrency, pausing, progress reporting, and cross-workflow calls.

flux.tasks ships a small standard library of tasks you can drop into any workflow. Each one follows the same rules as a hand-written task: the result is recorded in the execution log, replays skip the call, and the task runs inside the current ExecutionContext. Import what you need and await it.

from flux.tasks import now, uuid4, sleep, choice, randint, randrange
from flux.tasks import parallel, pipeline
from flux.tasks import pause, progress, call

now — deterministic timestamps

now() -> datetime

Returns datetime.now() as a durable task. Calling datetime.now() directly inside a workflow is unsafe: on a replay the call returns a different value, which breaks determinism. now records the timestamp on the first run and returns the recorded value on every subsequent replay.

from flux import ExecutionContext
from flux.tasks import now
from flux.workflow import workflow


@workflow
async def timestamped_job(ctx: ExecutionContext):
    started_at = await now()
    # ... do work ...
    return {"started_at": str(started_at)}

Use now wherever you need the current time inside a workflow body. Avoid time.time(), datetime.utcnow(), or any other live clock call for the same reason.

uuid4 — durable unique IDs

uuid4() -> uuid.UUID

Generates a uuid.UUID (version 4) and records it. Like now, calling uuid.uuid4() directly is non-deterministic: each replay produces a different UUID. Using the task ensures the same ID is returned every time the workflow runs through that point.

from flux import ExecutionContext
from flux.tasks import uuid4
from flux.workflow import workflow


@workflow
async def create_order(ctx: ExecutionContext[dict]):
    order_id = await uuid4()
    return {"order_id": str(order_id), "payload": ctx.input}

The returned value is a uuid.UUID object. Convert to string with str(order_id) when you need a plain text representation.

sleep — durable delays

sleep(duration: float | timedelta)

Pauses workflow execution for duration seconds. Accepts either a plain float (number of seconds) or a datetime.timedelta.

from datetime import timedelta

from flux import ExecutionContext
from flux.tasks import sleep
from flux.workflow import workflow


@workflow
async def polling_loop(ctx: ExecutionContext[str]):
    for attempt in range(5):
        result = await check_external_api(ctx.input)
        if result["ready"]:
            return result
        await sleep(timedelta(seconds=30))   # wait 30 s between checks
    return {"status": "timed out"}

choice, randint, randrange — durable randomness

Raw calls to the random module are non-deterministic: each replay picks a different value. These three tasks wrap the corresponding random calls as durable tasks so the result is fixed after the first execution.

choice

choice(options: list[Any]) -> Any

Returns a randomly selected element from options. The return type matches the element type of the list.

from flux import ExecutionContext
from flux.tasks import choice
from flux.workflow import workflow


REGIONS = ["us-east-1", "eu-west-1", "ap-southeast-1"]


@workflow
async def deploy_canary(ctx: ExecutionContext[str]):
    region = await choice(REGIONS)
    return {"image": ctx.input, "region": region}

randint

randint(a: int, b: int) -> int

Returns a random integer N such that a <= N <= b. Both endpoints are inclusive, matching the behaviour of random.randint.

from flux import ExecutionContext
from flux.tasks import randint
from flux.workflow import workflow


@workflow
async def roll_dice(ctx: ExecutionContext):
    die1 = await randint(1, 6)
    die2 = await randint(1, 6)
    return {"roll": die1 + die2}

randrange

randrange(start: int, stop: int | None = None, step: int = 1)

Returns a randomly selected integer from range(start, stop, step). Mirrors random.randrange with the same argument semantics: stop is exclusive, step defaults to 1.

from flux import ExecutionContext
from flux.tasks import randrange
from flux.workflow import workflow


@workflow
async def pick_even(ctx: ExecutionContext):
    # Pick a random even number between 0 and 98 (inclusive)
    n = await randrange(0, 100, 2)
    return n

parallel — concurrent fan-out

parallel(*functions: Coroutine) -> list[Any]

Runs multiple task coroutines concurrently and returns their results as a list in the order the arguments were passed. This is the right primitive when tasks do not depend on each other’s output.

parallel has a dedicated page with full examples, .map() fan-out, and error semantics. See Tasks → parallel.

pipeline — linear chaining

pipeline(*tasks: Callable, input: Any)

Runs a sequence of tasks left to right, passing the output of each step as the input to the next, which fits linear data transformations cleanly. pipeline has a dedicated page with full examples and error propagation details. See Tasks → pipeline.

pause — human-in-the-loop checkpoints

pause(name: str, output: Any = None) -> Any

Suspends the workflow at the call site and waits for an external resume signal. When the workflow resumes, pause returns whatever value was passed to the resume call — typically a human decision, an approval, or a form submission.

from flux import ExecutionContext
from flux.tasks import pause
from flux.workflow import workflow


@workflow
async def document_approval(ctx: ExecutionContext[dict]):
    doc = ctx.input

    # Workflow pauses here; the execution_id is surfaced to the requester
    decision = await pause(
        "review-document",
        output={"document": doc["title"], "action": "approve_or_reject"},
    )

    if decision == "approved":
        return {"status": "approved", "title": doc["title"]}
    return {"status": "rejected", "title": doc["title"]}

The output argument is optional payload surfaced to the resume caller. Use it to pass context that helps a person or external system decide what to do next. The string name is a label that appears in the execution log and is used to route resume signals correctly when a workflow has multiple pause points.

For the full pause-and-resume lifecycle — including how to send the resume signal via CLI, REST API, or SDK — see Workflow control → pause and resume.

progress — streaming status updates

progress(value: Any) -> None

Emits an ephemeral progress event from inside a task. Progress events are streamed to connected clients in real time but are never written to the execution log and never replayed. They are fire-and-forget signals for dashboards and long-running task feedback loops.

from flux import ExecutionContext
from flux.task import task
from flux.tasks import progress
from flux.workflow import workflow


@task
async def process_batch(items: list[str]) -> list[str]:
    results = []
    for i, item in enumerate(items):
        await progress({"step": i + 1, "total": len(items), "item": item})
        results.append(item.upper())
    return results


@workflow
async def batch_job(ctx: ExecutionContext[list]):
    return await process_batch(ctx.input)

progress can be called multiple times per task, and the value can be anything JSON-serialisable: a percentage, a dict with step metadata, a log line, or a custom object. Because progress events are ephemeral, the task’s durable output is still only the value returned from the function.

For how to consume these events over SSE from the server, see Workflows → streaming output.

call — invoke another workflow

call(workflow: workflow | str, *args, mode: Literal["sync", "async"] = "sync") -> Any

Calls a registered workflow via the Flux HTTP API from inside a running workflow. Use it for cross-workflow orchestration when you need to trigger a workflow registered under a different name or namespace, or when you want the child workflow’s execution tracked independently.

from flux import ExecutionContext
from flux.tasks import call
from flux.workflow import workflow


@workflow
async def orchestrator(ctx: ExecutionContext[dict]):
    payload = ctx.input

    # Synchronous: waits for the child workflow to finish, returns its output
    result = await call("default/data-pipeline", payload)

    # Asynchronous: submits and returns the execution_id immediately
    execution_id = await call("default/notification-sender", payload, mode="async")

    return {"pipeline_result": result, "notification_id": execution_id}

mode="sync" (the default) blocks until the child workflow completes and returns its output. mode="async" submits the workflow and returns the execution_id string immediately, leaving the child to run independently.

Example: combining primitives

A workflow that generates a unique order ID, waits for human approval, then fans out work in parallel:

from flux import ExecutionContext
from flux.task import task
from flux.tasks import uuid4, pause, parallel
from flux.workflow import workflow


@task
async def process_item(item: str) -> str:
    return item.strip().upper()


@workflow
async def order_pipeline(ctx: ExecutionContext[list]):
    order_id = await uuid4()

    decision = await pause("approve-order", output={"order_id": str(order_id)})
    if decision != "approved":
        return {"order_id": str(order_id), "status": "rejected"}

    results = await parallel(*(process_item(item) for item in ctx.input))
    return {"order_id": str(order_id), "items": results, "status": "completed"}

What’s next