nebo 0.2.7

A modern logging SDK for multi-modal data

Nebo

Nebo is a modern logging SDK for multi-modal data. Decorate your functions with @nb.fn() and call nb.log() to write logs; nebo automatically infers a DAG from your call graph.

Why Nebo?

Nebo offers function-level logging capturing metrics, images, audio, and text at the granularity of individual functions, so you can monitor inputs, outputs, and execution flow of your code. Global logs, or logs not bound to a particular function, are also supported. This enables observability for applications such as:

• Agentic workflows with multimodal data
• DAG-structured data-processing pipelines
• ML training + inference

Features

• Captured log types: text, metrics, images, audio, progress
• Automatically infers a DAG from your call graph
• CLI, MCP and agent skill for AI agent query support
• MCP write tools so external agents can push metrics, images, audio, and text into a run
• Fully self-contained log file per run
• Mobile-first web UI
• Notebook embedding via nb.show() (Jupyter-renderable iframe of any slice of a run)
• One-command deploy to a Hugging Face Space (nebo deploy) with public/private read+write modes

Nebo is in active development and features will roll out according to its core principles.

Installation

pip install nebo

The CLI entry point is nebo:

nebo --help

Quick Start

import nebo as nb

@nb.fn()
def load_data(path: str = "data.csv") -> list[dict]:
    """Load records from a file."""
    records = [{"id": i, "value": i * 0.5} for i in range(100)]
    nb.log(f"Loaded {len(records)} records from {path}")
    return records

@nb.fn()
def transform(records: list[dict]) -> list[dict]:
    """Normalize values."""
    out = []
    for r in nb.track(records, name="transforming"):
        out.append({**r, "value": r["value"] / 50.0})
    nb.log(f"Transformed {len(out)} records")
    nb.log_line("record_count", float(len(out)))
    return out

def run():
    """Main pipeline entry point."""
    records = load_data()
    result = transform(records)
    return result

if __name__ == "__main__":
    run()

Running this writes events to ./.nebo/<timestamp>_<run_id>.nebo. Point nebo serve --logdir ./.nebo at the directory to inspect runs in the web UI. The DAG edges (run -> load_data, load_data -> transform) are inferred automatically from data flow -- no manual wiring required.

Core Concepts

@nb.fn() -- Register a function as a DAG node

Every function decorated with @nb.fn() becomes a node in the pipeline DAG. Edges are inferred from data flow: when a node's return value is passed as an argument to another node, an edge is created from the producer to the consumer.

@nb.fn()
def load_data():
    return [1, 2, 3]

@nb.fn()
def transform(data):
    return [x * 2 for x in data]

def run():
    records = load_data()        # edge: run -> load_data (no data dependency)
    result = transform(records)  # edge: load_data -> transform (data flows from load_data)
    return result

When a child node receives no node-produced arguments, the edge falls back to the calling parent node.

You can use it in several ways:

@nb.fn              # bare decorator
@nb.fn()            # with parentheses
@nb.fn(depends_on=[other_fn])  # with explicit dependencies
@nb.fn(ui={"collapsed": True})  # with per-node UI hints

Class Decoration

@nb.fn() can be applied to classes. All methods are wrapped with scope tracking, and the class name becomes a visual group in the DAG:

@nb.fn()
class Agent:
    def think(self, query):
        nb.log(f"Thinking about: {query}")
        return {"plan": "respond"}

    def act(self, plan):
        nb.log(f"Acting on: {plan}")
        return "result"

agent = Agent()
agent.think("hello")
agent.act({"plan": "respond"})

Methods appear as Agent.think and Agent.act in the DAG, grouped under Agent.

Automatic Materialization

Decorated functions appear in the DAG as soon as they execute for the first time — a call to nb.log(), nb.log_line(), etc. is not required. This keeps dependency chains intact when an intermediate function only orchestrates calls to other nodes without logging anything itself.

depends_on -- Explicit dependency declaration

