The classes underneath

Adapter compiles to these. Write them directly when a model needs more than the decorators give — several loading phases, a /capabilities that varies by host, or health semantics of its own. Same process, same endpoints, same published specs.

opennvr_adapter_sdk.AdapterService

Bases: ABC

The interface every contract-compliant adapter implements.

The methods are deliberately few:

• load() — eagerly load the model. Called once at lifespan startup. is_ready() should return True afterwards.
• is_ready() — used by /health to report loading vs ok.
• model_info() — describes the loaded model. Called on every /capabilities request so live-fingerprint drift detection per §11.3 works automatically.
• hardware_evaluation() — returns the §3.3 verdict + details.
• infer(payload) — runs one inference. Raises ServiceError on failure; the SDK translates to a typed §7 envelope.

Streaming is optional. Default is supports_stream = False and the SDK refuses the WebSocket upgrade with HTTP 501. Override handle_stream to implement the §6 protocol.

app property

app: 'AdapterApp'

The owning AdapterApp. Available after lifespan startup.

metrics property

metrics: 'Metrics'

The metrics registry: the owning AdapterApp's when attached (production — one registry, one /metrics), or a private standalone registry otherwise. The fallback exists so domain instrumentation (register_counter in load(), inc_counter in infer()) never crashes a service driven directly in unit tests without an AdapterApp.

load abstractmethod

load() -> None

Eagerly load the model. Idempotent — safe to call twice.

Implementations should: 1. Read weights from disk / download / connect to upstream. 2. Compute and cache an initial fingerprint. 3. Update internal state so is_ready() returns True.

Exceptions are caught by the implementation and reflected via is_ready() == False plus an error message in hardware_evaluation(); we never let load() exceptions kill the FastAPI app's startup.

Source code in opennvr_adapter_sdk/service.py

@abstractmethod
def load(self) -> None:
    """Eagerly load the model. Idempotent — safe to call twice.

    Implementations should:
      1. Read weights from disk / download / connect to upstream.
      2. Compute and cache an initial fingerprint.
      3. Update internal state so ``is_ready()`` returns True.

    Exceptions are caught by the implementation and reflected via
    ``is_ready() == False`` plus an error message in
    ``hardware_evaluation()``; we never let load() exceptions
    kill the FastAPI app's startup.
    """

is_ready abstractmethod

is_ready() -> bool

True iff the model is loaded and inference is possible.

Source code in opennvr_adapter_sdk/service.py

@abstractmethod
def is_ready(self) -> bool:
    """True iff the model is loaded and inference is possible."""

health_status

health_status() -> HealthStatus | None

Optional: report LOADING vs ERROR, not just "not ready".

is_ready() is a bool, so an adapter whose model FAILED to load looked identical to one still loading — /health said loading forever, Docker's healthcheck passed, and the conformance run went green on a dead adapter. Override this (or use the :class:~.facade.Adapter facade, which does) to answer honestly; returning None keeps the old bool-derived behaviour.

Source code in opennvr_adapter_sdk/service.py

def health_status(self) -> HealthStatus | None:
    """Optional: report LOADING vs ERROR, not just "not ready".

    ``is_ready()`` is a bool, so an adapter whose model FAILED to
    load looked identical to one still loading — ``/health`` said
    ``loading`` forever, Docker's healthcheck passed, and the
    conformance run went green on a dead adapter. Override this (or
    use the :class:`~.facade.Adapter` facade, which does) to answer
    honestly; returning ``None`` keeps the old bool-derived
    behaviour."""
    return None

fingerprint abstractmethod

fingerprint() -> str | None

Live model fingerprint for §11.3 drift detection.

Called on every /capabilities request — KAI-C polls every 60s and uses the returned value to detect tamper (file swap, weights rotation). Return None if the adapter can't compute one (cloud-fronted adapters, unsupported model formats); KAI-C will surface "model identity not verifiable" in the UI.

Source code in opennvr_adapter_sdk/service.py

@abstractmethod
def fingerprint(self) -> str | None:
    """Live model fingerprint for §11.3 drift detection.

    Called on every /capabilities request — KAI-C polls every 60s
    and uses the returned value to detect tamper (file swap,
    weights rotation). Return None if the adapter can't compute
    one (cloud-fronted adapters, unsupported model formats);
    KAI-C will surface "model identity not verifiable" in the UI.
    """

model_info abstractmethod

model_info() -> ModelInfo

Construct the §4 ModelInfo block for /capabilities.

Implementations should call self.fingerprint() for the fingerprint field — DO NOT cache it; live recomputation is the whole point of §11.3 drift detection.

Source code in opennvr_adapter_sdk/service.py

