Metadata-Version: 2.4
Name: opennvr-adapter-sdk
Version: 1.0.0
Summary: Boilerplate-free SDK for writing AI Adapter Contract v1 services for OpenNVR.
Author-email: OpenNVR <contact@open-nvr.org>
License: Apache-2.0
Project-URL: Homepage, https://github.com/open-nvr/ai-adapter
Project-URL: Documentation, https://github.com/open-nvr/ai-adapter/tree/main/opennvr_adapter_sdk
Project-URL: Repository, https://github.com/open-nvr/ai-adapter
Project-URL: Issues, https://github.com/open-nvr/ai-adapter/issues
Project-URL: Changelog, https://github.com/open-nvr/ai-adapter/blob/main/opennvr_adapter_sdk/CHANGELOG.md
Keywords: ai,inference,adapter,sdk,fastapi,websocket,computer-vision,nvr,object-detection,asr,tts
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Framework :: Pydantic :: 2
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Multimedia :: Video
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: fastapi<1.0,>=0.115
Requires-Dist: pydantic<3.0,>=2.7
Requires-Dist: python-multipart>=0.0.18
Provides-Extra: serve
Requires-Dist: uvicorn[standard]<1.0,>=0.30; extra == "serve"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: httpx>=0.27; extra == "dev"
Dynamic: license-file

# opennvr-adapter-sdk