Some dependencies cannot be detected automatically (shared mutable state, class attributes, global variables). Use depends_on to declare these explicitly:

@nb.fn()
def setup():
    """Initialize shared resources."""
    ...

@nb.fn(depends_on=[setup])
def process():
    """Uses resources initialized by setup."""
    ...

nb.log(message) -- Text logging

Log a message to the current node. Messages appear in the terminal dashboard and are queryable via MCP tools.

@nb.fn()
def train(data):
    nb.log(f"Training on {len(data)} samples")
    for epoch in range(10):
        loss = do_train(data)
        nb.log(f"Epoch {epoch}: loss={loss:.4f}")

Typed metric helpers — nb.log_line / log_bar / log_pie / log_scatter / log_histogram

One function per chart type. The chart type locks on first emission per (loggable, name) pair — reusing a name with a different log_* function raises ValueError.

log_line and log_scatter accumulate over time — every call appends another emission with an auto-incrementing step. log_bar / log_pie / log_histogram are snapshots — re-emitting the same name overwrites the prior value, and they don't take step or tags kwargs.

@nb.fn()
def train(model, data):
    # Line — accumulates; takes step / tags
    for epoch in range(100):
        loss = train_one_epoch(model, data)
        nb.log_line("loss", loss)                                  # scalar
        nb.log_line("lr", 3e-4, tags=["main"])                     # tagged for UI filter

    # Scatter — accumulates too; one or more {label: [(x, y), ...]} per call
    for i, (point, cluster) in enumerate(detections):
        nb.log_scatter("embed_2d", {cluster: [point]})             # step auto-advances

    # Snapshots — overwrite on re-emission, no step / tags
    nb.log_bar("counts", {"cat": 3, "dog": 5})                     # {label: number}
    nb.log_pie("budget", {"prompt": 800, "completion": 200})       # {label: number}
    nb.log_histogram(                                              # {label: list[number]}
        "latencies",
        {"p50": [...], "p95": [...], "p99": [...]},
        colors=True,                                               # palette per label
    )

log_scatter and log_histogram accept colors: bool = False. With colors=True the UI distinguishes labels using the shared palette (in addition to per-label shapes for scatter); not recommended in comparison views, where the palette is reserved for run identity.

Clicking any datapoint on a line or scatter chart in the web UI sets a global step filter — the timeline scrubber switches to Step mode, the active step is marked on every line/scatter chart, and the per-node logs/images/audio panels filter to entries whose step matches. Click the same point again or double-click the scrubber to clear.

nb.log_cfg(cfg) -- Configuration logging

Log configuration for the current node.

@nb.fn()
def train(lr=0.001, epochs=50):
    nb.log_cfg({"lr": lr, "epochs": epochs})
    ...

nb.track(iterable, name=None, total=None) -- Progress tracking

Wrap any iterable for tqdm-like progress tracking.

@nb.fn()
def process(items):
    for item in nb.track(items, name="processing"):
        transform(item)

nb.log_image(image, *, name=None, step=None, points=None, boxes=None, circles=None, polygons=None, bitmasks=None) -- Image logging

Log images (PIL, NumPy arrays, or PyTorch tensors) for visual inspection, with optional geometric labels overlaid. Each label kwarg accepts an nb.labels.<Class> instance — or a list of them, so the same image can carry multiple groups of the same kind in different colors (e.g. predictions vs. ground truth). Raw lists / tensors are rejected with a TypeError pointing at the matching nb.labels.* class.

nb.log_image(
    img,
    name="predictions",
    boxes=[
        nb.labels.Boxes(pred_boxes, color="#22d3ee"),
        nb.labels.Boxes(gt_boxes, color="#22c55e"),
    ],
    points=nb.labels.Points([[10, 20]], color="red"),
)

