teff.observability.api

teff.observability.api

FastAPI router exposing stored traces as a dashboard API.

Mount it against a :class:~teff.observability.exporter.SQLiteExporter that the same process (or another one) writes into::

from fastapi import FastAPI
from teff.observability import SQLiteExporter, dashboard_router

app = FastAPI()
app.include_router(dashboard_router(SQLiteExporter("./traces.db")))

Endpoints:

• GET /obs/ui — the dashboard (ui.html next to this module).
• GET /obs/ui/runs/{run_id} — the run-detail page for browsers (under the dashboard path).
• GET /obs/runs — recent runs (no payloads), with filters (status, name, owner, tag) and pagination (limit/offset); returns {"items": [...], "total": n}.
• GET /obs/runs/{run_id} — a dedicated HTML page for browsers (ui_run.html); returns the full run JSON for API clients.
• PATCH /obs/runs/{run_id} — update tags / notes on a run (body: {"tags": [...], "notes": "..."}).

Security: traces contain full LLM prompts and responses. Do not bind these routers to a non-loopback host without passing an auth dependency — every dashboard read and every ingest write leaks conversation data otherwise. attach_dashboard(..., auth=require_api_key) is the intended pattern for a public deployment.

Run ids are the stable Run.run_id uuid4 values assigned when a run is collected and shared across all exporters.

Classes:

Name	Description
RunPatch	Fields updatable on an existing run.

Functions:

Name	Description
attach_dashboard	Mount the trace dashboard on an existing FastAPI app.
attach_ingest	Mount the trace ingest endpoint on an existing FastAPI app.
dashboard_router	Build the trace dashboard router over exporter.
ingest_router	Build the trace ingest router over exporter.

RunPatch

Bases: BaseModel

Fields updatable on an existing run.

Source code in teff/observability/api.py

class RunPatch(BaseModel):
    """Fields updatable on an existing run."""

    tags: list[str] | None = None
    notes: str | None = None

attach_dashboard

attach_dashboard(app, exporter, *, prefix='/obs', auth=None)

Mount the trace dashboard on an existing FastAPI app.

Convenience wrapper around :func:dashboard_router for apps that assemble their endpoints elsewhere (e.g. app.include_router)::

attach_dashboard(app, SQLiteExporter("./traces.db"))

Pass auth (a FastAPI dependency) to protect the dashboard when the app is reachable beyond localhost — traces expose full prompts and responses.

Source code in teff/observability/api.py

def attach_dashboard(
    app: FastAPI,
    exporter: SQLiteExporter,
    *,
    prefix: str = "/obs",
    auth: Callable | None = None,
) -> None:
    """Mount the trace dashboard on an existing FastAPI *app*.

    Convenience wrapper around :func:`dashboard_router` for apps that
    assemble their endpoints elsewhere (e.g. ``app.include_router``)::

        attach_dashboard(app, SQLiteExporter("./traces.db"))

    Pass *auth* (a FastAPI dependency) to protect the dashboard when the app
    is reachable beyond localhost — traces expose full prompts and responses.
    """
    app.include_router(dashboard_router(exporter, prefix=prefix, auth=auth))

attach_ingest

attach_ingest(app, exporter, *, prefix='/obs', auth=None)

Mount the trace ingest endpoint on an existing FastAPI app.

Source code in teff/observability/api.py

def attach_ingest(
    app: FastAPI,
    exporter: SQLiteExporter,
    *,
    prefix: str = "/obs",
    auth: Callable | None = None,
) -> None:
    """Mount the trace ingest endpoint on an existing FastAPI *app*."""
    app.include_router(ingest_router(exporter, prefix=prefix, auth=auth))

dashboard_router

dashboard_router(exporter, *, prefix='/obs', auth=None)

Build the trace dashboard router over exporter.

Mount it anywhere in your FastAPI app, under any prefix::

app.include_router(dashboard_router(exporter))            # /obs/*
app.include_router(dashboard_router(exporter, prefix="/dash"))  # /dash/*

auth is an optional FastAPI dependency (e.g. require_api_key) enforced on every route. Traces contain full prompts/responses — provide one when the dashboard is reachable beyond localhost.

The HTML pages resolve their own links and fetches against prefix, so a custom prefix keeps the UI working.

Source code in teff/observability/api.py