@abstractmethod
def model_info(self) -> ModelInfo:
    """Construct the §4 ``ModelInfo`` block for /capabilities.

    Implementations should call ``self.fingerprint()`` for the
    fingerprint field — DO NOT cache it; live recomputation is
    the whole point of §11.3 drift detection.
    """

hardware_evaluation abstractmethod

hardware_evaluation() -> HardwareEvaluationResponse

Construct the §3.3 HardwareEvaluationResponse.

The adapter decides verdict semantics — local hardware probe, cloud-endpoint ping, model-load status. The contract only standardizes the response shape.

Source code in opennvr_adapter_sdk/service.py

@abstractmethod
def hardware_evaluation(self) -> HardwareEvaluationResponse:
    """Construct the §3.3 ``HardwareEvaluationResponse``.

    The adapter decides verdict semantics — local hardware probe,
    cloud-endpoint ping, model-load status. The contract only
    standardizes the response shape.
    """

infer abstractmethod

infer(payload: dict[str, Any]) -> InferResponse

Run one inference. Return §3.5 InferResponse.

payload is a dict produced by the SDK's body parser:

• For BodyShape.TEXT adapters: the request body (JSON object, or multipart text fields) merged into a flat dict.
• For BodyShape.IMAGE / AUDIO / GENERIC adapters: the binary content lives at payload["__file__"] as bytes, and the parsed params JSON (or the JSON body itself minus the base64 field) is merged at the top level.

Raise ServiceError on every failure path so the SDK can translate to a typed §7 envelope with the correct HTTP status.

Source code in opennvr_adapter_sdk/service.py

@abstractmethod
def infer(self, payload: dict[str, Any]) -> InferResponse:
    """Run one inference. Return §3.5 ``InferResponse``.

    ``payload`` is a dict produced by the SDK's body parser:

    * For ``BodyShape.TEXT`` adapters: the request body (JSON
      object, or multipart text fields) merged into a flat dict.
    * For ``BodyShape.IMAGE`` / ``AUDIO`` / ``GENERIC`` adapters:
      the binary content lives at ``payload["__file__"]`` as
      bytes, and the parsed ``params`` JSON (or the JSON body
      itself minus the base64 field) is merged at the top level.

    Raise ``ServiceError`` on every failure path so the SDK can
    translate to a typed §7 envelope with the correct HTTP
    status.
    """

handle_stream async

handle_stream(websocket: Any) -> None

Override to implement the §6 WS protocol. The SDK calls this from the /infer/stream route AFTER the auth and lifespan readiness checks pass. Only called when the adapter declares supports_stream=True on its AdapterApp.

The websocket arrives un-accepted — handlers must call await websocket.accept() themselves so they can refuse the upgrade (with a §6.5 close code) if e.g. the model isn't loaded yet. The SDK has already wrapped the call with inc_stream_connection / dec_stream_connection, so handlers only need to manage per-frame inc_inflight / record_infer via self.metrics.

Source code in opennvr_adapter_sdk/service.py

async def handle_stream(self, websocket: Any) -> None:  # pragma: no cover
    """Override to implement the §6 WS protocol. The SDK calls
    this from the /infer/stream route AFTER the auth and lifespan
    readiness checks pass. Only called when the adapter declares
    ``supports_stream=True`` on its ``AdapterApp``.

    The websocket arrives un-accepted — handlers must call
    ``await websocket.accept()`` themselves so they can refuse
    the upgrade (with a §6.5 close code) if e.g. the model isn't
    loaded yet. The SDK has already wrapped the call with
    ``inc_stream_connection`` / ``dec_stream_connection``, so
    handlers only need to manage per-frame ``inc_inflight`` /
    ``record_infer`` via ``self.metrics``."""
    raise NotImplementedError(
        "AdapterService.handle_stream() must be overridden when "
        "supports_stream=True is set on AdapterApp."
    )

attach_app

attach_app(app: 'AdapterApp') -> None

Called by AdapterApp at lifespan startup so streaming handlers can reach the SDK's metrics + config.

Subclasses don't normally need to override this — use self.metrics or self.app to read what was attached.

Source code in opennvr_adapter_sdk/service.py

def attach_app(self, app: "AdapterApp") -> None:
    """Called by ``AdapterApp`` at lifespan startup so streaming
    handlers can reach the SDK's metrics + config.

    Subclasses don't normally need to override this — use
    ``self.metrics`` or ``self.app`` to read what was attached.
    """
    self._app = app

opennvr_adapter_sdk.AdapterApp