Available classes: nb.labels.Points ([[x, y], ...]), nb.labels.Boxes ([[x1, y1, x2, y2], ...] xyxy), nb.labels.Circles ([[x, y, r], ...]), nb.labels.Polygons (list of [[x, y], ...]; takes an extra fill: bool = True for filled vs. outline-only), nb.labels.Bitmasks (2D HxW, 3D NxHxW, or list of 2D). Each pairs the geometry with a CSS color string. The UI's Settings pane > "Image labels" section exposes per-(loggable, image, key) visibility and opacity controls.

nb.log_audio(audio, sr=16000, name=None, step=None) -- Audio logging

Log audio data for playback and analysis.

nb.md(description) -- Workflow description

Set a workflow-level description (Markdown supported). Visible in MCP tools and the dashboard.

nb.md("A pipeline that loads images, runs inference, and exports predictions.")

nb.ui() -- Run-level UI defaults

Set default layout and display options for the web UI:

nb.ui(layout="horizontal", view="dag", minimap=True, theme="dark")

CLI Reference

Start the daemon server

nebo serve                              # foreground
nebo serve -d                           # background (daemon mode)
nebo serve --port 3000                  # custom port
nebo serve --no-store                   # disable .nebo file storage
nebo serve --store-dir /data            # write .nebo files into /data
nebo serve --api-token nb_…             # require a token on API requests
nebo serve --read public --write private  # default access modes when token is set

Run a pipeline

Launch pipelines from your shell — the SDK auto-detects a running daemon and connects:

uv run python my_pipeline.py

The SDK prints a banner to stdout on connect:

Nebo daemon fully connected. Your run id is: abc123def456.

Use that run id with the read/write CLI subcommands (nebo runs show, nebo metrics get, etc.). To kill a running pipeline, use the shell (Ctrl+C, kill, pkill).

Load a .nebo file

# local daemon
nebo load .nebo/2026-04-06_143000_run-1.nebo

# remote daemon (e.g. an HF Space) — events read locally, replayed via /events
nebo load run.nebo --url https://user-space.hf.space --api-token nb_…

Deploy the daemon to a Hugging Face Space

pip install 'nebo[deploy]'
huggingface-cli login

# Public dashboard, private writes (defaults). Random token printed once.
nebo deploy --space-id <user>/nebo-test --from-source

# Fully private (read + write require token)
nebo deploy --space-id <user>/private-dash --read private --write private

After the Space builds, point the SDK at it:

import nebo as nb
nb.init(uri="https://<user>-nebo-test.hf.space", api_token="nb_…")
# or set NEBO_URI / NEBO_API_TOKEN in the environment.

Check status, logs, errors

nebo status
nebo logs
nebo logs --run experiment-1 --node train --limit 50
nebo errors
nebo errors --run experiment-1

Stop the daemon

nebo stop

MCP integration

nebo mcp   # print Claude Code MCP config

Tools for AI Agents

Nebo exposes a Q&A surface for AI agents over two parallel transports — the nebo CLI (no setup required) and an MCP server (for clients that prefer it). Run-control deliberately isn't part of the agent surface: the user starts and stops pipelines from their own shell.

Install the agent skill into Claude Code or any AGENTS.md-aware tool:

nebo skill install --platform claude-code --skill runs-qa
nebo skill install --platform agents-md   --skill runs-qa

Observation tools

CLI	MCP	Description
nebo runs list	nebo_get_run_history	All runs with outcomes and timestamps
nebo runs show <id>	nebo_get_run_status	One run's summary + metrics_index
nebo graph show	nebo_get_graph	Full DAG: nodes, edges, execution counts
nebo loggables show <id>	nebo_get_loggable_status	One loggable: logs, metrics, errors, params
nebo logs	nebo_get_logs	Log entries, filterable by loggable and run
nebo metrics get <loggable>	nebo_get_metrics	Metric series — supports --tag / --step filters
nebo errors	nebo_get_errors	All errors with full tracebacks
nebo describe	nebo_get_description	Workflow description + node docstrings

Utility tools

CLI	MCP	Description
nebo load <file>	nebo_load_file	Load a .nebo file into the daemon
nebo runs wait <id>	nebo_wait_for_alert	Block until nb.alert(...) fires at or above --min-level

