App surfaces

What the app exposes back. Declare a state view or an action and the App Catalog renders it with no frontend of yours; declare a licence gate and core will not enable the app until your code says the key is good.

opennvr_app_sdk.StateView dataclass

StateView(
    name: str,
    label: str,
    kind: str = "metric",
    path: str = "",
    columns: list[str] = list(),
    description: str = "",
    min: float | None = None,
    max: float | None = None,
    unit: str = "",
    warn: float | None = None,
    danger: float | None = None,
    limit: int | None = None,
)

One declarative view over the app's GET /state payload.

The catalog renders these with ZERO app-specific UI code — the same bet as params → config form. An app that exposes richer live state (occupancy per zone, plates deduped, tracks active) declares how to show it instead of shipping a frontend:

kind="metric" A single scalar at path rendered as a stat chip (e.g. path="denylist_size" → "Denylist · 4"). kind="table" A list at path; columns names the keys to show when the rows are dicts. A list of scalars renders as one column. kind="gauge" A numeric path rendered as a horizontal bar between min and max, coloured amber past warn and red past danger (e.g. zone occupancy). A dict-of-numbers renders one gauge per key (per camera / per zone). kind="log" A recent-events feed: path is a list of strings or dicts {message, time, level}; newest limit shown first. kind="gallery" A thumbnail wall: path is a list of dicts {image|url, label, time} — for plate crops, package or doorbell snapshots. image may be a data: URI.

path is a dot-path into the /state dict ("zones", "counters.in"). A missing path renders as an em-dash, never an error — /state is live data and may not have filled in yet.

opennvr_app_sdk.Action dataclass

Action(
    name: str,
    label: str,
    params: list[Param] = list(),
    description: str = "",
    confirm: bool = False,
)

One operator-invokable action on the app's contract surface.

Declared like params, rendered like params: the catalog builds a generic form from params and POSTs it to /actions/{name} on the app — proxied through the server's POST /api/v1/apps/{id}/actions/{name}, which is user-JWT only. The governance boundary is deliberate: actions are operator verbs (search footage, enroll a face); the OpenNVR Agent's service key can read state but can NEVER invoke an action.

confirm=True makes the catalog ask before invoking (for actions with side effects). The app implements the verb by overriding :meth:ContractMixin.on_action.

opennvr_app_sdk.ContractServer

ContractServer(
    *,
    health: Callable[[], dict[str, Any]],
    manifest: Callable[[], dict[str, Any]],
    state: Callable[[], dict[str, Any]],
    openapi: "Callable[[], dict[str, Any]] | None" = None,
    asyncapi: "Callable[[], dict[str, Any]] | None" = None,
    action: "Callable[[str, dict[str, Any]], Any] | None" = None,
    action_token: "str | None" = None,
    ui: "Callable[[], str] | None" = None,
    user_secret: "Callable[[], str | None] | None" = None,
    app_id: "str | None" = None,
    license_verifier: "Callable[[str], Any] | None" = None,
    host: str = "0.0.0.0",
    port: int = 0,
)

Serve the §03 contract endpoints on a background daemon thread.

stdlib-only by design (http.server) — the contract surface must not drag a web framework into every 60-line detector. Three GETs a few times a second is comfortably inside ThreadingHTTPServer territory.

Source code in opennvr_app_sdk/contract.py

def __init__(
    self,
    *,
    health: Callable[[], dict[str, Any]],
    manifest: Callable[[], dict[str, Any]],
    state: Callable[[], dict[str, Any]],
    openapi: "Callable[[], dict[str, Any]] | None" = None,
    asyncapi: "Callable[[], dict[str, Any]] | None" = None,
    action: "Callable[[str, dict[str, Any]], Any] | None" = None,
    action_token: "str | None" = None,
    ui: "Callable[[], str] | None" = None,
    user_secret: "Callable[[], str | None] | None" = None,
    app_id: "str | None" = None,
    license_verifier: "Callable[[str], Any] | None" = None,
    host: str = "0.0.0.0",
    port: int = 0,
) -> None:
    self._host = host
    self._requested_port = int(port)
    self._routes = {"/health": health, "/manifest": manifest, "/state": state}
    if openapi is not None:
        self._routes["/openapi.json"] = openapi
    if asyncapi is not None:
        self._routes["/asyncapi.json"] = asyncapi
    self._action = action
    self._action_token = action_token
    self._ui = ui
    self._user_secret = user_secret
    self._app_id = app_id
    self._license_verifier = license_verifier
    self._server: _ContractHTTPServer | None = None
    self._thread: threading.Thread | None = None

