streamlitai-flow-components 0.1.5

A collection of Streamlit components for building AI workflows.

streamlit-ai-workflow-components

A drag-and-connect workflow canvas for Streamlit. Build visual AI agent pipelines: wire together Agents, Teams, Steps, Loops, Parallel branches, Conditions, and Routers, all from Python.

import streamlit as st
from streamlitai_flow_components import (
    AgentNode, StepNode, Edge, workflow_canvas,
)

result = workflow_canvas(
    nodes=[
        AgentNode(id="researcher", name="Researcher", model_name="claude-sonnet-4-6", position=(100, 100)),
        StepNode(id="write", name="Write Draft", position=(400, 100)),
    ],
    edges=[Edge(source="researcher", target="write")],
)

Users can drag nodes to reposition, draw edges between handles, drop Agents onto Steps, and drag Steps into Loops or Parallel containers - all interactions are returned to Python as structured state.

Quick start

"""minimal_demo.py"""
import streamlit as st
from streamlitai_flow_components import (
    AgentNode, StepNode, TeamNode, Edge, workflow_canvas,
)

st.set_page_config(layout="wide")

result = workflow_canvas(
    nodes=[
        AgentNode(
            id="researcher",
            name="Researcher",
            model_name="claude-sonnet-4-6",
            position=(50, 100),
        ),
        StepNode(
            id="analysis",
            name="Analysis Step",
            description="Drop an agent here to assign it",
            position=(400, 100),
        ),
        TeamNode(
            id="review-team",
            name="Review Team",
            agents=(
                AgentNode(id="reviewer", name="Reviewer", model_name="claude-opus-4-6"),
                AgentNode(id="editor", name="Editor", model_name="claude-haiku-4-5"),
            ),
            mode="Coordinate",
            position=(400, 300),
        ),
    ],
    edges=[Edge(source="researcher", target="analysis")],
    height=500,
    key="my-canvas",
)

if result:
    st.json(result)

streamlit run minimal_demo.py

Using on your project

Install the package from PyPI and use it directly:

pip install streamlitai-flow-components
# or: uv add ... / poetry add ...

Node types

AgentNode

A single AI agent. Can live standalone on the canvas or be dropped into a Step as its executor.

AgentNode(
    id="writer",
    name="Writer",
    model_name="claude-sonnet-4-6",
    position=(100, 200),
    metadata={"temperature": 0.7},  # optional
)

Field	Type	Default	Description
id	str	required	Unique identifier
name	str	required	Display name
model_name	str	required	Model identifier shown as badge
position	tuple[float, float]	(0, 0)	Canvas (x, y) coordinates
metadata	dict | None	None	Key-value data

TeamNode

A group of agents with a coordination mode. Can be dropped into a Step.

TeamNode(
    id="review-team",
    name="Review Team",
    agents=(
        AgentNode(id="reviewer", name="Reviewer", model_name="claude-opus-4-6"),
        AgentNode(id="editor", name="Editor", model_name="claude-haiku-4-5"),
    ),
    mode="Coordinate",
    position=(300, 200),
)

Field	Type	Default	Description
id	str	required	Unique identifier
name	str	required	Display name
agents	tuple[AgentNode, ...]	()	Member agents
mode	str | None	"Coordinate"	Coordination mode label
position	tuple[float, float]	(0, 0)	Canvas (x, y) coordinates
metadata	dict | None	None	Key-value data

StepNode

A workflow step that wraps a single executor (Agent or Team). Empty Steps act as drop zones — drag an Agent or Team onto one to assign it. Steps can also be dragged into Loops or Parallel containers.

StepNode(
    id="writing-step",
    name="Write Draft",
    description="Drafts content based on research",
    executor=AgentNode(id="writer", name="Writer", model_name="claude-sonnet-4-6"),
    position=(400, 100),
)

Field	Type	Default	Description
id	str	required	Unique identifier
name	str	required	Display name
description	str | None	None	Description text
executor	AgentNode | TeamNode | None	None	Assigned executor
position	tuple[float, float]	(0, 0)	Canvas (x, y) coordinates
metadata	dict | None	None	Key-value data

LoopNode

A container that repeats an ordered sequence of Steps. Embedded steps can be reordered with up/down controls and ejected back to the canvas.

LoopNode(
    id="refinement-loop",
    name="Refinement Loop",
    steps=(
        StepNode(id="draft", name="Draft", executor=AgentNode(
            id="drafter", name="Drafter", model_name="claude-sonnet-4-6",
        )),
        StepNode(id="critique", name="Critique"),
    ),
    max_iterations=3,
    condition="Until quality score > 0.9",
    position=(600, 100),
)

Field	Type	Default	Description
id	str	required	Unique identifier
name	str	required	Display name
steps	tuple[StepNode, ...]	()	Ordered steps
max_iterations	int	1	Maximum loop count
condition	str | None	None	Exit condition label
position	tuple[float, float]	(0, 0)	Canvas (x, y) coordinates
metadata	dict | None	None	Key-value data

ParallelNode

A container that executes multiple Steps concurrently with outputs joined together. Drag Steps onto it to add them.

ParallelNode(
    id="research-parallel",
    name="Research Parallel",
    steps=(
        StepNode(id="web-search", name="Web Search", executor=AgentNode(
            id="searcher", name="Searcher", model_name="claude-haiku-4-5",
        )),
        StepNode(id="db-lookup", name="DB Lookup"),
    ),
    position=(100, 400),
)