Write tools

These let an agent push derived data into a run. Each defaults loggable_id to __agent__ — a sandbox loggable namespaced separately from user code's __global__.

CLI	MCP	Description
nebo metrics log --entries-json '[...]'	nebo_log_metric	Push metric points (line / bar / pie / scatter / histogram)
nebo text log --entries-json '[...]'	nebo_log_text	Push text log entries
nebo images log --entries-json '[...]'	nebo_log_image	Push images by path (local file), url, or data (base64)
nebo audio log --entries-json '[...]'	nebo_log_audio	Push audio recordings, same input forms as images

Every read/write subcommand also accepts --url, --port, --api-token (or NEBO_URL / NEBO_PORT / NEBO_API_TOKEN env vars) and --json for machine-readable output.

.nebo File Format

Runs are persisted as .nebo binary files using MessagePack serialization. Each file contains a header (magic, version, metadata) followed by append-only event entries. Use nebo load to replay a file into the daemon.

Architecture

graph LR
    A[Your Python Pipeline] --> B[Nebo SDK<br>@fn, log, track, ...]
    B --> C[Daemon Server<br>FastAPI, port 7861]
    B --> D[Terminal Dashboard<br>Rich]
    C --> E[CLI<br>nebo]
    C --> F[MCP Tools<br>Claude]
    C --> G[Web UI]

Two execution modes:

• Local mode (default): In-process only. No daemon needed.
• Server mode: Events stream to a persistent daemon via HTTP. Use nebo serve to start the daemon.

The daemon can run on your laptop, in CI, or on a Hugging Face Space (nebo deploy). The same SDK code works against any of them — set NEBO_URL and NEBO_API_TOKEN to point at the target. When the daemon enforces auth, every API request must carry the token via the X-Nebo-Token header (HTTP) or the ?token=… query param (browsers / WebSocket).

API Reference

Module: nebo

Function	Signature	Description
fn	@fn(), @fn(depends_on=[...]), @fn(ui={...})	Register a function/class as a DAG node
log	log(message: str)	Log a text message
log_line	log_line(name, value, *, step=None, tags=None)	Log a scalar line-chart datapoint
log_bar	log_bar(name, value)	Bar-chart snapshot ({label: number}); overwrites
log_pie	log_pie(name, value)	Pie-chart snapshot ({label: number}); overwrites
log_scatter	log_scatter(name, value, *, step=None, tags=None, colors=False)	Labeled scatter ({label: list[(x, y)]}); accumulates, step auto-increments
log_histogram	log_histogram(name, value, *, colors=False)	Labeled histogram snapshot ({label: list[number]}); overwrites
log_cfg	log_cfg(cfg: dict)	Log node configuration
log_image	log_image(image, *, name=None, step=None, points=None, boxes=None, circles=None, polygons=None, bitmasks=None)	Log an image (label kwargs accept nb.labels.<Class> instances or lists of them)
log_audio	log_audio(audio, sr=16000, name=None, step=None)	Log audio data
labels	nb.labels.{Points, Boxes, Circles, Polygons, Bitmasks}(data, color)	Image-label dataclasses; each pairs raw geometry with a CSS color
track	track(iterable, name=None, total=None)	Progress tracking
md	md(description: str)	Set workflow description
ui	ui(layout, view, collapsed, minimap, theme)	Set run-level UI defaults
init	init(uri=None, *, dag_strategy="object", flush_interval=0.1, api_token=None, webhook_url=None, webhook_min_level=None)	Manual initialization. uri selects file mode (path, default .nebo/) or network mode (http://… or host:port). Pass api_token (or set NEBO_URI/NEBO_API_TOKEN env vars) to target a remote daemon
show	show(*, run=None, node=None, metric=None, image=None, audio=None, logs=False, dag=False, width="100%", height=600)	Jupyter-renderable iframe of a slice of a run
get_state	get_state() -> SessionState	Access the global state singleton