---
title: "Python SDK reference"
description: "The tatolab.stream package, a section at a time. Streams and the builder, node declarations, contexts, ports, bag types, GPU tools, logging and the clock."
type: "docs-page"
url: "https://tatolab.com/docs/reference/python-api"
site: "Tatolab"
section: "Reference"
---

# Python SDK reference

> The tatolab.stream package, a section at a time. Streams and the builder, node declarations, contexts, ports, bag types, GPU tools, logging and the clock.

The Python SDK is the `tatolab.stream` package, installed from PyPI as `tatolab-stream`. Everything a stream or a node imports comes from it. It's fully typed, and it imports and type-checks without the runtime, so your editor, type checker and tests work on any machine.

```python
from tatolab.stream import StreamBuilder, VideoFrame, node, stream
```

The package's own source is the definitive reference: each class and protocol documents itself in [`sdk/tatolab-stream/tatolab/stream/`](https://github.com/tatolab/streamlib/tree/main/sdk/tatolab-stream/tatolab/stream).

## Streams

### `@stream`

```python
@stream
def main(stream_builder: StreamBuilder) -> None:
    """What this stream does."""
```

Marks a function as a stream. The function runs once when the stream loads, receives a `StreamBuilder`, and describes the graph. Its name is the stream's name, made URL-safe. Its docstring documents it in your code; `tatolab run` shows the first line when a file holds several streams, and agents don't see it. Use `@stream` bare, without arguments.

### `StreamBuilder`

| Member | |
|---|---|
| `add(node_class, *, name=None, config=None) -> NodeReference` | Adds a node. Pass the class, not an instance. Without `name`, the node is named after its class in lowercase, and a second node of the same class gets `-2`, and so on. A name you give is made URL-safe but never suffixed, so two names that come out the same are an error. `config` is a dict of the node's settings. |
| `connect(source, destination)` | Links an output (`node.output("port")`, or `remote_output(...)`) to an input (`node.input("port")`) |
| `expose(output, level=Exposure.PRIVATE)` | Lets an output be read from outside the stream. Each output can be exposed once. |
| `remote_output(address) -> NodeOutputPortReferenceByAddress` | Names another stream's exposed output on this machine: `"<stream>/<node>/<port>"` |
| `name` | The stream's name |

### `NodeReference`

Returned by `add`. `.output("port")` and `.input("port")` name its ports for `connect` and `expose`.

### `Exposure`

`Exposure.PRIVATE`: readable by other streams and agents on this machine. `Exposure.PUBLIC`: also readable from other machines, by URL.

### `compile_stream_to_graph(stream_function, *, name=None) -> dict`

Runs a stream function on a fresh builder and returns the graph it describes, as the runtime would load it. Useful in tests: assert on the nodes, links and exposures without a runtime.

## Declaring nodes

### `@node`

```python
@node
class Detector: ...

@node(execution="continuous", interval_ms=1000, scheduling="high", description="Polls the sensor")
class SensorReader: ...
```

| Argument | |
|---|---|
| `execution` | `"reactive"` (run when input arrives; the default for a node with inputs), `"continuous"` (run every `interval_ms`), or `"manual"` (the node drives itself). A node without inputs must choose `continuous` or `manual`. |
| `interval_ms` | For `continuous`, the milliseconds between runs. The first run is immediate, and missed ticks are skipped, not queued. |
| `scheduling` | `"high"` asks the operating system to raise the node's priority, and `"realtime"` asks for real-time scheduling. If the system refuses, the node runs at normal priority and the log says so. Most nodes don't need either. |
| `description` | The node's description in the catalog; defaults to the class docstring |

A node class must live in an importable module, not in `stream.py` and not inside a function.

### `@node.input` and `@node.output`

```python
@node.input(delivery_profile="newest", description="Frames to analyze")
def video(self) -> VideoFrame: ...

@node.output(description="One bag per analyzed frame")
def detections(self) -> dict: ...
```

| Argument | |
|---|---|
| `name` | The port's name; defaults to the method's name |
| `delivery_profile` | Inputs only, required: `"newest"` or `"ordered"` |
| `audio_window` | Inputs only, optional: an `AudioWindowContract` (needs `"ordered"`) |
| `description` | Shown to agents in the catalog |

Port names are made URL-safe, and every port on a node, input or output, needs a different name. The method is never called; its return annotation is documentation.

Neither delivery profile guarantees delivery. Under sustained load both drop bags at the reader's port, and a link never blocks its producer.

### `AudioWindowContract`

```python
AudioWindowContract(sample_rate=16_000, dtype="f32", window_size=480)
```

The audio format an input wants; the runtime converts and cuts the audio to match. All arguments are keywords:

