Front door

Four names. If you are publishing a model, this is the whole API — Adapter declares it, @adapter.on_image is the model, and InferCall is what each request arrives as. Everything else on this site is for the adapter that outgrows them.

opennvr_adapter_sdk.Adapter

Adapter(
    adapter_id: str,
    *,
    version: str = "1.0.0",
    vendor: str = "",
    license: str = "",
    tasks: Sequence[str] = (),
    framework: str = "custom",
    weights: str | PathLike[str] | None = None,
    model_version: str | None = None,
    model_card_url: str | None = None,
    modalities_in: Sequence[str] | None = None,
    modalities_out: Sequence[str] | None = None,
    gpu: bool = False,
    network_egress: Sequence[str] = (),
    max_inflight: int = 1,
    max_body_bytes: int = 8 * 1024 * 1024,
    cost: Cost | None = None,
)

A whole AI adapter: identity, model, inference, lifecycle.

Construct one at module scope, decorate the model loader and the inference handler, and expose :attr:app to your server. The six contract endpoints, auth, correlation ids, Prometheus metrics, body parsing, the failure envelope, the OpenAPI and AsyncAPI documents and the lifespan are inherited from :class:~.adapter_app.AdapterApp.

Source code in opennvr_adapter_sdk/facade.py

def __init__(
    self,
    adapter_id: str,
    *,
    version: str = "1.0.0",
    vendor: str = "",
    license: str = "",
    tasks: Sequence[str] = (),
    framework: str = "custom",
    weights: str | os.PathLike[str] | None = None,
    model_version: str | None = None,
    model_card_url: str | None = None,
    modalities_in: Sequence[str] | None = None,
    modalities_out: Sequence[str] | None = None,
    gpu: bool = False,
    network_egress: Sequence[str] = (),
    max_inflight: int = 1,
    max_body_bytes: int = 8 * 1024 * 1024,
    cost: Cost | None = None,
) -> None:
    if not _ID_RE.match((adapter_id or "").strip()):
        raise ValueError(
            f"Adapter({adapter_id!r}): the id must be kebab-case — "
            f"lowercase letters and digits with single hyphens "
            f"(e.g. 'fall-detection'). It becomes the adapter's name in "
            f"/capabilities, the image name and the KAI-C registration."
        )
    self.id = adapter_id.strip()
    self.version = version
    self.vendor = vendor
    self.license = license
    self.tasks = tuple(tasks)
    self.framework = framework
    #: Path to the weights, when there is a file. Used for the
    #: fingerprint and the reported size, and handy in ``load()``.
    self.weights: str | None = str(weights) if weights else None
    self.model_version = model_version or version
    self.model_card_url = model_card_url
    self._modalities_in = tuple(modalities_in or ())
    self._modalities_out = tuple(modalities_out or ())
    self.gpu = gpu
    self._network_egress = tuple(network_egress)
    self._max_inflight = max_inflight
    #: Deliberately smaller than AdapterApp's own 32 MiB default:
    #: a frame or a clip that large is nearly always a caller
    #: mistake, and the limit is one constructor argument away.
    self._max_body_bytes = max_body_bytes
    self._cost = cost

    self._load_fn: Callable[[], Any] | None = None
    self._handler: _Handler | None = None
    self._hardware_fn: Callable[[Any], Any] | None = None
    self._shutdown_fn: Callable[[Any], Any] | None = None
    self._stream_fn: Callable[[Any], Any] | None = None
    self._service: "_FacadeService | None" = None
    #: ``(path, size, mtime_ns) -> digest``. Re-hashing a
    #: multi-gigabyte weights file on every /health and twice
    #: per /capabilities blew the contract's 1000 ms budget on
    #: a 60s poll; a swapped file changes size or mtime, so
    #: drift detection still sees it.
    self._fp_cache: tuple[tuple[str, int, int], str] | None = None
    self._app: AdapterApp | None = None

fingerprint property

fingerprint: str

sha256 of the weights file, or a deterministic value derived from the adapter's identity when there is no file.

Never None: KAI-C's drift detection skips a null fingerprint, so an adapter with one is silently exempt from the tamper check that protects the operator.

service property

service: AdapterService

The :class:~.service.AdapterService this adapter compiles to — anything that accepts one accepts this.

adapter_app property

adapter_app: AdapterApp

The :class:~.adapter_app.AdapterApp wrapper.

app property

app

The ASGI application — point uvicorn at this.

load

load() -> Callable[[Callable[[], Any]], Callable[[], Any]]

Register the model loader, called once at startup.

Whatever it returns becomes call.model. Import heavy ML libraries inside it, not at module top: a broken dependency then shows up as a red /health with the real error message rather than as a container that will not import.

Source code in opennvr_adapter_sdk/facade.py

def load(self) -> Callable[[Callable[[], Any]], Callable[[], Any]]:
    """Register the model loader, called once at startup.

    Whatever it returns becomes ``call.model``. Import heavy ML
    libraries inside it, not at module top: a broken dependency then
    shows up as a red ``/health`` with the real error message rather
    than as a container that will not import."""

    self._assert_open("a load() hook")

    def decorate(fn: Callable[[], Any]):
        self._load_fn = fn
        return fn

    return decorate