Field	Type	Default	Description
id	str	required	Unique identifier
name	str	required	Display name
steps	tuple[StepNode, ...]	()	Concurrent steps
position	tuple[float, float]	(0, 0)	Canvas (x, y) coordinates
metadata	dict | None	None	Key-value data

ConditionNode

A binary decision gate with True and False output handles. Use source_handle on edges to connect from a specific branch.

ConditionNode(
    id="quality-check",
    name="Quality Check",
    condition="Score > 0.9",
    position=(800, 150),
)

# Connect from the "false" branch
Edge(source="quality-check", target="retry-step", source_handle="false")

Field	Type	Default	Description
id	str	required	Unique identifier
name	str	required	Display name
condition	str	required	Criteria expression
position	tuple[float, float]	(0, 0)	Canvas (x, y) coordinates
metadata	dict | None	None	Key-value data

Source handles: "true", "false"

RouterNode

An N-way routing hub. Each named route gets its own output handle on the right side of the node. Routes can have optional conditions.

from streamlitai_flow_components import Route, RouterNode

RouterNode(
    id="content-router",
    name="Content Router",
    routes=(
        Route(name="Technical", condition="Topic is technical"),
        Route(name="Creative", condition="Topic is creative"),
        Route(name="General"),
    ),
    position=(400, 400),
)

# Connect from a specific route
Edge(source="content-router", target="tech-step", source_handle="Technical")

Field	Type	Default	Description
id	str	required	Unique identifier
name	str	required	Display name
routes	tuple[Route, ...]	()	Named branches
position	tuple[float, float]	(0, 0)	Canvas (x, y) coordinates
metadata	dict | None	None	Key-value data

Route fields: name: str (required), condition: str | None (optional)

Source handles: One per route, using the route name as the handle ID.

Edge

A directed connection between two nodes. Supports handle-specific connections for Condition and Router nodes.

Edge(source="node-a", target="node-b")
Edge(source="condition-1", target="step-2", source_handle="true")
Edge(source="router-1", target="step-3", source_handle="Technical")

Field	Type	Default	Description
source	str	required	Source node ID
target	str	required	Target node ID
id	str | None	None	Custom edge ID (auto-generated if omitted)
source_handle	str | None	None	Source port name
target_handle	str | None	None	Target port name

Canvas function

workflow_canvas(
    nodes: Sequence[WorkflowNode],
    edges: Sequence[Edge] | None = None,
    height: int = 600,
    key: str | None = None,
) -> dict | None

Parameter	Type	Default	Description
nodes	Sequence[WorkflowNode]	required	Node objects to render
edges	Sequence[Edge] | None	None	Connections between nodes
height	int	600	Canvas height in pixels
key	str | None	None	Streamlit widget key for stable identity

Returns None before first interaction, then a dict with updated state:

{
    "nodes": [
        {"id": "...", "type": "agent", "position": {"x": 100, "y": 200}, "data": {...}},
    ],
    "edges": [
        {"id": "...", "source": "a", "target": "b", "sourceHandle": "true"},
    ],
}

Canvas interactions

Action	Result
Drag a node	Repositions it; updated position in returned state
Drag from a handle	Creates a new edge to the target node
Drop Agent/Team onto empty Step	Absorbs it as the Step's executor; edges transfer
Drop Step onto Loop or Parallel	Absorbs the Step into the container
Click X on a Step's executor	Ejects executor back to a standalone node
Click X on embedded step in Loop/Parallel	Ejects step back to the canvas
Up/Down arrows in Loop	Reorders embedded steps
Delete key	Removes selected nodes or edges
Connect from Condition handle	Edges from "True" or "False" output
Connect from Router handle	Edges from any named route output

Full example

See examples/workflow_demo.py for a complete demo with all node types wired together.

streamlit run examples/workflow_demo.py

Development

Requirements: have Python 3.11 or over and uv installed

Setup

git clone <repo-url>
cd streamlit-ai-workflow-components
uv sync --group dev          # Python deps
cd frontend && pnpm install  # JS deps

Dev mode (with hot reload)

# Terminal 1 — Vite dev server
cd frontend && pnpm dev

# Terminal 2 — Streamlit app
STREAMLIT_COMPONENT_DEV=true uv run streamlit run examples/workflow_demo.py

Release mode

cd frontend && pnpm build
uv run streamlit run examples/workflow_demo.py

Tests and linting

uv run pytest tests/python/ -v    # Python tests
uv run ruff check .               # lint
uv run mypy src/                  # type check
cd frontend && pnpm exec tsc --noEmit  # TypeScript check

Arc

Single Streamlit custom component (workflow_canvas) backed by a React Flow canvas. Each Python node type is a frozen dataclass with a to_dict() method that serializes to React Flow node format. The React side renders custom node components and sends state back via Streamlit.setComponentValue() on every interaction.

Python                            React (Vite + React Flow)
------                            -------------------------
AgentNode.to_dict()     ------>   AgentNode + AgentCard
TeamNode.to_dict()      ------>   TeamNode + TeamCard
StepNode.to_dict()      ------>   StepNode + StepCard
LoopNode.to_dict()      ------>   LoopNode + LoopCard
ParallelNode.to_dict()  ------>   ParallelNode + ParallelCard
ConditionNode.to_dict() ------>   ConditionNode + ConditionCard
RouterNode.to_dict()    ------>   RouterNode + RouterCard

workflow_canvas(nodes, edges)
         |
         v
  React Flow canvas renders nodes + edges
         |
         v  (on drag / connect / delete)
  Streamlit.setComponentValue({nodes, edges})
         |
         v
  Returns to Python as dict

License

MIT