AdapterApp(
    *,
    service: AdapterService | None = None,
    service_factory: Any = None,
    name: str,
    version: str,
    vendor: str,
    license: str,
    tasks_advertised: Sequence[str],
    body_shape: BodyShape = BodyShape.TEXT,
    max_body_bytes: int = 32 * 1024 * 1024,
    permissions: Permissions | None = None,
    scheduling: Scheduling | None = None,
    cost: Cost | None = None,
    model_card_url: str | None = None,
    supported_contract_versions: Sequence[str] = ("1",),
    extra_input_content_types: Sequence[str] = (),
    latency_buckets_seconds: tuple[
        float, ...
    ] = DEFAULT_LATENCY_BUCKETS_SECONDS,
    cors_origins: Sequence[str] = ("*",),
    supports_stream: bool = False,
    stream_max_concurrent: int = 0,
    stream_supports_shared_memory: bool = False,
)

Wraps an AdapterService in a contract-compliant FastAPI app.

The fastapi_app attribute is what uvicorn loads:

.. code-block:: python

app = AdapterApp(
    service=MyService(),
    name="my-adapter",
    version="1.0.0",
    vendor="me",
    license="MIT",
    tasks_advertised=["my_task"],
).fastapi_app

Source code in opennvr_adapter_sdk/adapter_app.py

def __init__(
    self,
    *,
    service: AdapterService | None = None,
    service_factory: Any = None,
    name: str,
    version: str,
    vendor: str,
    license: str,
    tasks_advertised: Sequence[str],
    body_shape: BodyShape = BodyShape.TEXT,
    max_body_bytes: int = 32 * 1024 * 1024,
    permissions: Permissions | None = None,
    scheduling: Scheduling | None = None,
    cost: Cost | None = None,
    model_card_url: str | None = None,
    supported_contract_versions: Sequence[str] = ("1",),
    extra_input_content_types: Sequence[str] = (),
    latency_buckets_seconds: tuple[float, ...] = DEFAULT_LATENCY_BUCKETS_SECONDS,
    cors_origins: Sequence[str] = ("*",),
    supports_stream: bool = False,
    stream_max_concurrent: int = 0,
    stream_supports_shared_memory: bool = False,
) -> None:
    # ``service`` is the eager case (production typical); the
    # factory is invoked at lifespan startup and useful when the
    # adapter needs late binding (env-var-driven config,
    # test-fixture monkeypatching of __init__, etc.). Exactly one
    # must be supplied.
    if (service is None) == (service_factory is None):
        raise ValueError("AdapterApp requires exactly one of service= or service_factory=.")
    self._service: AdapterService | None = service
    self._service_factory = service_factory
    self._supports_stream = supports_stream
    self._stream_max_concurrent = stream_max_concurrent
    self._stream_supports_shared_memory = stream_supports_shared_memory
    self._name = name
    self._version = version
    self._vendor = vendor
    self._license = license
    self._tasks_advertised = list(tasks_advertised)
    self._body_shape = body_shape
    self._max_body_bytes = max_body_bytes
    self._permissions = permissions or Permissions()
    self._scheduling = scheduling or Scheduling()
    self._cost = cost or Cost()
    self._model_card_url = model_card_url
    self._supported_contract_versions = list(supported_contract_versions)
    self._started_at_dt = datetime.now(timezone.utc)
    self._started_at_mono = time.monotonic()
    # tasks_advertised is the CLOSED task-label set for metrics — task
    # strings arrive in client payloads, so anything unadvertised is
    # folded into "other" (cardinality guard, see Metrics).
    self._metrics = Metrics(
        latency_buckets_seconds=latency_buckets_seconds,
        known_tasks=tuple(self._tasks_advertised),
    )

    self._input_content_types = self._compute_input_content_types(extra_input_content_types)

    self.fastapi_app: FastAPI = self._build_fastapi_app(cors_origins)

replace_service

replace_service(service: AdapterService) -> None

Test-fixture hook: swap the service after construction. Production code uses service= or service_factory= and does not call this.

Source code in opennvr_adapter_sdk/adapter_app.py

def replace_service(self, service: AdapterService) -> None:
    """Test-fixture hook: swap the service after construction.
    Production code uses ``service=`` or ``service_factory=`` and
    does not call this."""
    self._service = service

opennvr_adapter_sdk.BodyShape

Bases: str, Enum

Hint to the SDK's /infer body parser about what to expect.

• TEXT — JSON only; no binary upload. (e.g., Piper TTS)
• IMAGE — multipart with a binary frame field + JSON params, or JSON with frame_b64. (e.g., YOLOv8)
• AUDIO — multipart with a binary audio field + JSON params, or JSON with audio_b64. (e.g., Whisper)
• GENERIC — multipart with a single binary file field + JSON params, or JSON with the binary as data_b64. Use when you don't fit the named patterns.

opennvr_adapter_sdk.BODY_BYTES_KEY module-attribute

BODY_BYTES_KEY: str = '__file__'