port property

port: int

The actually-bound port (resolves port=0 ephemerals).

opennvr_app_sdk.Entitlement dataclass

Entitlement(
    valid: bool,
    plan: str = "",
    expires_at: str | None = None,
    message: str = "",
    limits: dict[str, Any] = dict(),
)

A licence verdict from :meth:ContractMixin.verify_license.

opennvr_app_sdk.PRICING_MODELS module-attribute

PRICING_MODELS = frozenset(
    {"free", "paid", "subscription", "contact"}
)

opennvr_app_sdk.ENTITLEMENT_MODES module-attribute

ENTITLEMENT_MODES = frozenset({'none', 'license_key'})

opennvr_app_sdk.UserContext dataclass

UserContext(
    user_id: int,
    username: str,
    is_superuser: bool = False,
    cameras: frozenset[int] | None = None,
    manage: frozenset[int] | None = None,
    purpose: str = "",
    raw: dict = dict(),
)

can_see

can_see(camera) -> bool

camera as an int id or a camN handle.

Source code in opennvr_app_sdk/usercontext.py

def can_see(self, camera) -> bool:
    """``camera`` as an int id or a ``camN`` handle."""
    return self.cameras is None or _camera_id(camera) in self.cameras

visible

visible(camera_ids) -> list

Filter a list of ids/handles down to what this user may see.

Source code in opennvr_app_sdk/usercontext.py

def visible(self, camera_ids) -> list:
    """Filter a list of ids/handles down to what this user may see."""
    return [c for c in camera_ids if self.can_see(c)]

opennvr_app_sdk.current_user

current_user() -> UserContext | None

The operator behind the request being served, or None.

Source code in opennvr_app_sdk/usercontext.py

def current_user() -> UserContext | None:
    """The operator behind the request being served, or ``None``."""
    return _CURRENT.get()

opennvr_app_sdk.verify_call_token

verify_call_token(
    token: str | None,
    secret: str | None,
    *,
    audience: str | None = None,
    purpose: str | None = None,
    now: float | None = None,
) -> dict | None

Verify an X-OpenNVR-Call value — core proving a request to this app's write surfaces (/actions/*, /entitlement/verify) is its own, signed with the sha256 of the app's key so no site-wide credential ever reaches the app. purpose must match when given. Returns the claims, or None.

Source code in opennvr_app_sdk/usercontext.py

def verify_call_token(token: str | None, secret: str | None, *,
                      audience: str | None = None, purpose: str | None = None,
                      now: float | None = None) -> dict | None:
    """Verify an ``X-OpenNVR-Call`` value — core proving a request to
    this app's write surfaces (``/actions/*``, ``/entitlement/verify``)
    is its own, signed with the sha256 of the app's key so no site-wide
    credential ever reaches the app. ``purpose`` must match when given.
    Returns the claims, or ``None``."""
    claims = _verified_claims(token, secret, audience=audience, now=now)
    if claims is None:
        return None
    if purpose is not None and claims.get("purpose") != purpose:
        return None
    return claims

opennvr_app_sdk.contract_openapi

contract_openapi(
    manifest: AppManifest, *, port: int | None = None
) -> dict[str, Any]

The OpenAPI 3.1 document for this app's contract server.

Only the paths the app actually serves appear: /ui when the manifest sets has_ui with ui_mode="internal", one /actions/{name} path per declared action, and /entitlement/verify only when entitlement="license_key".

port fills the server URL's default when known (the contract server passes its bound port).

Source code in opennvr_app_sdk/openapi.py

