Write a node

Everything a Python node can declare and do, with working examples of a filter, a logger, a timed source and a source driven by its own thread.

A node is a Python class decorated with @node. It declares ports, optionally takes settings, and does its work in a few methods the runtime calls. If you’d rather learn by building, the tutorial walks through a first node step by step.

The shape of a node

Here’s a complete node that appends every motion report it receives to a CSV file. It shows most of what a node can do:

"""Appends every motion report to a CSV file."""

import csv
from dataclasses import dataclass
from typing import Annotated, TypedDict

from tatolab.stream import RuntimeContextFullAccess, RuntimeContextLimitedAccess, log, node


class MotionReport(TypedDict):
    changed: float
    moving: bool
    timestamp_ns: int


@dataclass
class MotionCsvLoggerConfig:
    path: Annotated[str, "The CSV file to append to, relative to the project folder."] = "motion.csv"


@node
class MotionCsvLogger:
    """Appends every motion report it receives to a CSV file."""

    @node.input(delivery_profile="ordered", description="Motion reports, in the order they were made")
    def motion(self) -> MotionReport: ...

    def __init__(self, config: MotionCsvLoggerConfig) -> None:
        self.path = config.path

    def setup(self, ctx: RuntimeContextFullAccess) -> None:
        self.file = open(self.path, "a", newline="")
        self.writer = csv.writer(self.file)

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        # An ordered input can have several reports waiting; take them all.
        while (report := ctx.inputs.read("motion", into=MotionReport)) is not None:
            self.writer.writerow([report["timestamp_ns"], report["changed"], report["moving"]])
        self.file.flush()

    def teardown(self, ctx: RuntimeContextFullAccess) -> None:
        self.file.close()
        log.info("motion log closed", path=self.path)

Use it in a stream like any other node:

logger = stream_builder.add(MotionCsvLogger, config={"path": "motion.csv"})
stream_builder.connect(detector.output("motion"), logger.input("motion"))

Where nodes live

Put each node class in a module of its own, such as nodes/motion_csv_logger.py, and import it into stream.py. A node class can’t be defined in stream.py itself, or inside a function: every Python node runs in its own process, and that process imports the class by its module path (here, nodes.motion_csv_logger:MotionCsvLogger). The runtime refuses a class it couldn’t import that way, with a message saying how to fix it.

Your node’s process starts in the project folder, with your project’s .venv, so it can import anything you’ve installed with uv add and open files relative to the project.

Ports

Ports are declared by decorating methods. The methods are never called, so their body is ....

@node.input(delivery_profile="newest")
def video(self) -> VideoFrame: ...

@node.output(description="Detections for each frame")
def detections(self) -> dict: ...

The name is the method’s name, unless you pass one: @node.input("video", ...). Every port on a node, input or output, needs its own name, so an effect can’t have both an input and an output called video. A common pattern is video in and processed_video out, or the scaffold’s video_from_upstream and video_to_downstream.
Names are made URL-safe: lowercased, with anything other than letters, digits, -, ., _ and ~ turned into -. A method called Video declares the port video.
The return annotation documents what the port carries, for you and your editor. The runtime never checks it.
description is optional. Agents read it in the node catalog.

Delivery profiles

Every input must choose how it reads. There’s no default, because the right answer depends on what the node does:

	`newest`	`ordered`
What a read returns	The most recent bag; older ones are skipped	The next bag, in the order they were sent
Falling behind	Skips to the present	Drops bags if it falls far behind, and counts them
Typical nodes	Effects, detectors, windows, dashboards	Encoders, recorders, loggers, audio processing

Neither profile ever slows the node that sends the bags. A reader that can’t keep up loses bags at its own port, and tatolab graph shows how many.

Settings

A node’s settings are one class, named by the annotation on the config parameter of __init__. It can be a dataclass, a TypedDict or a pydantic model:

@dataclass
class ThermalAlarmConfig:
    threshold_celsius: Annotated[float, "Temperature that raises the alarm."] = 60.0
    zone: Annotated[str, "Name of the area this camera watches."] = "line 1"

The runtime builds the class from the config dict you pass to stream_builder.add. Fields you leave out take their defaults.
A dataclass refuses keys it doesn’t know, so a typo raises an error when the node starts instead of being ignored. A pydantic model ignores unknown keys unless you set extra="forbid", and a TypedDict takes whatever it’s given.
Annotated[type, "description"] describes a setting. The runtime publishes each node’s settings, with their types, defaults and descriptions, in its node catalog, which is how an agent learns what it can configure.
A node whose __init__ takes only self has no settings, and refuses any config you pass it.