| Argument | |
|---|---|
| `sample_rate` | Required. Samples per second. |
| `dtype` | Required. `"f32"` or `"i16"`. |
| `window_size` | Required. Samples per channel in each window. |
| `hop` | Samples between the starts of two windows. Defaults to `window_size`, for back-to-back windows; smaller makes overlapping windows; larger is an error. |
| `channels` | Defaults to the source's own count. Set it only when your code needs a particular count, such as a model trained on mono. |

See [Audio](https://tatolab.com/docs/build/audio).

### Settings

A node's settings are the class named by the annotation on its `__init__`'s `config` parameter: a dataclass, a `TypedDict` or a pydantic model. `Annotated[type, "description"]` on a field describes it in the catalog. A node whose `__init__` takes only `self` has no settings.

### Lifecycle methods

All optional:

| Method | Called | Context |
|---|---|---|
| `__init__(self, config)` | Once, when the node's process starts | none |
| `setup(ctx)` | Once, before data flows | `RuntimeContextFullAccess` |
| `start(ctx)` | `manual` nodes only, when the node should begin | `RuntimeContextFullAccess` |
| `process(ctx)` | On input (`reactive`) or on each tick (`continuous`); never for `manual` | `RuntimeContextLimitedAccess` |
| `stop(ctx)` | When the node stops | `RuntimeContextFullAccess` |
| `teardown(ctx)` | Once, at the end | `RuntimeContextFullAccess` |

## Contexts

Both context types carry:

| Member | |
|---|---|
| `inputs` | The node's inputs, a `LinkInputDataReader` |
| `outputs` | The node's outputs, a `LinkOutputDataWriter` |
| `config` | The node's settings, as a dict |
| `time` | The machine's monotonic clock, in nanoseconds |
| `gpu_limited_access` | GPU operations that don't allocate (`GpuContextLimitedAccess`) |
| `node_id` | The node's ID, as `graph` shows it |
| `runtime_id` | The ID of the runtime running the node |

`RuntimeContextFullAccess`, given to `setup`, `start`, `stop` and `teardown`, adds `gpu_full_access` (`GpuContextFullAccess`), which can create GPU resources. `process` gets `RuntimeContextLimitedAccess`, which can't.

## Ports at run time

### `ctx.inputs` (`LinkInputDataReader`)

| Method | Returns |
|---|---|
| `read(port, *, into=None)` | The next bag (the latest, for `newest`), or `None`. `into` casts it: a `TypedDict` passes it through; a dataclass, pydantic model or Tatolab type validates it. |
| `read_with_timestamp(port)` | `(bag, timestamp_ns)`, or `(None, None)` |
| `read_from_inbound_link(port, *, into=None)` | `(bag, link_name)` or `None`, to tell several links into one input apart. A link's name is `<sender's node ID>/<port>`, which changes each time the stream loads, so learn the names with `inbound_link_names` in `setup`. |
| `read_from_inbound_link_with_timestamp(port, *, into=None)` | `(bag, link_name, timestamp_ns)` or `None` |
| `inbound_link_names(port)` | The names of the links into `port`; readable in `setup` |
| `has_data(port)` | Whether a bag is waiting |

### `ctx.outputs` (`LinkOutputDataWriter`)

`write(port, bag, timestamp_ns=None)` sends a dict to every reader of `port`, stamped with `timestamp_ns`, or with the time of the write if you leave it out. Pass the stamp of the frame or block a bag was derived from, so readers can line things up; it's what `read_with_timestamp` returns. A bag holds dicts with string keys, lists and tuples, strings, `bytes`, Python ints and floats, booleans and `None`. Anything else raises `TypeError`, including numpy arrays and most numpy numbers: send `.tolist()` for a few values, `.tobytes()` for many, and `int()` or `float()` for a single numpy number. A bag too big for the link is dropped and counted against the port rather than raised.

## Bag types

### `VideoFrame`

Read with `ctx.inputs.read(port, into=VideoFrame)`.

| Member | |
|---|---|
| `surface_id`, `width`, `height`, `timestamp_ns` | The frame's GPU surface, size and capture time |
| `fps`, `color_info`, `content_light`, `mastering_display` | Optional metadata, when the producer sets it |
| `torch.from_dlpack(frame)` | The pixels as a height × width × 4 uint8 tensor, on the frame's device |
| `with frame.cpu() as pixels:` | The pixels as a numpy array, valid inside the block: the slow path. Read-only unless the frame can take an edit. |
| `with frame.writable() as tensor:` | A GPU tensor for editing the frame in place, through `torch.from_dlpack(tensor)`. Refused when the frame can't take an edit. |
| `VideoFrame.from_bag(bag)` | Validate a bag as a frame |

While the object exists, the producer can't reuse the frame's memory; letting it go releases it. Holding frames for a long time costs the producer memory, and then its own frames, but never slows another node.

### `AudioBlock`

| Member | |
|---|---|
| `samples` | A read-only numpy array, `sample_count` × `channels`, over the block's bytes |
| `interleaved_sample_bytes` | The same samples as raw bytes |
| `sample_rate`, `channels`, `sample_count`, `dtype` | The block's format |
| `first_sample_timestamp_ns` | When the first sample was captured |

To write audio, send a dict with `samples` (interleaved little-endian bytes), `sample_rate`, `channels`, `sample_count`, `first_sample_timestamp_ns` and `dtype` (`"f32"` if left out).

### `EncodedVideoFrame`, `EncodedAudioPacket`

One compressed video frame, or one Opus packet. `EncodedVideoFrame` has `codec`, `annex_b_access_unit_bytes`, `is_sync_point`, `group_index`, `sequence_index`, `width` and `height` (the coded size, which can be padded, such as 1088 for 1080p). `EncodedAudioPacket` has `codec`, `opus_packet_bytes`, `is_sync_point`, `sequence_index`, `sample_rate`, `channels`, `sample_count` and `pre_skip`. A `sequence_index` that jumps by more than one means data was lost; wait for the next sync point.

## GPU

### `GlslPixelEffect`

| Member | |
|---|---|
| `GlslPixelEffect.compile(gpu_full_access, effect_glsl, dials=None)` | In `setup`. `effect_glsl` defines `vec4 effect(vec4 source, ivec2 at)`; `dials` maps names to `"float"`, `"int"`, `"vec2"` or `"vec4"`. |
| `effect.apply_to_frame(gpu_limited_access, frame, dials=None)` | In `process`. Returns the bag of a new RGBA frame. Supply every dial every time. |

See [GPU effects](https://tatolab.com/docs/build/gpu-effects).

### `ModelInputTensorKernel`

| Member | |
|---|---|
| `ModelInputTensorKernel.compile(gpu_full_access, width=None, height=None, fit="stretch", pad_to_multiple_of=None, channel_order="rgb", layout="nchw", dtype="float32", scale=1/255, mean=(0, 0, 0), std=(1, 1, 1))` | In `setup`. `fit` is `"stretch"`, `"letterbox"` or `"pad_bottom_right"` (no resize, so leave out `width` and `height`); `channel_order` is `"rgb"` or `"bgr"`; `layout` is `"nchw"` or `"nhwc"`; `dtype` is `"float32"` or `"float16"`. Each channel value `x`, from 0 to 255, becomes `(x * scale - mean) / std`. |
| `kernel.apply_to_surface(gpu_limited_access, surface) -> ModelInputTensor` | In `process`, with a surface from `gpu_limited_access.resolve_surface(frame.surface_id)` |
| `model_input.tensor_surface` | Use as `with model_input.tensor_surface as tensor:`, then `torch.from_dlpack(tensor)` |
| `model_input.geometry.boxes_to_source(boxes)` | Maps `x0, y0, x1, y1` boxes from the tensor back to frame pixels |

See [Use your model](https://tatolab.com/docs/build/models).

### GPU contexts

`ctx.gpu_full_access` (in `setup` and friends) creates resources: `create_compute_kernel`, `create_graphics_kernel`, `create_ray_tracing_kernel` (Linux), `build_triangles_blas` and `build_tlas`, `create_window`, `kernel_dispatch_batch`, and raw handle export for native code: `export_dma_buf` and `export_opaque_fd` on Linux, `export_iosurface` on macOS.

`ctx.gpu_limited_access` (everywhere) uses them: `resolve_surface`, `claim_surface_against_producer_reuse`, `copy_surface_to_surface`, and acquiring pixel buffers, textures and storage buffers, including from the node's output pool.

These are the building blocks under `GlslPixelEffect` and `ModelInputTensorKernel`. The [camera-compute-kernel](https://github.com/tatolab/streamlib/tree/main/examples/camera-compute-kernel) and [raytracing-showcase](https://github.com/tatolab/streamlib/tree/main/examples/raytracing-showcase) examples show them in full.

## Logging

```python
from tatolab.stream import log

log.info("message", key=value, other=1)
```

`log.trace`, `log.debug`, `log.info`, `log.warn` (or `log.warning`) and `log.error` each take a message and any keyword fields. Records appear in `tatolab run` and `tatolab dev` output and in `tatolab logs`, tagged with the node and stream.

## The clock

| | |
|---|---|
| `ctx.time` | Nanoseconds on the machine's monotonic clock, inside a node |
| `monotonic_now_ns()` | The same clock, anywhere in a node's process |
| `start_monotonic_timer(interval_ns)` | A timer that ticks on that clock without drifting. Use it in a `with` block; `wait(timeout_ms=100)` returns how many ticks fired, or 0 on timeout. |

Every timestamp in a stream comes from this clock, so they compare directly.