def contract_openapi(manifest: AppManifest, *, port: int | None = None) -> dict[str, Any]:
    """The **OpenAPI 3.1** document for this app's contract server.

    Only the paths the app actually serves appear: ``/ui`` when the
    manifest sets ``has_ui`` with ``ui_mode="internal"``, one
    ``/actions/{name}`` path per declared action, and
    ``/entitlement/verify`` only when ``entitlement="license_key"``.

    ``port`` fills the server URL's default when known (the contract
    server passes its bound port)."""
    paths: dict[str, Any] = {
        "/health": {
            "get": {
                "operationId": "getHealth",
                "summary": "Liveness and pipeline vitals",
                "description": (
                    "Polled by core to drive the App Catalog's status dot. "
                    "Cheap and unauthenticated by design — it exposes no app data."
                ),
                "tags": ["contract"],
                "responses": {
                    "200": {
                        "description": "The app is up.",
                        "content": {_JSON: {
                            "schema": {"$ref": "#/components/schemas/Health"}}},
                    },
                    "500": _error("A snapshot callable raised; the app stays up."),
                },
            }
        },
        "/manifest": {
            "get": {
                "operationId": "getManifest",
                "summary": "The app's declarative identity",
                "description": (
                    "What the catalog renders the app's card, config form, "
                    "state views and actions from, with no app-specific UI code."
                ),
                "tags": ["contract"],
                "responses": {
                    "200": {
                        "description": "The manifest.",
                        "content": {_JSON: {
                            "schema": {"$ref": "#/components/schemas/Manifest"}}},
                    },
                },
            }
        },
        "/state": {
            "get": {
                "operationId": "getState",
                "summary": "Live standing state",
                "description": (
                    "Whatever the app chooses to expose from "
                    "``ContractMixin.state_snapshot``. Read-only, and cheap: "
                    "it is called from the contract server's thread."
                ),
                "tags": ["contract"],
                "responses": {
                    "200": {
                        "description": "The app's live state.",
                        "content": {_JSON: {
                            "schema": {"$ref": "#/components/schemas/State"}}},
                    },
                    "500": _error("state_snapshot raised."),
                },
            }
        },
    }

    if manifest.has_ui and manifest.ui_mode == "internal":
        paths["/ui"] = {
            "get": {
                "operationId": "getUi",
                "summary": "The app's embedded dashboard",
                "description": (
                    "HTML, proxied by core at /api/v1/apps/{id}/ui and rendered "
                    "sandboxed in the catalog."
                ),
                "tags": ["contract"],
                "responses": {
                    "200": {
                        "description": "An HTML fragment or document.",
                        "content": {"text/html": {"schema": {"type": "string"}}},
                    },
                    "500": _error("The UI callable raised."),
                },
            }
        }

    for action in manifest.actions:
        paths[f"/actions/{action.name}"] = {
            "post": {
                "operationId": f"action{_pascal(action.name)}",
                "summary": action.label or action.name,
                "description": (action.description or "")
                + ("\n\nThe catalog asks for confirmation before invoking this."
                   if action.confirm else "")
                + (
                    "\n\nReached only through core's `POST "
                    "/api/v1/apps/{id}/actions/" + action.name + "`, which is "
                    "**user-JWT only**: actions are operator verbs, and the "
                    "OpenNVR Agent's service key can never invoke one."
                ),
                "tags": ["actions"],
                "security": [{"internalKey": [], "callToken": []}],
                "requestBody": {
                    "required": bool(action.params),
                    "content": {_JSON: {"schema": action_body_schema(action)}},
                },
                "responses": {
                    "200": {
                        "description": "The action's result.",
                        "content": {_JSON: {"schema": {
                            "type": "object", "additionalProperties": True}}},
                    },
                    "400": _error("Malformed body, or the app rejected the params."),
                    "401": _error("Missing or bad X-OpenNVR-Call / internal key."),
                    "404": _error("The app does not handle this action."),
                    "413": _error("Body over the 8 MB cap."),
                    "500": _error("The action raised."),
                },
            }
        }

    if manifest.entitlement == "license_key":
        paths["/entitlement/verify"] = {
            "post": {
                "operationId": "verifyEntitlement",
                "summary": "Verify a licence key",
                "description": (
                    "Core asks the app whether a key the administrator entered "
                    "is valid. The verdict is the app's — core stores the key "
                    "encrypted and refuses to enable the app until the app says "
                    "yes. OpenNVR takes no part in the transaction."
                ),
                "tags": ["contract"],
                "security": [{"internalKey": []}],
                "requestBody": {
                    "required": True,
                    "content": {_JSON: {"schema": {
                        "$ref": "#/components/schemas/LicenseKey"}}},
                },
                "responses": {
                    "200": {
                        "description": "The verdict.",
                        "content": {_JSON: {"schema": {
                            "$ref": "#/components/schemas/Entitlement"}}},
                    },
                    "400": _error("Malformed body."),
                    "401": _error("Missing or bad internal key."),
                    "404": _error("This app declares no licence verifier."),
                },
            }
        }

    server: dict[str, Any] = {
        "url": "http://{host}:{port}",
        "description": "The app's contract port on the deployment's internal network.",
        "variables": {
            "host": {"default": manifest.id,
                     "description": "The app's hostname on the compose network."},
            "port": {"default": str(port or 9000),
                     "description": "cfg.contract_port."},
        },
    }

    return {
        "openapi": "3.1.0",
        "info": {
            "title": f"{manifest.name} — app contract",
            "version": manifest.version,
            "summary": manifest.summary or None,
            "description": (
                f"The HTTP surface **{manifest.name}** serves as an OpenNVR app "
                f"(app contract v{CONTRACT_API_VERSION}). Core polls `/health`, "
                "reads `/manifest` at registration, renders `/state` through the "
                "manifest's declared views, and proxies `/actions/*` for "
                "operators.\n\nThis document is generated from the app's own "
                "manifest by `opennvr_app_sdk.openapi`, so it cannot drift from "
                "the app."
            ),
            "license": ({"name": manifest.license} if manifest.license else None),
            "contact": ({"name": manifest.author or manifest.name,
                         "url": manifest.website or None}
                        if (manifest.author or manifest.website) else None),
            "x-opennvr-app-id": manifest.id,
            "x-opennvr-contract-version": CONTRACT_API_VERSION,
        },
        "servers": [server],
        "tags": [
            {"name": "contract",
             "description": "The endpoints every OpenNVR app serves."},
            {"name": "actions",
             "description": "Operator verbs this app declares in its manifest."},
        ],
        "paths": paths,
        "components": _components(manifest),
    }