Settings are fixed for the life of the node. To change one, edit stream.py; tatolab dev reloads the stream when you save.

What the runtime calls

All of these are optional; define the ones you need.

Method	When it runs	Context
`__init__(self, config)`	Once, when the node’s process starts	none
`setup(self, ctx)`	Once, before any data flows. Load models, compile shaders, open files and devices here.	full
`start(self, ctx)`	`manual` nodes only: when the node should begin. Start your own thread here.	full
`process(self, ctx)`	Whenever there’s work: new input, or the next tick of a `continuous` node	limited
`stop(self, ctx)`	When the node stops. Stop what `start` started.	full
`teardown(self, ctx)`	Once, at the end. Close files and release what `setup` acquired.	full

The full context (RuntimeContextFullAccess) can create GPU resources, through ctx.gpu_full_access. The limited one (RuntimeContextLimitedAccess), which process gets, can use GPU resources but not create them. That keeps allocations out of the per-frame path. Both carry:

ctx.inputs and ctx.outputs: the node’s ports.
ctx.config: the node’s settings, as a dict.
ctx.time: the machine’s monotonic clock, in nanoseconds.
ctx.node_id: the node’s ID, as tatolab graph shows it.

Keep setup reasonably quick. If it has to download model weights, download them once beforehand, so a node doesn’t spend its start-up time on the network.

When `process` runs

@node takes an execution mode:

reactive (the default for a node with inputs): process runs when a bag arrives on any input.
continuous: process runs once straight away, then every interval_ms milliseconds. If a run takes longer than the interval, the ticks it missed are skipped, not queued. Use it for sources that poll something, or produce on a schedule.
manual: the runtime never calls process. The node drives itself, usually from a thread it starts in start. Use it for sources that are pushed data, like a socket or a device callback.

A node with no inputs is a source, and has to choose continuous or manual; there’d be nothing to react to.

You can also ask for a scheduling priority. scheduling="high" asks the operating system to raise the node’s priority, and "realtime" asks for real-time scheduling, for a node that must keep a steady pace, such as one producing audio. If the system refuses, the node runs at normal priority and the log says so. Most nodes need neither.

A timed source

This one polls a sensor’s JSON endpoint once a second:

"""Reads a JSON sensor over HTTP once a second."""

import json
import urllib.request
from dataclasses import dataclass

from tatolab.stream import RuntimeContextLimitedAccess, log, node


@dataclass
class HttpSensorConfig:
    url: str = "http://192.168.1.50/reading.json"
    timeout_seconds: float = 0.5


@node(execution="continuous", interval_ms=1000)
class HttpSensor:
    """Polls a sensor's JSON endpoint once a second and publishes each reading."""

    @node.output()
    def reading(self) -> dict: ...

    def __init__(self, config: HttpSensorConfig) -> None:
        self.url = config.url
        self.timeout = config.timeout_seconds

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        try:
            with urllib.request.urlopen(self.url, timeout=self.timeout) as response:
                reading = json.load(response)
        except (OSError, ValueError) as error:
            log.warn("sensor unreachable", url=self.url, error=str(error))
            return
        ctx.outputs.write("reading", {**reading, "timestamp_ns": ctx.time})

A source with its own thread

This one publishes every UDP datagram that arrives on a port. It’s manual: start hands the node’s outputs to a thread, and stop ends it.

"""Publishes UDP datagrams as they arrive."""

import socket
import threading
from dataclasses import dataclass

from tatolab.stream import LinkOutputDataWriter, RuntimeContextFullAccess, monotonic_now_ns, node


@dataclass
class UdpListenerConfig:
    host: str = "127.0.0.1"
    port: int = 9000


@node(execution="manual")
class UdpListener:
    """Publishes every UDP datagram that arrives on a port."""

    @node.output()
    def datagrams(self) -> dict: ...

    def __init__(self, config: UdpListenerConfig) -> None:
        self.address = (config.host, config.port)
        self.stopping = threading.Event()
        self.thread: threading.Thread | None = None

    def start(self, ctx: RuntimeContextFullAccess) -> None:
        outputs = ctx.outputs
        self.thread = threading.Thread(target=self.listen, args=(outputs,), daemon=True)
        self.thread.start()

    def listen(self, outputs: LinkOutputDataWriter) -> None:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            sock.bind(self.address)
            sock.settimeout(0.2)  # wake up regularly to check whether to stop
            while not self.stopping.is_set():
                try:
                    data, sender = sock.recvfrom(65535)
                except TimeoutError:
                    continue
                outputs.write(
                    "datagrams",
                    {"data": data, "sender": sender[0], "timestamp_ns": monotonic_now_ns()},
                )

    def stop(self, ctx: RuntimeContextFullAccess) -> None:
        self.stopping.set()
        if self.thread is not None:
            self.thread.join(timeout=1.0)

