tantu flow

A visual flow-based programming platform with NATS orchestration. Build data pipelines by wiring nodes together on a canvas — every connection is a NATS pub/sub under the hood, so nodes can run in-process or as independent processes without changing the wiring.

Key features:

• 🎨 Visual flow editor with zoom, drag-and-drop node palette
• 🔄 Hot-sync: edit flows while they're running, changes deploy instantly
• 🚀 Process-mode nodes: run compute-heavy nodes as isolated OS processes
• 📊 Real-time debug feed and node activity monitoring
• 🎯 Bidirectional process control: ping, status, reload code, shutdown
• 🌊 Pure NATS messaging: nodes communicate via publish/subscribe only

Two core files:

• flow_editor.html — complete visual editor (single HTML file, no build step)
• orchestrator.py — backend runtime that manages flows and NATS connections

Quick Start

1. Start NATS server

docker compose up -d
# or without docker:
# brew install nats-server && nats-server

2. Run the orchestrator

pip install -r requirements.txt
python3 orchestrator.py

Listens on http://localhost:8000. The visual editor is served at the root URL. If NATS isn't ready, the orchestrator stays up and retries every 3 seconds — watch the status indicator (top right) in the UI.

3. Open the editor

Visit http://localhost:8000 in your browser to access the visual flow editor.

Alternatively, you can open flow_editor.html directly as a standalone file (double-click or open flow_editor.html). In standalone mode, the backend URL defaults to http://localhost:8000 — change it in the toolbar if your orchestrator runs elsewhere.

Try the example flow:

1. Select "AdvScrnr" from the dropdown (stock screener demo)
2. Click Deploy to validate and persist
3. Click Run ▶ to start the flow
4. Watch the Debug tab for real-time output

Architecture

Message Flow

Every node publishes to flow.<node_id>.out.<port>. When you wire node A to node B, the orchestrator subscribes B to A's output subject. That's it — no direct function calls, no in-process graph traversal, just NATS subjects.

Packet format (JSON dict):

{
    "_id": "abc123",           # unique packet ID
    "_ts": 1234567890.123,     # timestamp
    "payload": <any>,          # your data
    "subject": "flow.xyz.out", # originating subject
    # ... custom fields ...
}

Node Types

Type	Category	Description
inject	source	Fires on interval/manual trigger with fixed payload
nats-in	source	Bridges external NATS subject into flow
function	transform	Pure Python code: payload = packet['payload']
function-process	transform	Runs in separate OS process (true parallelism, crash isolation)
switch	transform	Routes to different ports based on rules (==, >, contains)
delay	transform	Delays message by N milliseconds
http-request	transform	HTTP GET/POST, response goes in packet['payload']
nats-out	sink	Bridges flow out to external NATS subject
debug	sink	Shows in Debug tab and flow._internal.debug subject

Function vs Function-Process Nodes

Function nodes run in the orchestrator process:

# Your code in the editor:
quotes = packet['payload']
filtered = [q for q in quotes if q['price'] > 100]
packet['payload'] = filtered
node.emit(packet)

Gets wrapped automatically:

def _user_fn(packet, node, context):
    quotes = packet['payload']
    filtered = [q for q in quotes if q['price'] > 100]
    packet['payload'] = filtered
    node.emit(packet)
    return packet

Function-process nodes run as independent workers (see worker_template.py):

• Separate OS process with dedicated CPU core
• Can't crash or block the orchestrator
• Full bidirectional control via NATS (ping, reload code, shutdown)
• Status heartbeat every 10 seconds
• Perfect for CPU-intensive tasks (data processing, ML inference, screening)

API Reference

REST Endpoints

GET  /api/flows              list all flow names
GET  /api/flow?name=X        load flow definition
POST /api/deploy?name=X      deploy flow {nodes:[...], wires:[...]}
POST /api/run?name=X         start deployed flow
POST /api/stop               stop running flow
POST /api/inject/{node_id}   manually fire inject node
GET  /api/status             orchestrator + NATS state, per-node stats
POST /api/process/{id}/control  send control command to process node

WebSocket

WS   /ws/debug               live debug feed + wire activity (drives UI animations)

Control Commands for Process Nodes

Send via POST /api/process/{node_id}/control:

{"command": "ping"}           // health check
{"command": "get_status"}     // full status dump
{"command": "reload_code"}    // hot-reload user function
{"command": "shutdown"}       // graceful shutdown

Project Structure

tantu-flow/
├── flow_editor.html           # visual editor (single-file SPA)
├── orchestrator.py            # backend runtime (FastAPI + NATS)
├── worker_template.py         # template for process-mode workers
├── requirements.txt           # Python dependencies
├── docker-compose.yml         # NATS server setup
└── flows/                     # deployed flows
    └── AdvScrnr/              # example: stock screener
        ├── flow.json          # node graph + wires
        ├── n_1_mndx.py        # fetch quotes (yfinance)
        ├── n_3_5me4.py        # screen by criteria
        └── n_5_zi85.py        # rank & format

Writing Function Nodes

Your code has access to three magic variables (no imports needed):

# 'packet' — incoming message dict
symbol = packet['payload']['symbol']

# 'node' — emit packets to output ports
node.emit(packet)                    # default 'out' port
node.emit(other_packet, port='alt')  # named port

# 'context' — persistent dict (survives across fires)
if 'count' not in context:
    context['count'] = 0
context['count'] += 1

No return statement needed — just mutate packet and call node.emit().

Extending with New Node Types

Add to orchestrator.py:

@register_node("my-type")
class MyNode(BaseNode):
    async def handle(self, packet: dict):
        # your logic here
        await self.emit(packet)

Add to NODE_TYPES in flow_editor.html:

"my-type": {
    category: "transform",
    color: "#22d3ee",
    icon: "fa-star",
    label: "My Node",
    outputs: {out: "output"},
    config: [
        {key: "myfield", label: "My Field", type: "text", default: ""}
    ]
}

Notes

• Security: Function nodes run via exec() in-process. Only deploy code you trust.
• Persistence: Last deployed flow saved to flows/<name>/flow.json and auto-deploys on restart once NATS connects.
• Error handling: Node crashes are logged to Debug tab. The rest of the flow continues running.
• Scalability: Single-process by default. For horizontal scaling, run function-process nodes as separate containers — they're already NATS-native.
• No validation: Wire connections don't validate message schemas. If a node expects different data, it will error.

Example: Stock Screener Flow (AdvScrnr)

inject (daily) → fetch quotes (yfinance) → screen criteria → rank → debug

1. Inject fires daily at market close
2. Fetch quotes (function-process): pulls data for S&P 500 via yfinance
3. Screen (function-process): filters by price change %, volume ratio, distance from 52w low
4. Rank (function): sorts by score and formats output
5. Debug: displays top picks in Debug tab

All compute-heavy nodes run as isolated processes with full status monitoring.

License

MIT — do whatever you want with it.

---

🎯 tantu flow — visual + isolated + observable