opennvr_app_sdk.contract_asyncapi

contract_asyncapi(
    manifest: AppManifest, *, publishes: Sequence[str] = ()
) -> dict[str, Any]

The AsyncAPI 3.0 document for this app's NATS surface.

Three groups of channels, all derived from the manifest: what the app subscribes to (subscribes), what it publishes (opennvr.alerts.app.<id>.<camera_id>, one per declared alert type), and the contracted domain events its requires_scopes grant — scopes are how an app asks for PII-bearing events, so they belong in the spec rather than in prose.

publishes names contracted domain events the app emits (app.publishes(...) on the facade), so a consumer can generate a client for them.

Source code in opennvr_app_sdk/openapi.py

def contract_asyncapi(manifest: AppManifest,
                      *, publishes: Sequence[str] = ()) -> dict[str, Any]:
    """The **AsyncAPI 3.0** document for this app's NATS surface.

    Three groups of channels, all derived from the manifest: what the
    app subscribes to (``subscribes``), what it publishes
    (``opennvr.alerts.app.<id>.<camera_id>``, one per declared alert
    type), and the contracted domain events its ``requires_scopes``
    grant — scopes are how an app asks for PII-bearing events, so they
    belong in the spec rather than in prose.

    ``publishes`` names contracted domain events the app emits
    (``app.publishes(...)`` on the facade), so a consumer can generate a
    client for them."""
    channels: dict[str, Any] = {}
    operations: dict[str, Any] = {}

    if manifest.subscribes:
        key, channel, summary = _subscribe_channel(manifest.subscribes)
        channels[key] = channel
        operations["receive" + _pascal(key)] = {
            "action": "receive",
            "channel": {"$ref": f"#/channels/{key}"},
            "summary": summary,
        }

    if manifest.emits:
        channels["alerts"] = {
            "address": f"{DEFAULT_ALERT_SUBJECT_PREFIX}.app.{manifest.id}.{{camera_id}}",
            "title": "Alerts this app fires",
            "description": (
                "The §11.5 alert envelope. Subject segments mirror the alert's "
                "source block, so `opennvr.alerts.app.>` is every app-emitted "
                f"alert and `{DEFAULT_ALERT_SUBJECT_PREFIX}.app.{manifest.id}.>` "
                "is this app's."
            ),
            "parameters": {"camera_id": {
                "description": "The camera the alert is about."}},
            "messages": {"alert": {"$ref": "#/components/messages/Alert"}},
        }
        operations["sendAlerts"] = {
            "action": "send",
            "channel": {"$ref": "#/channels/alerts"},
            "summary": "Fire an operator-visible alert.",
            "description": "Declared alert types: "
                           + ", ".join(f"{a.name} ({a.severity})" for a in manifest.emits),
        }

    for schema in publishes:
        key = "publish" + _pascal(schema)
        channels[key] = {
            "address": f"opennvr.events.{schema}.{{camera_id}}",
            "title": f"{schema} (published)",
            "description": (
                "A contracted domain event this app publishes — how other "
                "apps consume its output without knowing it exists. Defined "
                "in EVENT_CONTRACTS.md."
            ),
            "parameters": {"camera_id": {"description": "The camera."}},
            "messages": {"domainEvent": {
                "$ref": "#/components/messages/DomainEvent"}},
        }
        operations["send" + _pascal(schema)] = {
            "action": "send",
            "channel": {"$ref": f"#/channels/{key}"},
            "summary": f"Publish {schema}.",
        }

    for scope in manifest.requires_scopes:
        if ":" not in scope:
            continue
        name = scope.split(":", 1)[1]
        key = "event" + _pascal(name)
        if key in channels or any(
            c.get("address") == f"opennvr.events.{name}.v1.>" for c in channels.values()
        ):
            # Already covered by the subscribe channel — a domain-event
            # consumer names the same subject in both places.
            continue
        channels[key] = {
            "address": f"opennvr.events.{name}.v1.{{camera_id}}",
            "title": f"{name} (contracted domain event)",
            "description": (
                f"Granted by the `{scope}` scope. Domain events are versioned "
                "in the subject and defined in EVENT_CONTRACTS.md; consuming a "
                "PII-bearing one is a declared, granted and audited capability."
            ),
            "parameters": {"camera_id": {"description": "The camera."}},
            "messages": {"domainEvent": {
                "$ref": "#/components/messages/DomainEvent"}},
        }
        operations["receive" + _pascal(name)] = {
            "action": "receive",
            "channel": {"$ref": f"#/channels/{key}"},
            "summary": f"Consume {name} events.",
        }

    return {
        "asyncapi": "3.0.0",
        "info": {
            "title": f"{manifest.name} — bus surface",
            "version": manifest.version,
            "description": (
                f"What **{manifest.name}** consumes from and publishes to the "
                "OpenNVR event bus. Generated from the app's manifest by "
                "`opennvr_app_sdk.openapi`; the normative definitions of the "
                "envelopes are EVENT_CONTRACTS.md (domain events) and §11.5 "
                "(alerts)."
            ),
            "license": ({"name": manifest.license} if manifest.license else None),
        },
        "servers": {"bus": {
            "host": "nats:4222",
            "protocol": "nats",
            "description": (
                "The deployment's NATS event bus. Apps connect with their own "
                "credential to the apps-facing server, never with the site key."
            ),
        }},
        "channels": channels,
        "operations": operations,
        "components": {"messages": {
            "InferenceCompleted": {
                "name": "InferenceCompletedEvent",
                "title": "One completed inference",
                "contentType": _JSON,
                "payload": {
                    "type": "object",
                    "properties": {
                        "correlation_id": {"type": "string"},
                        "adapter": {"type": "string"},
                        "adapter_version": {"type": "string"},
                        "camera_id": {"type": "string"},
                        "model_fingerprint": {"type": "string"},
                        "completed_at": {"type": "string", "format": "date-time"},
                        "result": {
                            "type": "object",
                            "properties": {"detections": {
                                "type": "array",
                                "items": {"$ref": "#/components/schemas/Detection"},
                            }},
                        },
                    },
                    "required": ["camera_id", "result"],
                },
            },
            "Alert": {
                "name": "Alert",
                "title": "An app-emitted alert (§11.5)",
                "contentType": _JSON,
                "payload": {"$ref": "#/components/schemas/Alert"},
            },
            "DomainEvent": {
                "name": "DomainEvent",
                "title": "A contracted domain event",
                "contentType": _JSON,
                "payload": {"$ref": "#/components/schemas/DomainEvent"},
            },
        }, "schemas": {
            "Detection": {
                "type": "object",
                "properties": {
                    "label": {"type": "string"},
                    "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                    "track_id": {"type": ["string", "null"]},
                    "bbox": {
                        "type": "object",
                        "description": "NormalizedBBox — x/y/w/h in 0–1 of the frame.",
                        "properties": {
                            "x": {"type": "number"}, "y": {"type": "number"},
                            "w": {"type": "number"}, "h": {"type": "number"},
                        },
                    },
                },
                "required": ["label"],
            },
            "Alert": {
                "type": "object",
                "properties": {
                    "alert_id": {"type": "string"},
                    "fired_at": {"type": "string", "format": "date-time"},
                    "title": {"type": "string"},
                    "description": {"type": "string"},
                    "severity": {"enum": ["low", "medium", "high", "critical"]},
                    "source": {
                        "type": "object",
                        "properties": {
                            "kind": {"enum": ["app", "adapter", "kai-c"]},
                            "name": {"type": "string"},
                            "version": {"type": "string"},
                        },
                    },
                    "camera_id": {"type": "string"},
                    "correlation_id": {"type": ["string", "null"]},
                    "evidence": {"type": "object", "additionalProperties": True},
                    "tags": {"type": "array", "items": {"type": "string"}},
                },
                "required": ["alert_id", "fired_at", "title", "camera_id", "severity"],
            },
            "DomainEvent": {
                "type": "object",
                "description": "The EVENT_CONTRACTS.md envelope.",
                "properties": {
                    "id": {"type": "string"},
                    "schema": {"type": "string"},
                    "correlation_id": {"type": ["string", "null"]},
                    "camera_id": {"type": "string"},
                    "ts": {"type": "string", "format": "date-time"},
                    "producer": {"type": "string"},
                    "payload": {"type": "object", "additionalProperties": True},
                },
                "required": ["id", "schema", "camera_id", "ts", "producer", "payload"],
            },
        }},
    }