Keep stop prompt: the runtime gives a node a few seconds to wind down, so join with a timeout, as here, rather than waiting forever.

Reading inputs

bag = ctx.inputs.read("motion")                       # a dict, or None
report = ctx.inputs.read("motion", into=MotionReport) # typed
bag, timestamp_ns = ctx.inputs.read_with_timestamp("motion")

read returns None when there’s nothing new. Always check.
into= sets how strict the read is. A TypedDict costs nothing: you get the dict, with type hints. A dataclass or pydantic model checks the bag and raises if it doesn’t fit. Use VideoFrame, AudioBlock, EncodedVideoFrame and EncodedAudioPacket for the runtime’s own bag types.
On an ordered input, several bags can be waiting. Read in a loop until read returns None, as the CSV logger does.

Several links into one input

Any number of links can connect to one input. To tell them apart, read with the link’s name:

while (received := ctx.inputs.read_from_inbound_link("readings")) is not None:
    bag, link_name = received  # "<sender's node ID>/<port>"
    self.latest[link_name] = bag

ctx.inputs.inbound_link_names("readings") lists the links in setup, before any data arrives. A link’s name contains the sender’s node ID, which changes each time the stream loads, so learn the names there rather than writing them into your code. Bags from one link arrive in that link’s order; bags from different links can interleave in any order, so compare their timestamps if order matters.

Writing outputs

ctx.outputs.write("detections", {"labels": ["person"], "count": 1}, timestamp_ns=frame.timestamp_ns)

A bag is a dict whose values msgpack can encode: Python ints and floats, strings, booleans, None, lists, dicts and bytes. Numpy numbers need converting with int() or float(). For a numpy array, send array.tobytes() along with its shape and dtype, or array.tolist() if it’s small. Keep bags small; a video bag only names its frame, which stays on the GPU.

Every reader of the output gets the bag. If nothing is connected, the write goes nowhere, and that’s fine.

Every bag carries a timestamp. When you derive a bag from a frame or an audio block, pass the source’s stamp with timestamp_ns=, as above, so downstream nodes can line things up; it’s what read_with_timestamp returns. Leave it out and the stamp is the moment of the write.

Logging

from tatolab.stream import log

log.info("person entered", zone="door", confidence=0.91)
log.warn("camera is slow", fps=4)
log.error("model failed to load", path=self.weights)

Each record carries your message and fields, plus the node and stream it came from. You see them in the terminal running tatolab run or tatolab dev, and with tatolab logs. Anything a node prints with print() is captured too, but log keeps the fields separate and filterable.

Time

Use ctx.time, or monotonic_now_ns() outside process, for anything time-based. It’s the same monotonic clock every node and built-in uses, so timestamps are comparable across the whole stream. To do something “five times a second”, compare ctx.time against the next due time, as the tutorial’s detector does. Don’t sleep in process: a continuous node with interval_ms is the way to run on a schedule.

Test your logic

tatolab-stream imports without the runtime, so your node modules import in a plain test too. Keep the interesting logic in ordinary functions, and test those with pytest:

import numpy


def changed_fraction(previous: numpy.ndarray, current: numpy.ndarray, threshold: float = 25) -> float:
    """The share of pixels whose brightness moved by more than `threshold`."""
    return float((numpy.abs(current - previous) > threshold).mean())

import numpy

from nodes.motion_math import changed_fraction


def test_a_still_picture_has_not_changed():
    frame = numpy.full((90, 160), 100.0)
    assert changed_fraction(frame, frame) == 0.0


def test_a_moved_block_is_counted():
    before = numpy.full((90, 160), 100.0)
    after = before.copy()
    after[0:45, 0:80] = 200.0
    assert changed_fraction(before, after) == 0.25

Run them from the project folder with uv run --with pytest python -m pytest. (python -m puts the project folder on the import path, so from nodes… works; plain pytest can’t find nodes.) The node class stays a thin wrapper that reads, calls your functions, and writes.

GPU effects: nodes that work on pixels.
Use your model: nodes that run inference.
Python SDK reference: every class and method.