# Node API Python

> Python Node API

- Skill: `zhanghandong/node-api-python` (Agent Skill)
- Install (CLI): `npx skillmds@latest add zhanghandong/node-api-python`
- Raw SKILL.md: https://api.skillmd.com/api/skills/zhanghandong/node-api-python/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: ZhangHanDong (https://skillmd.com/u/zhanghandong)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/zhanghandong/node-api-python

---


# Python Node API

> Complete guide to building dora nodes in Python

## Installation

```bash
pip install dora-rs
```

## Basic Usage

```python
from dora import Node

node = Node()

for event in node:
    if event["type"] == "INPUT":
        input_id = event["id"]
        data = event["value"]  # numpy array or pyarrow array
        metadata = event["metadata"]

        # Process data...

        # Send output
        node.send_output("output_id", data)

    elif event["type"] == "STOP":
        break
```

## Event Types

```python
event = {
    "type": "INPUT" | "STOP" | "INPUT_CLOSED" | "ERROR",
    "id": "input_name",           # For INPUT events
    "value": data,                # numpy/pyarrow array
    "metadata": {...},            # Timestamp, type info
    "error": "message",           # For ERROR events
}
```

### Handling Events

```python
from dora import Node

node = Node()

for event in node:
    match event["type"]:
        case "INPUT":
            handle_input(event["id"], event["value"])
        case "INPUT_CLOSED":
            print(f"Input {event['id']} closed")
        case "STOP":
            print("Stopping...")
            break
        case "ERROR":
            print(f"Error: {event['error']}")
```

## Sending Outputs

### NumPy Arrays

```python
import numpy as np

# Integer array
data = np.array([1, 2, 3], dtype=np.int32)
node.send_output("numbers", data)

# Float array
data = np.array([1.0, 2.5, 3.14], dtype=np.float64)
node.send_output("floats", data)

# Image (HWC format)
image = np.zeros((480, 640, 3), dtype=np.uint8)
node.send_output("image", image)
```

### PyArrow Arrays

```python
import pyarrow as pa

# String array
data = pa.array(["hello", "world"])
node.send_output("text", data)

# Struct array (for complex data)
data = pa.array([{
    "x": 1.0,
    "y": 2.0,
    "label": "point"
}])
node.send_output("point", data)
```

### With Metadata

```python
metadata = {
    "watermark": 1000,
    "deadline": 5000,
}
node.send_output("data", array, metadata)
```

## Complete Examples

### Camera Node

```python
# camera_node.py
import cv2
import numpy as np
from dora import Node

node = Node()
cap = cv2.VideoCapture(0)

for event in node:
    if event["type"] == "INPUT" and event["id"] == "tick":
        ret, frame = cap.read()
        if ret:
            # Convert BGR to RGB
            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            node.send_output("image", frame_rgb)

    elif event["type"] == "STOP":
        break

cap.release()
```

### Processing Node

```python
# processor_node.py
import numpy as np
from dora import Node

node = Node()

for event in node:
    if event["type"] == "INPUT":
        if event["id"] == "image":
            image = event["value"]

            # Process image
            gray = np.mean(image, axis=2).astype(np.uint8)

            node.send_output("processed", gray)

    elif event["type"] == "STOP":
        break
```

### Detection Node

```python
# detector_node.py
import numpy as np
import pyarrow as pa
from dora import Node
from ultralytics import YOLO

node = Node()
model = YOLO("yolov8n.pt")

for event in node:
    if event["type"] == "INPUT" and event["id"] == "image":
        image = event["value"]

        # Run detection
        results = model(image)

        # Convert to bounding boxes
        boxes = []
        for r in results:
            for box in r.boxes:
                boxes.append({
                    "x1": float(box.xyxy[0][0]),
                    "y1": float(box.xyxy[0][1]),
                    "x2": float(box.xyxy[0][2]),
                    "y2": float(box.xyxy[0][3]),
                    "confidence": float(box.conf[0]),
                    "class": int(box.cls[0]),
                })

        node.send_output("bbox", pa.array(boxes))

    elif event["type"] == "STOP":
        break
```

## Dataflow Configuration

```yaml
nodes:
  - id: camera
    path: camera_node.py
    inputs:
      tick: dora/timer/millis/33
    outputs:
      - image

  - id: processor
    path: processor_node.py
    inputs:
      image: camera/image
    outputs:
      - processed

  - id: detector
    path: detector_node.py
    inputs:
      image: camera/image
    outputs:
      - bbox
```

## Async Python (Experimental)

```python
import asyncio
from dora import Node

async def main():
    node = Node()

    async for event in node:
        if event["type"] == "INPUT":
            # Async processing
            result = await process_async(event["value"])
            node.send_output("result", result)

        elif event["type"] == "STOP":
            break

asyncio.run(main())
```

## Environment Variables

Access environment variables from dataflow config:

```python
import os

model_path = os.environ.get("MODEL_PATH", "default.pt")
device = os.environ.get("DEVICE", "cpu")
```

## Best Practices

1. **Use NumPy for numerical data** - Native support, zero-copy when possible
2. **Use PyArrow for structured data** - Complex types, nested structures
3. **Handle STOP events** - Clean up resources properly
4. **Check input IDs** - Multiple inputs may arrive
5. **Use environment variables** - Keep config in YAML

## Debugging

### Interactive Mode

Run node directly for testing:

```bash
python my_node.py
```

Without `DORA_NODE_CONFIG` set, the node enters interactive mode where you can manually provide inputs.

### Logging

```python
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

for event in node:
    logger.debug(f"Received event: {event['type']}")
```

## Related Skills

- **dataflow-config** - YAML configuration
- **node-api-rust** - Rust alternative
- **hub-nodes** - Pre-built Python nodes