Graph tasks

Graph from flux.tasks lets you describe a task execution graph explicitly: you name each node, declare directed edges between them, and designate which nodes start and end the graph. When the graph runs, Flux traverses the edges in order, passing each node’s output to its downstream neighbors.

Use Graph when the relationship between tasks is a directed acyclic graph that cannot be expressed as a flat pipeline chain. For example, when one task fans out to two independent successors, or when you want the graph structure to be self-documenting in code.

Core concepts

A Graph has four kinds of API calls:

Every graph must have at least one start node and one end node. Calling the graph object with an input value runs the full traversal and returns the output of the terminal (end) node.

Basic usage

The simplest graph is a two-step linear chain. Import Graph from flux.tasks, declare nodes, connect them, then await the call:

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


@task
async def get_name(input: str) -> str:
    return input


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


@workflow
async def hello_workflow(ctx: ExecutionContext[str]):
    hello = (
        Graph("hello_world")
        .add_node("get_name", get_name)
        .add_node("say_hello", say_hello)
        .add_edge("get_name", "say_hello")
        .start_with("get_name")
        .end_with("say_hello")
    )
    return await hello(ctx.input)


if __name__ == "__main__":
    ctx = hello_workflow.run("World")
    print(ctx.output)       # Hello, World
    print(ctx.has_succeeded)  # True

The graph passes ctx.input into get_name. get_name’s return value flows into say_hello, and say_hello’s return value becomes the graph’s output, which is also the workflow’s return value.

A realistic example: ETL pipeline

Graph works well for data pipelines where each stage is a named processing step and the data dependency structure should be visible in the declaration:

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


@task
async def fetch_record(record_id: str) -> dict:
    # Replace with a real data source call
    return {"id": record_id, "raw_value": "  42  ", "tags": "a,b,c"}


@task
async def clean(record: dict) -> dict:
    return {k: v.strip() if isinstance(v, str) else v for k, v in record.items()}


@task
async def enrich(record: dict) -> dict:
    return {**record, "tags": record["tags"].split(",")}


@workflow
async def etl_workflow(ctx: ExecutionContext[str]):
    pipeline = (
        Graph("etl")
        .add_node("fetch", fetch_record)
        .add_node("clean", clean)
        .add_node("enrich", enrich)
        .add_edge("fetch", "clean")
        .add_edge("clean", "enrich")
        .start_with("fetch")
        .end_with("enrich")
    )
    return await pipeline(ctx.input)


if __name__ == "__main__":
    ctx = etl_workflow.run("rec-001")
    print(ctx.output)
    # {'id': 'rec-001', 'raw_value': '42', 'tags': ['a', 'b', 'c']}
    print(ctx.has_succeeded)  # True

Each edge makes a dependency explicit: clean cannot run until fetch completes; enrich cannot run until clean completes. Adding a step is add_node + add_edge, no argument threading needed.

Validation

Calling the graph validates its structure before execution. validate() raises ValueError for three structural problems. A fourth case — duplicate node names — is raised by add_node() at registration time, before validate() runs.

No start node — you declared nodes but did not call start_with:

from flux.tasks import Graph
from flux.task import task


@task
async def process(x): return x


g = Graph("incomplete")
g.add_node("process", process)
g.end_with("process")
g.validate()
# ValueError: Graph must have a starting node.

No end node — you called start_with but not end_with:

g = Graph("no_end")
g.add_node("process", process)
g.start_with("process")
g.validate()
# ValueError: Graph must have a ending node.

Disconnected node — a node is registered but not reachable from the start:

@task
async def orphan(x): return x


g = Graph("orphan_graph")
g.add_node("process", process)
g.add_node("orphan", orphan)   # not connected to anything
g.start_with("process")
g.end_with("process")
g.validate()
# ValueError: Not all nodes are connected.

Duplicate node name — two nodes registered under the same string key. This case is raised by add_node() itself at registration time, not by validate():

g = Graph("dup")
g.add_node("process", process)
g.add_node("process", process)   # raised here, before validate()
# ValueError: Node process already present.

validate() runs automatically when the graph is called, so you do not need to invoke it manually. Structural errors surface as a ValueError before any task is executed.

Cycle — a back-edge creates a loop. validate() runs a depth-first traversal that detects cycles and raises a ValueError before any task executes:

g = Graph("cyclic")
g.add_node("a", process)
g.add_node("b", process)
g.add_edge("a", "b")
g.add_edge("b", "a")   # back-edge — forms a cycle
g.start_with("a")
g.end_with("b")
g.validate()
# ValueError: Graph contains a cycle involving node 'a'.

Graphs must be acyclic — directed edges flow forward only.

Conditional edges

add_edge accepts an optional condition argument: a callable that receives the upstream node’s output and returns a boolean. When a condition is present, the downstream node only runs if the condition evaluates to True.

The condition callable is awaited during traversal and receives the upstream node’s output. A downstream node runs only when every condition on its incoming edges evaluates to True; if a condition returns False, that downstream node — and everything reachable only through it — is skipped.

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


@task
async def validate_record(record: dict) -> dict:
    return {**record, "valid": record.get("value", 0) > 0}


@task
async def process_record(result: dict) -> str:
    return f"processed:{result['value']}"


@task
async def report_outcome(result: dict) -> str:
    if result["valid"]:
        return f"processed:{result['value']}"
    return "rejected:value_must_be_positive"


@workflow
async def validated_etl(ctx: ExecutionContext[dict]):
    flow = (
        Graph("validated_etl")
        .add_node("validate", validate_record)
        .add_node("process", process_record)
        .add_node("report", report_outcome)
        .add_edge("validate", "process",
                  condition=lambda result: result.get("valid"))
        .add_edge("validate", "report")
        .start_with("validate")
        .end_with("report")
    )
    return await flow(ctx.input)

Here process runs only when validate produces a valid record; report always runs because its incoming edge has no condition. The condition is awaited, so the branch is taken or skipped correctly.

Because the END node only completes once every node feeding it has completed, design conditional graphs so the terminal node is always reachable — gate the optional work on a condition, and route the unconditional path to the end node directly, rather than calling end_with() on two mutually exclusive branches.

Graph vs. pipeline

Both Graph and pipeline express a sequence of dependent tasks. The difference is structure and flexibility:

Use pipeline for straight-line transformations where simplicity matters. Use Graph when you want named nodes, explicit edge declarations, or a structure that reads like a dependency diagram.

What’s next

• Parallel — run independent task coroutines concurrently with parallel().
• Pipeline — sequential composition where each task’s output feeds the next.