Workflow Components¶

A workflow component defines a multi-step automation as a directed graph of nodes and edges. Workflows orchestrate agents, code blocks, transforms, and other components into a coherent process with conditional branching, triggers, and execution tracking.

Workflow Structure¶

A workflow is a graph where nodes are execution steps and edges define the flow between them. The platform supports four workflow types:

Type	Description
`sequential`	Nodes execute in linear order
`parallel`	Multiple branches execute concurrently
`graph`	Full DAG with conditional branching and merging
`single`	A single-node workflow (wrapper around one component)

graph TD
    Start["Trigger<br/>(webhook)"]
    Classify["Classify Ticket<br/>(agent node)"]
    Route{"Priority?"}
    AutoReply["Auto-Reply<br/>(agent node)"]
    Escalate["Escalate<br/>(code block node)"]
    Notify["Notify Team<br/>(code block node)"]
    End["End"]

    Start --> Classify
    Classify --> Route
    Route -->|"low / medium"| AutoReply
    Route -->|"high / critical"| Escalate
    AutoReply --> End
    Escalate --> Notify
    Notify --> End

Creating a Workflow¶

PythonUI

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

workflow = client.workflows.create({
    "name": "Ticket Routing Workflow",
    "slug": "ticket-routing",
    "description": "Classifies incoming tickets and routes to appropriate handler",
    "trigger_type": "webhook",
    "trigger_config": {
        "path": "/tickets/incoming",
        "method": "POST",
    },
    "nodes": [
        {
            "id": "classify",
            "type": "agent",
            "component_ref": "<classifier-agent-uuid>",
            "label": "Classify Ticket",
        },
        {
            "id": "auto_reply",
            "type": "agent",
            "component_ref": "<reply-agent-uuid>",
            "label": "Auto Reply",
        },
        {
            "id": "escalate",
            "type": "codeblock",
            "component_ref": "<escalation-codeblock-uuid>",
            "label": "Escalate to Team",
        },
    ],
    "edges": [
        {
            "source": "classify",
            "target": "auto_reply",
            "condition": "priority in ['low', 'medium']",
        },
        {
            "source": "classify",
            "target": "escalate",
            "condition": "priority in ['high', 'critical']",
        },
    ],
    "tags": ["support", "automation"],
})

Navigate to Components > Workflows > New, or browse the template gallery. Configure trigger, nodes, edges. Click Create.

Nodes¶

Each node in a workflow represents an execution step. Nodes reference a component that performs the actual work.

Node Properties¶

Property	Type	Description
`id`	string	Unique identifier within the workflow
`type`	string	Component type: `agent`, `codeblock`, `transform`, `validation`
`component_ref`	UUID	Reference to the component that executes this step
`label`	string	Human-readable name displayed in the workflow editor
`config`	dict	Optional node-specific configuration overrides

Node Types¶

Workflows support several types of nodes, each backed by a different component type:

Node Type	Backed By	Description
`agent`	Agent	AI agent that reasons about input and produces output
`codeblock`	Code Block	Deterministic code execution
`transform`	Transform	Data reshaping between steps
`validation`	Validation	Data quality gate

Edges¶

Edges connect nodes and define the execution flow. Each edge has a source node, a target node, and an optional condition.

Edge Properties¶

Property	Type	Description
`source`	string	Source node ID
`target`	string	Target node ID
`condition`	string	Python expression evaluated against the source node's output
`label`	string	Optional display label for the edge

Conditional Branching¶

Conditions are Python expressions evaluated against the output of the source node. If the condition evaluates to true, the edge is followed.

edges=[
    # Route based on classification result
    {"source": "classify", "target": "billing_handler",
     "condition": "category == 'billing'"},

    {"source": "classify", "target": "technical_handler",
     "condition": "category == 'technical'"},

    {"source": "classify", "target": "general_handler",
     "condition": "category == 'general'"},
]

When an agent node has conditional edges, the platform can auto-generate an output schema from the field references in the conditions. See Agents -- Structured Output.

Default edges

If no condition is specified on an edge, it acts as a default (always followed). When multiple edges leave a node, edges with conditions are evaluated first. If none match, edges without conditions are followed.

Triggers¶

Every workflow has a trigger that defines how it starts:

Trigger Type	Description	Configuration
`manual`	Triggered by explicit API call or UI action	No additional config
`webhook`	Triggered by an incoming HTTP request	`path`, `method`, `headers`
`schedule`	Triggered on a cron schedule	`cron_expression`, `timezone`
`event`	Triggered by a platform event	`event_type`, `filters`

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

# Scheduled workflow that runs every hour
workflow = client.workflows.create({
    "name": "Hourly Data Sync",
    "slug": "hourly-data-sync",
    "trigger_type": "schedule",
    "trigger_config": {
        "cron_expression": "0 * * * *",
        "timezone": "UTC",
    },
    "nodes": [...],
    "edges": [...],
})

Execution¶

Running a Workflow¶

Workflows with a manual trigger can be executed directly:

from flow_sdk import FlowSDK

sdk = FlowSDK()

result = await sdk.workflows.run(
    workflow_id=workflow["id"],
    input_data={
        "ticket_id": "TK-1234",
        "subject": "Cannot access my account",
        "body": "I've been locked out since yesterday...",
    },
)

print(f"Execution ID: {result['execution_id']}")
print(f"Status: {result['status']}")

Execution Status¶

Workflow executions progress through these statuses:

Status	Description
`queued`	Execution is queued, waiting to start
`running`	Workflow is actively executing nodes
`completed`	All nodes finished successfully
`failed`	A node failed and the workflow halted
`timeout`	Execution exceeded the timeout limit
`cancelled`	Execution was manually cancelled

Monitoring Execution¶

Track execution progress by streaming events:

from flow_sdk import FlowSDK

sdk = FlowSDK()

handle = await sdk.workflows.submit(
    workflow_id=workflow["id"],
    input_data={
        "ticket_id": "TK-1234",
        "subject": "Cannot access my account",
        "body": "I've been locked out since yesterday...",
    },
)

status = await sdk.workflows.get_status(handle["execution_id"])
print(f"Status: {status['state']}")

Workflow Configuration¶

Execution Limits¶

Field	Type	Description
`max_iterations`	int	Maximum number of node executions (prevents infinite loops)
`execution_timeout_seconds`	int	Maximum total execution time

Input and Output Schemas¶

Workflows can define typed input and output contracts using Schema components:

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

workflow = client.workflows.update(workflow["id"], {
    "input_schema_ref": input_schema["id"],
    "output_schema_ref": output_schema["id"],
})

Workflow Templates¶

The platform provides pre-built workflow templates for common patterns. Templates help you get started quickly without building from scratch.

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

# List available templates
templates = client.workflows.list_templates(
    category="data-processing",
    difficulty="beginner",
)

for t in templates:
    print(f"{t['name']} — {t['description']}")

Templates are available in categories like data processing, customer support, compliance, and more. Each template includes pre-configured nodes, edges, and trigger configuration that you can customize.