opennvr_app_sdk.CONTRACT_API_VERSION module-attribute

CONTRACT_API_VERSION = '1.3'

opennvr_app_sdk.proxy_address

proxy_address(
    scheme: str = "https",
) -> tuple[str, int] | None

(host, port) of the proxy, or None.

Source code in opennvr_app_sdk/egress.py

def proxy_address(scheme: str = "https") -> tuple[str, int] | None:
    """``(host, port)`` of the proxy, or ``None``."""
    url = proxy_url(scheme)
    if not url:
        return None
    parts = urlsplit(url if "://" in url else f"http://{url}")
    if not parts.hostname:
        return None
    return parts.hostname, parts.port or 3128

opennvr_app_sdk.connect_via_proxy

connect_via_proxy(
    host: str, port: int
) -> tuple[str, int] | None

Where a plain-TCP client should tunnel to reach host:port: the proxy address, or None to connect directly (no proxy set, or the host is on NO_PROXY).

Source code in opennvr_app_sdk/egress.py

def connect_via_proxy(host: str, port: int) -> tuple[str, int] | None:
    """Where a plain-TCP client should tunnel to reach ``host:port``:
    the proxy address, or ``None`` to connect directly (no proxy set,
    or the host is on ``NO_PROXY``)."""
    if bypasses(host):
        return None
    return proxy_address("https")