An agent should see what's happening now

Plenty of questions are about right now: is anyone at the loading dock, has the oven door been open too long, why did the camera feed go black. Answering one usually means a person opening a dashboard or a video feed and looking. An agent could answer it, if it could read live data, see the picture, and find out why something stopped.

The runtime serves MCP, so an agent connected to it can do those things with the streams you already run.

What an agent gets

• Live data. tap samples, for half a second, any output that something reads: detections, counts, transcripts, readings.
• Eyes. exchange turns a frame into an image the agent can look at.
• History. logs reads a stream’s log records, so what your nodes log with log.info(...) is something the agent can look back through.
• Context. The node catalog describes every node type, with its settings and ports, from your nodes’ docstrings, and the live graph shows how each stream fits together.
• Hands. It can add, connect and remove nodes in a running stream, and load and stop streams.

Three ways to use it

A watcher answers questions about what’s happening from live data: “Is anyone at the loading dock?” It reads the outputs and logs that matter, and looks at a frame when the numbers aren’t enough.

An operator keeps streams healthy: “Why is the window black?” It reads the graph, finds the node that isn’t running, reads its error, and fixes the configuration or tells you what’s wrong.

A builder changes streams: “Add a face blur after the camera and show me a frame.” This needs an agent that can also edit files in your project, such as Claude Code running on that machine. It writes the node, adds it to the running stream, and checks the result.

Give the agent answers, not raw data

An agent does better with outputs that answer questions directly than with every detection. A small node that summarizes, and logs its summary, gives it both a live value and a history:

nodes/scene_summary.py

"""Summarizes what the detector has seen, for agents and dashboards."""

from collections import deque

from tatolab.stream import RuntimeContextLimitedAccess, log, node

SECOND_NS = 1_000_000_000


@node
class SceneSummary:
    """Keeps a one-minute summary of the detector's results: how many of each kind are in view now, and the most at once."""

    @node.input(delivery_profile="newest")
    def detections(self) -> dict: ...

    @node.output(description="Once a second: for each label, the count in view now and the peak over the last minute")
    def summary(self) -> dict: ...

    def __init__(self) -> None:
        self.history: deque[tuple[int, dict[str, int]]] = deque()
        self.next_summary_at_ns = 0

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        detections = ctx.inputs.read("detections")
        if detections is None:
            return
        counts: dict[str, int] = {}
        for found in detections["objects"]:
            counts[found["label"]] = counts.get(found["label"], 0) + 1
        self.history.append((ctx.time, counts))
        while self.history and self.history[0][0] < ctx.time - 60 * SECOND_NS:
            self.history.popleft()

        if ctx.time < self.next_summary_at_ns:
            return
        self.next_summary_at_ns = ctx.time + SECOND_NS
        labels = {label for _, seen in self.history for label in seen}
        summary = {
            label: {"now": counts.get(label, 0), "peak_last_minute": max(seen.get(label, 0) for _, seen in self.history)}
            for label in sorted(labels)
        }
        log.info("scene summary", labels=summary)
        ctx.outputs.write("summary", {"labels": summary}, timestamp_ns=ctx.time)

stream.py

from tatolab.stream import CameraSource, StreamBuilder, stream

from nodes.object_detector import ObjectDetector
from nodes.scene_summary import SceneSummary


@stream
def loading_dock(stream_builder: StreamBuilder) -> None:
    """Watches the loading dock camera: what's in view now, and over the last minute."""
    camera = stream_builder.add(CameraSource, name="dock_camera")
    detector = stream_builder.add(ObjectDetector, name="detector")
    summary = stream_builder.add(SceneSummary, name="summary")
    stream_builder.connect(camera.output("video"), detector.input("video"))
    stream_builder.connect(detector.output("detections"), summary.input("detections"))
    stream_builder.expose(camera.output("video"))
    stream_builder.expose(summary.output("summary"))

ObjectDetector is the detector from Use your model. Now “is anyone at the loading dock?” is the agent reading the latest scene summary records with logs, and “show me” is an exchange of a frame from loading_dock/dock_camera/video. The node’s docstring and the output’s description tell the agent what the summary means. A dashboard in another stream can read loading_dock/summary/summary too, since it’s exposed.

Connect the agent

On the machine that runs the streams:

claude mcp add tatolab -- tatolab mcp

For a camera machine elsewhere, over ssh, with the path where the installer put tatolab:

claude mcp add dock -- ssh dock-box .local/bin/tatolab mcp

See Agents and MCP for every tool, the four built-in recipes, and setup for other clients.

Keep it safe

• What the agent sees leaves the machine. Frames it exchanges, bags it taps, the logs it reads and the node settings in the graph go to the agent’s model provider. Decide for each camera whether that’s acceptable before connecting an agent, and keep secrets out of node settings.
• Control stays local. The runtime’s control socket is reachable only by your user account on that machine; anything remote goes through ssh, with ssh’s keys and permissions.
• Approve changes. Agents can stop and rewire streams. Use your client’s permission settings to approve the tool calls that change things.
• Live changes don’t stick. What an agent changes in a running stream lasts until the stream next loads. To keep a change, put it in stream.py and run tatolab run -d again.

Limits

• Frames are scaled down when exchange hands them to an agent; the result says the frame’s real size. tatolab exchange on the command line saves full resolution.
• A tap sees half a second. For anything slower or rarer, log it; the agent reads logs.
• The agent drives. Tatolab answers tool calls; scheduling regular checks is up to your agent or its framework.

Related guides

• Agents and MCP: tools, recipes and setup.
• Tutorial step 7: connect an agent to the motion watcher.
• Logs: what a node’s log.info calls turn into.