Metadata-Version: 2.3
Name: streamlitai-flow-components
Version: 0.1.1
Summary: A collection of Streamlit components for building AI workflows.
Author: Anne Aguirre
Author-email: Anne Aguirre <aaguirre.rdit@gmail.com>
Requires-Dist: streamlit>=1.55.0
Requires-Python: >=3.11
Description-Content-Type: text/markdown

# 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.

```python
import streamlit as st
from streamlit_ai_workflow_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

```python
"""minimal_demo.py"""
import streamlit as st
from streamlit_ai_workflow_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)
```

```bash
streamlit run minimal_demo.py
```

## Node types

### AgentNode

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

```python
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` | Arbitrary key-value data |

### TeamNode

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

```python
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` | Arbitrary 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.

```python
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` | Arbitrary 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.

```python
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` | Arbitrary key-value data |

### ParallelNode

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

```python
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` | Arbitrary 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.

```python
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` | Arbitrary 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.

```python
from streamlit_ai_workflow_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` | Arbitrary 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.

```python
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

```python
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:

```python
{
    "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`](examples/workflow_demo.py) for a complete demo with all node types wired together.

```bash
streamlit run examples/workflow_demo.py
```

## Development

### Setup

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

### Dev mode (hot reload)

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

# Terminal 2 — Streamlit app
uv run streamlit run examples/workflow_demo.py
```

### Release mode

```bash
cd frontend && pnpm build
STREAMLIT_COMPONENT_RELEASE=true uv run streamlit run examples/workflow_demo.py
```

### Tests and linting

```bash
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
```

## Architecture

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
