Workflow composition, failure handlers, and nodes

Workflows: Tasks, Nodes, and Promises

When you define a workflow with @workflow, flytekit does not run your Python code on the Flyte backend. Instead, it reads your function body at compile time and builds a directed acyclic graph (DAG) of task nodes. The values flowing between tasks are not Python objects — they are Promise wrappers. Understanding how these three concepts interlock is essential to writing correct workflows.

The @workflow decorator and compile-time graph construction

Decorating a function with @workflow wraps it in a PythonFunctionWorkflow instance (defined in flytekit/core/workflow.py). When that function is called — either explicitly or when the workflow is serialized — the flyte_entity_call_handler decorator intercepts the call. In compilation mode (when ctx.compilation_state is set), it calls create_and_link_node instead of executing the body. This function, defined in flytekit/core/promise.py, creates a Node object and appends it to the compilation state, then returns a named-tuple of Promise objects wrapping NodeOutput references to that node.

from flytekit import workflow, task

@task
def t1(a: int) -> (int, str):
    return a + 2, f"result-{a}"

@workflow
def my_wf(a: int) -> int:
    x, y = t1(a=a)   # t1() returns a named-tuple of two Promises, not actual values
    return x          # the return value is also a Promise

In the body of my_wf, t1(a=a) does not execute t1's Python code. The flyte_entity_call_handler function (at flytekit/core/promise.py:1442) detects compilation mode and calls create_and_link_node, which returns a named-tuple of Promise objects for o0 and o1. Unpacking that tuple gives you two individual Promise objects. When you return x, that Promise is used to construct the workflow's output binding.

Task outputs as Promise objects

The Promise class (defined in flytekit/core/promise.py:438) wraps either a resolved literal value or a reference to an upstream node's output. When a Promise wraps a NodeOutput, its is_ready property is False — the actual value is not yet available and will be resolved at execution time by the Flyte engine.

@workflow
def my_wf(a: int) -> int:
    result = t1(a=a)
    # result is a named-tuple: Outputs(o0=Promise, o1=Promise)
    # access individual outputs by attribute:
    int_output = result.o0
    str_output = result.o1
    return int_output

You can also access outputs by unpacking the named tuple immediately:

@workflow
def my_wf(a: int) -> str:
    _, s = t1(a=a)   # directly unpack by output name
    return s

The named-tuple wrapper around multiple outputs is produced by the create_task_output function (flytekit/core/promise.py:772). This function returns the single Promise directly when there is only one output, or wraps multiple Promise objects in a custom named tuple class when there are more. The named tuple has a with_overrides method that forwards to the first output's with_overrides — this is how t1().with_overrides(...) works even though t1() returns a tuple-like object.

Programmatic node creation with create_node

The create_node function in flytekit/core/node_creation.py creates a Node without going through the @workflow decorator. This is useful when you need explicit control over node ordering or when wiring up dependencies imperatively.

from flytekit import task
from flytekit.core.node_creation import create_node

@task
def producer() -> int:
    return 42

@task
def consumer(val: int) -> str:
    return f"got {val}"

# Inside a @workflow or @dynamic function:
n1 = create_node(producer)
n2 = create_node(consumer, val=n1.o0)

create_node calls the entity in compilation mode, which triggers flyte_entity_call_handler and creates a Node added to the compilation state. After the node is created, create_node sets node._outputs to a dict keyed by output name, then attaches each output Promise as both a dict entry (node.outputs[output_name]) and a dynamic attribute (setattr(node, output_name, attr)). This is why you can access n1.o0 on a node returned by create_node.

The key difference from ordinary task-call promises: when you call producer() directly in a workflow body, the returned Promise objects are standalone objects disconnected from the node object. When you call create_node(producer), the node is returned and the promises are attached to it. You can then pass n2 to n1.runs_before(n2) or use the >> operator:

n1 = create_node(producer)
n2 = create_node(consumer, val=n1.o0)
n1 >> n2   # n1 runs before n2

The Node.runs_before method (defined at flytekit/core/node.py:82) appends self to other._upstream_nodes. The >> operator (__rshift__) calls runs_before and returns other, enabling chaining.

Important: Accessing node.outputs on a Node that was not created via create_node raises an AssertionError with the message "Cannot use outputs with all Nodes, node must've been created from create_node()". The _outputs attribute is only set during create_node.

Per-node overrides with with_overrides

