From dora-skills
Guides Python dora node development: installation, event types (INPUT, STOP, INPUT_CLOSED, ERROR), sending outputs with NumPy/PyArrow arrays, and complete camera/processing node examples.
How this skill is triggered — by the user, by Claude, or both
Slash command
/dora-skills:node-api-pythonThe summary Claude sees in its skill listing — used to decide when to auto-load this skill
> Complete guide to building dora nodes in Python
Complete guide to building dora nodes in Python
pip install dora-rs
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 = {
"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
}
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']}")
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)
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)
metadata = {
"watermark": 1000,
"deadline": 5000,
}
node.send_output("data", array, metadata)
# 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()
# 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
# 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
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
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())
Access environment variables from dataflow config:
import os
model_path = os.environ.get("MODEL_PATH", "default.pt")
device = os.environ.get("DEVICE", "cpu")
Run node directly for testing:
python my_node.py
Without DORA_NODE_CONFIG set, the node enters interactive mode where you can manually provide inputs.
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
for event in node:
logger.debug(f"Received event: {event['type']}")
npx claudepluginhub zhanghandong/dora-skills --plugin dora-skillsProvides behavioral guidelines to reduce common LLM coding mistakes, focusing on simplicity, surgical changes, assumption surfacing, and verifiable success criteria.
Searches, retrieves, and installs Agent Skills from prompts.chat registry using MCP tools like search_skills and get_skill. Activates for finding skills, browsing catalogs, or extending Claude.
Creates, edits, and optimizes skills for Claude Code, including drafting, evaluating with test prompts, iterating on performance, and improving skill descriptions for better triggering accuracy.