The boilerplate-free way to write an [AI Adapter Contract v1](https://github.com/open-nvr/open-nvr/blob/main/docs/AI_ADAPTER_CONTRACT.md) service.

A minimal adapter is around thirty lines of FastAPI. The SDK gives you `AdapterService` — the ABC every adapter implements with four abstract methods — and `AdapterApp`, the builder that wraps your service in a FastAPI app with the six mandatory endpoints, auth, correlation-ID propagation, Prometheus metrics, and multipart plus JSON body parsing. `ServiceError` is the typed failure envelope from §7 of the contract. Every contract Pydantic type is re-exported from the package root, so adapter authors get one import line.

## Install

```bash
uv add opennvr-adapter-sdk
# with the uvicorn server bundled too:
uv add 'opennvr-adapter-sdk[serve]'
```

If you're not on [uv](https://docs.astral.sh/uv/) yet, `pip install opennvr-adapter-sdk` works the same — the package is a single wheel with FastAPI, Pydantic, and python-multipart as its only runtime dependencies. We recommend `uv` because adapter projects tend to grow heavy ML deps fast, and `uv sync` keeps resolve time and the lockfile manageable as they do.

## The minimum viable adapter

```python
# my_adapter/main.py
from datetime import datetime, timezone

from opennvr_adapter_sdk import (
    AdapterApp, AdapterService, BodyShape, ErrorCategory,
    HardwareEvaluationResponse, HardwareVerdict, InferResponse,
    ModelInfo, ServiceError,
)

class MyService(AdapterService):
    def __init__(self):
        self._ready = False

    def load(self):
        # Heavy lifting goes here.
        self._ready = True

    def is_ready(self): return self._ready

    def fingerprint(self):
        return "sha256:..."

    def model_info(self):
        return ModelInfo(
            name="my-model", version="1.0",
            framework="numpy", modalities_in=["text"],
            modalities_out=["text"], fingerprint=self.fingerprint(),
        )

    def hardware_evaluation(self):
        return HardwareEvaluationResponse(
            verdict=HardwareVerdict.OK, reasoning="ready",
            checked_at=datetime.now(timezone.utc), details={},
        )

    def infer(self, payload):
        if "text" not in payload:
            raise ServiceError(
                ErrorCategory.TRANSPORT_ERROR, code="malformed_input",
                message="'text' required", transient=False, http_status=400,
            )
        return InferResponse(
            model_name="my-model", model_version="1.0",
            inference_ms=1, result={"echoed": payload["text"]},
        )

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

Run it:

```bash
OPENNVR_ADAPTER_TOKEN=dev-token \
  uvicorn my_adapter.main:app --host 0.0.0.0 --port 9001
```

Verify conformance:

```bash
python -m conformance http://localhost:9001 --token dev-token
```

That's a complete contract-compliant adapter. The SDK handles `/health`, `/capabilities`, `/hardware/evaluation`, `/metrics`, `/infer`, `/infer/stream` (HTTP 501 refusal), auth, correlation_id, multipart + JSON body parsing, Prometheus metrics, lifespan startup. You write only the model wrapper.

## Body shapes

| Shape | Wire | Use for |
|---|---|---|
| `BodyShape.TEXT` | JSON + multipart (text-only fields) | TTS, LLM chat, any text-only adapter |
| `BodyShape.IMAGE` | multipart `frame` file + JSON `frame_b64` | Vision detection, classification, OCR |
| `BodyShape.AUDIO` | multipart `audio` file + JSON `audio_b64` | ASR, audio classification, TTS post-process |
| `BodyShape.GENERIC` | multipart `data` file + JSON `data_b64` | Anything else with binary input |

For non-TEXT shapes, the SDK puts the binary content at `payload[BODY_BYTES_KEY]` (bytes) and merges any `params` JSON into the dict. `BODY_BYTES_KEY` is re-exported from the SDK root — import it rather than hard-coding the literal so future renames don't silently break adapters. Caller-supplied params that shadow this key are rejected with `malformed_input` so collisions surface at the wire, not as silently-overwritten values.

## Streaming adapters

Add `supports_stream=True` to `AdapterApp(...)` and override `AdapterService.handle_stream(websocket)`:

```python
class MyDetector(AdapterService):
    async def handle_stream(self, websocket):
        await websocket.accept()
        # ... §6 protocol ...

app = AdapterApp(
    service=MyDetector(),
    ...
    supports_stream=True,
    stream_max_concurrent=16,
    stream_supports_shared_memory=False,
).fastapi_app
```

The SDK handles auth on the WebSocket upgrade (§6.5 close code 4001 on auth failure) and delegates to your handler. The §6 protocol itself — handshake → frame_meta + binary → result_message — is the adapter's responsibility (YOLOv8 has the reference implementation under `adapters/yolov8/`).

## Constructor reference

```python
AdapterApp(
    # Service — exactly one required:
    service: AdapterService | None,           # eager construction
    service_factory: Callable[[], AdapterService] | None,  # lazy (lifespan startup)

    # Adapter identity (required):
    name: str, version: str,
    vendor: str, license: str,
    tasks_advertised: Sequence[str],

    # Body shape + size cap:
    body_shape: BodyShape = BodyShape.TEXT,
    max_body_bytes: int = 32 * 1024 * 1024,

    # Capabilities metadata (optional, with defaults):
    permissions: Permissions = Permissions(),
    scheduling: Scheduling = Scheduling(),  # default max_inflight=1
    cost: Cost = Cost(),
    model_card_url: str | None = None,
    supported_contract_versions: Sequence[str] = ("1",),

    # Streaming (default off):
    supports_stream: bool = False,
    stream_max_concurrent: int = 0,
    stream_supports_shared_memory: bool = False,

    # Tuning:
    latency_buckets_seconds: tuple[float, ...] = (...),  # Prometheus buckets
    cors_origins: Sequence[str] = ("*",),
)
```

## Real-world examples

`opennvr-adapter-sdk` is the production runtime for the eight adapters shipped in this repo: `adapters/yolov8/` for object detection (`BodyShape.IMAGE` with WebSocket streaming); `adapters/piper/` for text-to-speech (`BodyShape.TEXT` with a custom `/voices` route and inline-audio response); `adapters/whisper/` for speech-to-text (`BodyShape.AUDIO`, multipart decode); `adapters/fast_plate_ocr/` for license-plate text recognition on a pre-cropped plate image (`BodyShape.IMAGE`, designed to chain downstream of YOLOv8); `adapters/insightface/` for face detection plus recognition with a REST face DB; `adapters/blip/` for scene captioning, used by the OpenNVR camera-agent; `adapters/vlm/` for open-vocabulary detection (OWL-ViT v2, detects free-text queries like "red truck"); and `adapters/bytetrack/` for stateful multi-object tracking as a post-processor over an upstream detector's results. Their `main.py` files are non-trivial reference implementations worth reading before authoring your own.

## Versioning

SDK ships with the same major version as the contract. SDK v1.x targets contract v1; a future contract v2 ships SDK v2.x. `AdapterApp.supported_contract_versions` defaults to `["1"]` — bump when you implement multi-version support.

## Why this isn't a "framework"

`AdapterService` is an ABC, not a metaclass. `AdapterApp` is a builder, not a base class. The SDK lives between you and FastAPI — your service code never imports FastAPI directly, but you can still drop down to `app.add_route(...)` for adapter-specific endpoints (see Piper's `/voices` route).

The contract is the source of truth. The SDK is a convenience layer on top of it. If the SDK gets in your way, write the service hand-rolled the way the early reference adapters did before the SDK was extracted — the contract is implementable without it, just more boilerplate.