def dashboard_router(
    exporter: SQLiteExporter,
    *,
    prefix: str = "/obs",
    auth: Callable | None = None,
) -> APIRouter:
    """Build the trace dashboard router over *exporter*.

    Mount it anywhere in your FastAPI app, under any prefix::

        app.include_router(dashboard_router(exporter))            # /obs/*
        app.include_router(dashboard_router(exporter, prefix="/dash"))  # /dash/*

    *auth* is an optional FastAPI dependency (e.g. ``require_api_key``)
    enforced on every route.  Traces contain full prompts/responses — provide
    one when the dashboard is reachable beyond localhost.

    The HTML pages resolve their own links and fetches against *prefix*,
    so a custom prefix keeps the UI working.
    """

    router = _router(prefix, auth)

    @router.get("/ui")
    async def ui() -> Any:
        return HTMLResponse(_ui_html(prefix))

    @router.get("/runs")
    async def runs(
        limit: int = Query(20, ge=1, le=500),
        offset: int = Query(0, ge=0),
        status: str | None = Query(None),
        name: str | None = Query(None),
        owner: str | None = Query(None),
        tag: str | None = Query(None),
    ) -> dict[str, Any]:
        return exporter.list_runs(
            limit=limit,
            offset=offset,
            status=status,
            name=name,
            owner=owner,
            tag=tag,
        )

    @router.get("/ui/runs/{run_id}")
    async def ui_run_detail(run_id: str) -> Any:
        """The run-detail page under the dashboard path (``/obs/ui/runs/<id>``).

        The dashboard list links here for browsers; the pure JSON API stays at
        ``/obs/runs/<id>``.  The page resolves its own fetches against the
        dashboard *prefix*, so this URL is independent of the API path.
        """
        run = exporter.get_run(run_id)
        if run is None:
            raise HTTPException(status_code=404, detail="run not found")
        return HTMLResponse(_ui_run_html(run_id, prefix))

    @router.get("/runs/{run_id}")
    async def run_detail(run_id: str, request: Request) -> Any:
        run = exporter.get_run(run_id)
        if run is None:
            raise HTTPException(status_code=404, detail="run not found")
        # Browsers get the dedicated page; API clients (fetch, curl) get JSON.
        if "text/html" in request.headers.get("accept", ""):
            return HTMLResponse(_ui_run_html(run_id, prefix))
        return run

    @router.patch("/runs/{run_id}")
    async def run_patch(run_id: str, patch: RunPatch) -> JSONResponse:
        ok = exporter.update_run(run_id, tags=patch.tags, notes=patch.notes)
        if not ok:
            raise HTTPException(status_code=404, detail="run not found")
        return JSONResponse({"run_id": run_id, "updated": True})

    return router

ingest_router

ingest_router(exporter, *, prefix='/obs', auth=None)

Build the trace ingest router over exporter.

POST {prefix}/ingest accepts a run in :meth:Run.to_dict shape (as produced by an :class:~teff.observability.push.HttpExporter) and persists it, so another machine — or a workflow with no API — can push traces into a shared dashboard::

app.include_router(ingest_router(exporter))

auth is an optional FastAPI dependency enforced on the ingest endpoint. Without one, any network-reachable process can write (and a leaked key on the dashboard can read) arbitrary trace data.

Source code in teff/observability/api.py

def ingest_router(
    exporter: SQLiteExporter,
    *,
    prefix: str = "/obs",
    auth: Callable | None = None,
) -> APIRouter:
    """Build the trace ingest router over *exporter*.

    ``POST {prefix}/ingest`` accepts a run in :meth:`Run.to_dict` shape
    (as produced by an :class:`~teff.observability.push.HttpExporter`) and
    persists it, so another machine — or a workflow with no API — can push
    traces into a shared dashboard::

        app.include_router(ingest_router(exporter))

    *auth* is an optional FastAPI dependency enforced on the ingest endpoint.
    Without one, any network-reachable process can write (and a leaked key on
    the dashboard can read) arbitrary trace data.
    """
    router = _router(prefix, auth)

    @router.post("/ingest")
    async def ingest(payload: dict) -> JSONResponse:
        run = Run.from_dict(payload)
        exporter.export(run)
        return JSONResponse({"status": "ok"})

    return router