Both Node and Promise objects expose a with_overrides method. On Node (defined in flytekit/core/node.py:223), with_overrides can change the node name, set resource requests/limits, configure timeouts and retries, attach an accelerator, mark the node as interruptible, and more. On Promise (defined in flytekit/core/promise.py:584), with_overrides forwards to the underlying node's with_overrides if the promise is not yet ready.

from flytekit import Resources

@workflow
def my_wf(a: int) -> int:
    result = t1(a=a).with_overrides(
        name="first-task",
        timeout=300,        # seconds
        retries=2,
        interruptible=True,
        requests=Resources(cpu="2", mem="4Gi"),
        limits=Resources(cpu="4", mem="8Gi"),
    )
    return result.o0

Using with_overrides on the named-tuple returned by a task call works because create_task_output injects a with_overrides method that forwards to the first output's promise, which in turn forwards to the underlying node.

Failure handlers with on_failure

The on_failure parameter to @workflow registers a task or sub-workflow to execute when the workflow fails. Flytekit validates the handler's interface at decoration time: the handler must accept every workflow input as a parameter (the workflow inputs are a subset of the handler's inputs), and any additional parameters beyond the workflow inputs must be Optional.

The validation logic is in two places:

• ImperativeWorkflow.add_on_failure_handler (flytekit/core/workflow.py:685): validates when using ImperativeWorkflow
• PythonFunctionWorkflow._validate_add_on_failure_handler (flytekit/core/workflow.py:789): validates when using @workflow

Both methods check that failure_node_inputs | workflow_inputs == failure_node_inputs (workflow inputs are a subset of handler inputs) and that any additional keys in the handler's interface are optional types. If these constraints are violated, a FlyteFailureNodeInputMismatchException is raised at compile time.

A valid failure handler looks like this:

from typing import Optional
from flytekit import workflow, task
from flytekit.types.error import FlyteError

@task
def clean_up(name: str, err: Optional[FlyteError] = None) -> None:
    print(f"Cleaning up after failure in {err.failed_node_id}")
    print(f"Error message: {err.message}")

@task
def risky_task(name: str) -> str:
    raise ValueError("something went wrong")

@workflow(on_failure=clean_up)
def wf(name: str = "default") -> str:
    return risky_task(name=name)

The err parameter must be Optional[FlyteError] (flytekit injects it at runtime when the task/workflow has an err parameter in its interface). Workflow inputs passed to the failure handler use the same Promise bindings as the main workflow nodes.

Imperative workflows with ImperativeWorkflow

The ImperativeWorkflow class (flytekit/core/workflow.py:399) provides a fully programmatic alternative to the @workflow decorator. You add inputs, tasks, and outputs explicitly:

from flytekit import task
from flytekit.core.workflow import ImperativeWorkflow

@task
def t1(a: str) -> str:
    return a + " world"

wb = ImperativeWorkflow(name="my.imperative.wf")
wb.add_workflow_input("in1", str)
node = wb.add_entity(t1, a=wb.inputs["in1"])
wb.add_workflow_output("wf_out", node.outputs["o0"], str)

Under the hood, add_entity calls create_node, which sets node._outputs. The add_workflow_output method (flytekit/core/workflow.py:644) constructs output bindings from Promise or NodeOutput references. The execute method (flytekit/core/workflow.py:514) runs nodes sequentially using get_promise_map to resolve bindings from previously completed node outputs.

You can also register a failure handler on an imperative workflow:

wb.add_on_failure_handler(cleanup_task)

This uses the same validation logic as the @workflow(on_failure=...) decorator.

Promise attribute access and the attr_path

When you access an attribute on a Promise using . or [], the key is appended to the promise's attr_path list. This path is used during local execution to resolve nested values from the underlying literal. The _append_attr method (flytekit/core/promise.py:700) creates a copy of the promise with the updated path — the original promise is never mutated because it might be referenced in multiple places.

@workflow
def my_wf() -> str:
    result = t1(a=1)
    # Access a nested attribute on the output:
    nested = result.o0.some_field.nested_field
    # nested.attr_path == ["some_field", "nested_field"]

If you try to iterate over a Promise (e.g. in a for loop or range()), Python raises a ValueError saying "Promise objects are not iterable". This is intentional — the Flyte engine does not support iterating over unresolved promise references at compile time. For list-of-promises scenarios, use indexed access (promise[0]) or the @eager workflow decorator.

Thread safety during workflow construction

Promise objects are not thread-safe. The workflow compilation phase mutates promise objects in place (setting their node references). Accessing promises from multiple threads during construction will cause data corruption. Do not use @workflow-decorated functions from within multithreaded code during the compilation phase.