on_image

on_image(*tasks: str) -> Callable[..., Any]

The inference handler for an adapter that takes a frame.

call.image is the JPEG/PNG bytes, call.params the caller's knobs, call.model what load() returned. Return a list of :meth:InferCall.detection items, a result dict, or an :class:~.contract.InferResponse for full control.

Source code in opennvr_adapter_sdk/facade.py

def on_image(self, *tasks: str) -> Callable[..., Any]:
    """The inference handler for an adapter that takes a frame.

    ``call.image`` is the JPEG/PNG bytes, ``call.params`` the
    caller's knobs, ``call.model`` what ``load()`` returned. Return
    a list of :meth:`InferCall.detection` items, a result dict, or an
    :class:`~.contract.InferResponse` for full control."""
    return self._handler_decorator("image", tasks)

on_audio

on_audio(*tasks: str) -> Callable[..., Any]

The inference handler for an adapter that takes an audio clip (call.audio).

Source code in opennvr_adapter_sdk/facade.py

def on_audio(self, *tasks: str) -> Callable[..., Any]:
    """The inference handler for an adapter that takes an audio clip
    (``call.audio``)."""
    return self._handler_decorator("audio", tasks)

on_text

on_text(*tasks: str) -> Callable[..., Any]

The inference handler for a text-in adapter (call.text); no binary upload is parsed.

Source code in opennvr_adapter_sdk/facade.py

def on_text(self, *tasks: str) -> Callable[..., Any]:
    """The inference handler for a text-in adapter (``call.text``);
    no binary upload is parsed."""
    return self._handler_decorator("text", tasks)

on_data

on_data(*tasks: str) -> Callable[..., Any]

The inference handler for any other binary payload (call.data).

Source code in opennvr_adapter_sdk/facade.py

def on_data(self, *tasks: str) -> Callable[..., Any]:
    """The inference handler for any other binary payload
    (``call.data``)."""
    return self._handler_decorator("data", tasks)

check_hardware

check_hardware() -> Callable[..., Any]

Override the derived hardware verdict.

Called with the loaded model; return a :class:~.contract.HardwareEvaluationResponse, a :class:~.contract.HardwareVerdict, a bool, or a (verdict, reasoning) pair. Use it when the model has a real requirement to test — a CUDA device, an NPU, enough RAM.

Source code in opennvr_adapter_sdk/facade.py

def check_hardware(self) -> Callable[..., Any]:
    """Override the derived hardware verdict.

    Called with the loaded model; return a
    :class:`~.contract.HardwareEvaluationResponse`, a
    :class:`~.contract.HardwareVerdict`, a bool, or a
    ``(verdict, reasoning)`` pair. Use it when the model has a real
    requirement to test — a CUDA device, an NPU, enough RAM."""

    self._assert_open("a check_hardware() hook")

    def decorate(fn: Callable[[Any], Any]):
        self._hardware_fn = fn
        return fn

    return decorate

on_shutdown

on_shutdown() -> Callable[..., Any]

Run on the way out, with the loaded model — release a device, close a session.

Source code in opennvr_adapter_sdk/facade.py

def on_shutdown(self) -> Callable[..., Any]:
    """Run on the way out, with the loaded model — release a device,
    close a session."""

    self._assert_open("an on_shutdown() hook")

    def decorate(fn: Callable[[Any], Any]):
        self._shutdown_fn = fn
        return fn

    return decorate

on_stream

on_stream() -> Callable[..., Any]

Implement the §6 WebSocket protocol yourself.

Declaring it advertises streaming in /capabilities and publishes the protocol in the adapter's AsyncAPI document. The handler is called with the raw WebSocket.

Source code in opennvr_adapter_sdk/facade.py

def on_stream(self) -> Callable[..., Any]:
    """Implement the §6 WebSocket protocol yourself.

    Declaring it advertises streaming in ``/capabilities`` and
    publishes the protocol in the adapter's AsyncAPI document. The
    handler is called with the raw WebSocket."""

    self._assert_open("an on_stream() handler")

    def decorate(fn: Callable[[Any], Any]):
        self._stream_fn = fn
        return fn

    return decorate

model_info

model_info() -> ModelInfo

The /capabilities model block, derived from the declaration and the handler that was registered.

Source code in opennvr_adapter_sdk/facade.py

def model_info(self) -> ModelInfo:
    """The ``/capabilities`` model block, derived from the
    declaration and the handler that was registered."""
    kind = self._handler.kind if self._handler else "data"
    size_mb = None
    if self.weights and Path(self.weights).is_file():
        size_mb = round(Path(self.weights).stat().st_size / (1024 * 1024), 2)
    modalities_in = list(self._modalities_in) or [
        {"image": "image", "audio": "audio", "text": "text"}.get(kind, "binary")]
    modalities_out = list(self._modalities_out) or [
        {"image": "bbox_classes", "audio": "text", "text": "text"}.get(kind, "json")]
    return ModelInfo(
        name=self.id,
        version=self.model_version,
        framework=self.framework,
        size_mb=size_mb,
        modalities_in=modalities_in,
        modalities_out=modalities_out,
        fingerprint=self.fingerprint,
    )

