Workflows

Workflows are directed graphs of nodes that define what operations to perform and how data flows between them. WorkflowBuilder is the primary interface for constructing workflows.

WorkflowBuilder

Creating a Workflow

from kailash.workflow.builder import WorkflowBuilder

workflow = WorkflowBuilder()

Adding Nodes

The add_node method accepts a node type (string), a unique node ID (string), a configuration dictionary, and optionally a connections specification:

# 3-parameter form (most common)
workflow.add_node("PythonCodeNode", "process", {
    "code": "result = {'value': input_data * 2}"
})

# 4-parameter form (with inline connections)
workflow.add_node("PythonCodeNode", "transform", {
    "code": "result = {'output': data + 1}"
}, {"data": ("process", "result")})

Important

Node IDs must be string literals. Never use variables or f-strings for node IDs.

# CORRECT
workflow.add_node("PythonCodeNode", "my_node", {})

# WRONG -- do not use variables
# workflow.add_node("PythonCodeNode", node_id_var, {})

Connecting Nodes

Use add_connection to wire output from one node to input of another:

workflow.add_connection(
    "source_node",    # source node ID
    "target_node",    # target node ID
    "output_param",   # output parameter name from source
    "input_param"     # input parameter name on target
)

Example with multiple connections:

workflow = WorkflowBuilder()

workflow.add_node("CSVReaderNode", "read", {
    "file_path": "data.csv"
})

workflow.add_node("PythonCodeNode", "filter", {
    "code": """
filtered = [r for r in records if r.get('active')]
result = {'data': filtered}
"""
})

workflow.add_node("PythonCodeNode", "summarize", {
    "code": "result = {'total': len(data)}"
})

# Wire the data flow
workflow.add_connection("read", "filter", "data", "records")
workflow.add_connection("filter", "summarize", "data", "data")

Building and Executing

Always call .build() before execution:

from kailash.runtime import LocalRuntime

with LocalRuntime() as runtime:
    results, run_id = runtime.execute(workflow.build())

Warning

You must always call .build(). Passing the workflow directly will fail:

# CORRECT
results, run_id = runtime.execute(workflow.build())

# WRONG
# results, run_id = runtime.execute(workflow)

Passing Parameters at Execution

Provide runtime parameters to nodes:

results, run_id = runtime.execute(
    workflow.build(),
    parameters={
        "my_node": {"param1": "value1", "param2": 42}
    }
)

Parallel Branches

When nodes have no dependencies between them, the runtime can execute them in parallel (especially with AsyncLocalRuntime):

workflow = WorkflowBuilder()

workflow.add_node("PythonCodeNode", "source", {
    "code": "result = {'data': [1, 2, 3, 4, 5]}"
})

# Two independent processing branches
workflow.add_node("PythonCodeNode", "branch_a", {
    "code": "result = {'sum': sum(data)}"
})
workflow.add_node("PythonCodeNode", "branch_b", {
    "code": "result = {'avg': sum(data) / len(data)}"
})

# Merge results
workflow.add_node("PythonCodeNode", "merge", {
    "code": "result = {'sum': sum_result, 'avg': avg_result}"
})

# Fan-out
workflow.add_connection("source", "branch_a", "data", "data")
workflow.add_connection("source", "branch_b", "data", "data")

# Fan-in
workflow.add_connection("branch_a", "merge", "sum", "sum_result")
workflow.add_connection("branch_b", "merge", "avg", "avg_result")

Cyclic Workflows

Enable iterative processing with convergence detection:

from kailash.workflow.builder import WorkflowBuilder
from kailash.runtime import LocalRuntime

workflow = WorkflowBuilder()

workflow.add_node("PythonCodeNode", "iterate", {
    "code": """
x = cycle_state.get('x', 10.0)
new_x = x * 0.9  # Decay toward 0
converged = abs(new_x) < 0.01
result = {'x': new_x, 'converged': converged}
"""
})

with LocalRuntime(enable_cycles=True) as runtime:
    results, run_id = runtime.execute(workflow.build())

Note

Topological sort and cycle edge classification are cached per workflow. The cache is invalidated when you call add_node() or connect().

Conditional Execution

Use SwitchNode for conditional branching:

workflow = WorkflowBuilder()

workflow.add_node("SwitchNode", "router", {
    "condition_field": "category"
})

workflow.add_node("PythonCodeNode", "handle_a", {
    "code": "result = {'handled': 'category A'}"
})

workflow.add_node("PythonCodeNode", "handle_b", {
    "code": "result = {'handled': 'category B'}"
})

with LocalRuntime(conditional_execution="skip_branches") as runtime:
    results, run_id = runtime.execute(workflow.build())

Connection Validation

The runtime validates connections between nodes to prevent parameter injection:

# Three validation modes
runtime = LocalRuntime(connection_validation="strict")  # Block invalid
runtime = LocalRuntime(connection_validation="warn")    # Log warnings
runtime = LocalRuntime(connection_validation="off")     # No validation

In strict mode, the runtime will raise an error if a connection references parameters that do not exist on the target node.

Workflow with Trust

Attach a CARE trust context to any workflow execution:

import os
from dotenv import load_dotenv
load_dotenv()

from kailash.workflow.builder import WorkflowBuilder
from kailash.runtime import LocalRuntime
from kailash.runtime.trust import (
    RuntimeTrustContext,
    TrustVerificationMode,
    TrustVerifier,
    TrustVerifierConfig,
)

workflow = WorkflowBuilder()
workflow.add_node("PythonCodeNode", "secure_process", {
    "code": "result = {'status': 'processed with trust'}"
})

ctx = RuntimeTrustContext(
    trace_id="trace-workflow-001",
    delegation_chain=["human-operator", "agent-orchestrator"],
    verification_mode=TrustVerificationMode.ENFORCING,
)

verifier = TrustVerifier(
    config=TrustVerifierConfig(mode="enforcing"),
)

with LocalRuntime(
    trust_context=ctx,
    trust_verifier=verifier,
    trust_verification_mode="enforcing",
) as runtime:
    results, run_id = runtime.execute(workflow.build())

See CARE Trust Framework for the complete CARE trust documentation.

Best Practices

Always use runtime.execute(workflow.build()) – never skip .build()
Use string literals for node IDs
Read model names from .env – never hardcode "gpt-4" or similar
Use AsyncLocalRuntime in Docker/FastAPI contexts
Use LocalRuntime for CLI/scripts
Enable connection validation in production: connection_validation="strict"
Attach trust context for auditable workflows