Composing workflows

A single workflow handles one coherent unit of work. When that unit grows, or when you want to reuse logic across multiple callers, compose it: one workflow calls another, fans out in parallel, or chains steps through a pipeline. Flux provides three primitives for this: call, parallel, and pipeline.

Calling a sub-workflow with call

call is a durable task that invokes another workflow over HTTP and waits for the result. It requires a running Flux server — use it in deployed or multi-worker environments where each workflow runs in its own execution context.

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


@task
async def fetch_price(ticker: str) -> float:
    # Replace with a real data source
    prices = {"AAPL": 182.5, "MSFT": 415.0, "GOOG": 177.3}
    return prices.get(ticker, 0.0)


@workflow
async def price_workflow(ctx: ExecutionContext[str]) -> float:
    return await fetch_price(ctx.input)


@workflow
async def portfolio_workflow(ctx: ExecutionContext[list[str]]):
    """Fetch prices for a list of tickers by calling a sub-workflow for each."""
    results = {}
    for ticker in ctx.input:
        results[ticker] = await call(price_workflow, ticker)
    return results

call records each sub-workflow invocation as a durable event. If portfolio_workflow is interrupted and replayed, tickers whose call already completed are skipped, and the previously recorded output is returned immediately from the execution log.

Nested sub-workflows

Sub-workflows can themselves call further sub-workflows. The durability guarantee composes: each call is its own checkpoint, so a failure at any depth replays only from the nearest recorded result.

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


@task
async def fetch_order(order_id: str) -> dict:
    # Stub: return a minimal order record
    return {"id": order_id, "amount": 99.00, "currency": "USD"}


@task
async def fetch_customer(customer_id: str) -> dict:
    return {"id": customer_id, "name": "Acme Corp", "tier": "premium"}


@workflow
async def enrich_order(ctx: ExecutionContext[str]) -> dict:
    """Fetch and enrich a single order."""
    order = await fetch_order(ctx.input)
    customer = await fetch_customer("cust-001")
    return {**order, "customer": customer}


@workflow
async def process_orders(ctx: ExecutionContext[list[str]]):
    """Enrich a batch of orders, each as a separate sub-workflow."""
    enriched = []
    for order_id in ctx.input:
        result = await call(enrich_order, order_id)
        enriched.append(result)
    return enriched

Each call(enrich_order, order_id) creates an independent execution with its own execution ID and event log. You can inspect, replay, or retry any individual order enrichment without re-running the full batch.

Fanning out in parallel

When sub-tasks are independent, run them concurrently with parallel. Import it from flux.tasks:

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


@task
async def word_count(text: str) -> int:
    return len(text.split())


@task
async def char_count(text: str) -> int:
    return len(text)


@task
async def sentence_count(text: str) -> int:
    return text.count(".") + text.count("!") + text.count("?")


@workflow
async def analyze_text(ctx: ExecutionContext[str]):
    """Run three independent analyses concurrently."""
    words, chars, sentences = await parallel(
        word_count(ctx.input),
        char_count(ctx.input),
        sentence_count(ctx.input),
    )
    return {"words": words, "chars": chars, "sentences": sentences}


if __name__ == "__main__":
    ctx = analyze_text.run("Flux is fast. It is durable!")
    print(ctx.output)
    # {'words': 6, 'chars': 28, 'sentences': 2}

parallel accepts any number of task coroutines and returns a list of results in the same order as the inputs. The entire fan-out is recorded as a single durable event. If the workflow is interrupted mid-fan-out and replayed, the already-completed results are restored and the remaining tasks continue from where they left off.

Nested parallel composition

Fan-outs can be nested. A common pattern is collecting one type of measurement per service, then aggregating across services, with both dimensions running concurrently:

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


@task
async def fetch_metric(service: str, metric: str) -> float:
    # Replace with a call to your monitoring system
    return float(hash(f"{service}:{metric}") % 100)


@task
async def aggregate(values: list[float]) -> dict:
    return {
        "min": min(values),
        "max": max(values),
        "avg": sum(values) / len(values),
    }


@workflow
async def monitor_services(ctx: ExecutionContext[list[str]]):
    """Collect CPU and memory metrics for each service in parallel."""
    services = ctx.input
    cpu_tasks = [fetch_metric(svc, "cpu") for svc in services]
    mem_tasks = [fetch_metric(svc, "memory") for svc in services]

    cpu_readings, mem_readings = await parallel(
        parallel(*cpu_tasks),
        parallel(*mem_tasks),
    )
    return {
        "cpu": await aggregate(cpu_readings),
        "memory": await aggregate(mem_readings),
    }


if __name__ == "__main__":
    ctx = monitor_services.run(["api", "worker", "scheduler"])
    print(ctx.output)
    # {'cpu': {'min': ..., 'max': ..., 'avg': ...}, 'memory': {...}}

The inner parallel(*cpu_tasks) and parallel(*mem_tasks) run concurrently with each other, and within each group the individual metric fetches also run concurrently. All results are checkpointed before the outer parallel resolves.

Chaining steps with pipeline

pipeline threads the output of each step into the input of the next. The result is a linear chain where order and type compatibility matter:

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


@task
async def normalize(text: str) -> str:
    return text.strip().lower()


@task
async def tokenize(text: str) -> list[str]:
    return text.split()


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


@workflow
async def text_pipeline(ctx: ExecutionContext[str]) -> int:
    """Normalize, tokenize, and count tokens in a text string."""
    return await pipeline(normalize, tokenize, count_tokens, input=ctx.input)


if __name__ == "__main__":
    ctx = text_pipeline.run("  Hello World, this is Flux!  ")
    print(ctx.output)        # 5
    print(ctx.has_succeeded) # True

pipeline takes the task functions (not coroutines — no call parentheses), applies them left to right, and passes input as the first argument. Each intermediate result is durably checkpointed, so a failure mid-chain only re-runs from the last successful step.

When to use pipeline vs. sequential awaits

pipeline and a plain sequence of await calls are equivalent in behavior; both checkpoint every step. Use pipeline when:

• The chain is long enough that explicit variables would clutter the function body.
• The steps are pure data transformations with a clear “in → out” shape.
• You want to express the chain as a declaration rather than imperative code.

Use sequential await when steps need local variables, branching, or side effects between them.

Choosing the right primitive

All three are durable: each awaited result is checkpointed. A replay skips any step whose result was already recorded.

What’s next

• Tasks — parallel — detailed options for parallel, including error propagation and partial results.
• Tasks — pipeline — typed pipeline variants and error handling mid-chain.