run

run(host: str = '0.0.0.0', port: int = 9000) -> None

Serve with uvicorn. Convenience for local runs; in a container, point uvicorn at :attr:app directly.

Source code in opennvr_adapter_sdk/facade.py

def run(self, host: str = "0.0.0.0", port: int = 9000) -> None:  # pragma: no cover
    """Serve with uvicorn. Convenience for local runs; in a
    container, point uvicorn at :attr:`app` directly."""
    import uvicorn

    uvicorn.run(self.app, host=host, port=port)

opennvr_adapter_sdk.InferCall

InferCall(
    payload: dict[str, Any], model: Any, adapter: "Adapter"
)

One inference request — what a handler is called with.

Flat on purpose: the binary payload is an attribute, the caller's params are a dict, and the loaded model is right there. Anything the facade does not model stays available as :attr:payload, which is exactly what :meth:AdapterService.infer would have received.

Source code in opennvr_adapter_sdk/facade.py

def __init__(self, payload: dict[str, Any], model: Any,
             adapter: "Adapter") -> None:
    self.payload = payload
    #: Whatever ``@adapter.load()`` returned — the session, the
    #: pipeline, the weights handle.
    self.model = model
    self._adapter = adapter

body property

body: bytes

The raw binary payload, for any non-text adapter.

params property

params: dict[str, Any]

Everything the caller sent besides the binary body — the adapter's own knobs, plus task and camera_id.

text property

text: str

The prompt or utterance, for a text adapter. Looks at the conventional keys before giving up.

task property

task: str

Which of the adapter's advertised tasks this call is for.

camera_id property

camera_id: str

The camera, when the caller knows it.

param

param(name: str, default: Any = None) -> Any

One caller param, with a default.

Source code in opennvr_adapter_sdk/facade.py

def param(self, name: str, default: Any = None) -> Any:
    """One caller param, with a default."""
    return self.payload.get(name, default)

detection staticmethod

detection(
    label: str,
    confidence: float,
    x: float,
    y: float,
    w: float,
    h: float,
    *,
    track_id: str | int | None = None,
    **attributes: Any,
) -> dict[str, Any]

One §5.1 detection, in the shape every consumer expects.

Coordinates are NORMALIZED (0–1 of the frame) — the single most common thing to get wrong, and the reason a detection that looks right on the adapter shows up in the wrong place on the operator's screen. Divide pixel coordinates by the frame size.

Source code in opennvr_adapter_sdk/facade.py

@staticmethod
def detection(label: str, confidence: float, x: float, y: float,
              w: float, h: float, *, track_id: str | int | None = None,
              **attributes: Any) -> dict[str, Any]:
    """One §5.1 detection, in the shape every consumer expects.

    Coordinates are NORMALIZED (0–1 of the frame) — the single most
    common thing to get wrong, and the reason a detection that looks
    right on the adapter shows up in the wrong place on the
    operator's screen. Divide pixel coordinates by the frame size.
    """
    # Pixel coordinates silently clamped to 1.0 and every box landed
    # in the bottom-right corner, with nothing in the logs to say
    # why. Warn once per adapter instead.
    if any(_out_of_range(v) for v in (x, y, w, h)):
        _warn_pixel_coordinates(label, x, y, w, h)
    item: dict[str, Any] = {
        "label": str(label),
        "confidence": _clamp(confidence),
        "bbox": {"x": _clamp(x), "y": _clamp(y),
                 "w": _clamp(w), "h": _clamp(h)},
    }
    if track_id is not None:
        item["track_id"] = track_id
    if attributes:
        item["attributes"] = attributes
    return item

opennvr_adapter_sdk.Overloaded

Overloaded(
    message: str = "Adapter is at capacity.",
    *,
    retry_after_ms: int = 1000,
)

Bases: Exception

Raise from a handler to shed load: a 503 with retry_after_ms.

The honest way to apply backpressure. KAI-C backs off and retries rather than treating the call as a model failure.

Source code in opennvr_adapter_sdk/facade.py

def __init__(self, message: str = "Adapter is at capacity.",
             *, retry_after_ms: int = 1000) -> None:
    super().__init__(message)
    self.retry_after_ms = retry_after_ms

opennvr_adapter_sdk.ServiceError

ServiceError(
    category: ErrorCategory,
    *,
    code: str,
    message: str,
    transient: bool,
    http_status: int,
    retry_after_ms: int | None = None,
)

Bases: Exception

Carries enough information to construct a §7 FailureEnvelope without re-parsing exception strings in the FastAPI route.

Use prefix-namespaced codes (<adapter>.<code>) for adapter- specific failure modes that aren't in §7.1's canonical set.

Source code in opennvr_adapter_sdk/service.py

def __init__(
    self,
    category: ErrorCategory,
    *,
    code: str,
    message: str,
    transient: bool,
    http_status: int,
    retry_after_ms: int | None = None,
) -> None:
    super().__init__(message)
    self.category = category
    self.code = code
    self.message = message
    self.transient = transient
    self.http_status = http_status
    self.retry_after_ms = retry_after_ms