> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.amniscient.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.amniscient.com/_mcp/server.

# Amniscient Detection SDK for Python

#### [Python SDK](https://pypi.org/project/amniscient-detection/)

PyPI Repository

Python client for the Amniscient Inference API: low-latency object detection
behind a single call, `client.detect(image, model_id)`.

## Installation

```bash
pip install amniscient-detection
```

With optional dependencies for image compression:

```bash
pip install amniscient-detection[all]
```

## Quick Start

```python
from amniscient_detection import DetectionClient

# Initialize client. HTTP is the default transport.
client = DetectionClient(
    endpoint="https://api.amniscient.com/inference",
    api_key="your-api-key",
    organization_id="your-org-id",
)

# Detect objects in an image
result = client.detect("image.jpg", model_id="your-model-id")

# Process detections
for detection in result.detections:
    print(f"{detection.class_id} {detection.tags}: {detection.confidence:.2f}")
    print(f"  Location: ({detection.bbox.x1}, {detection.bbox.y1}) to ({detection.bbox.x2}, {detection.bbox.y2})")

# Close client when done
client.close()
```

> **Endpoint.** The production endpoint is `https://api.amniscient.com/inference`
> — plain HTTPS.

## How it works

The whole surface is `detect()`. You give it an image and a model; it returns a
`DetectionResult`. Everything in between is handled for you:

1. **Auth is validated first** — a missing or invalid API key raises
   `AuthenticationError` before anything else happens. It is never skipped.
2. **The image is prepared** — loaded, optionally compressed, and validated.
3. **The request is sent** and, on a transient failure, retried with backoff.
4. **The reply becomes a `DetectionResult`**, or a specific error is raised
   (see [Error Handling](#error-handling)).

How the request travels to the server is an internal detail with a sensible
default — see [Connecting](#connecting) if you need to change it.

For contributors, the pieces are small and single-purpose: `client.py`
orchestrates the flow and owns retries + error mapping; `request_builder.py`
builds and credential-checks the request; `response_parser.py` is the one place
a reply becomes a `DetectionResult`; `models.py` and `exceptions.py` hold the
result and error types.

## Features

* **One call, `detect()`**: connection, retries, and parsing are handled for you
* **Automatic Image Compression**: Resize and compress images to optimize payload size
* **Retry Logic**: Exponential backoff for transient errors
* **Validated auth**: a missing or invalid API key raises `AuthenticationError` — never a silent failure
* **Async Support**: `detect_async()` for asyncio compatibility
* **Raw Mode**: `detect(raw=True)` returns the unparsed server dict when you need it
* **Type Hints**: Full type annotations for IDE support

## Usage Examples

### Using Context Manager

```python
from amniscient_detection import DetectionClient

with DetectionClient(
    endpoint="https://api.amniscient.com/inference",
    api_key="your-api-key",
    organization_id="your-org-id",
) as client:
    result = client.detect("image.jpg", model_id="model-123")
    print(f"Found {len(result)} objects")
```

### Async Detection

```python
import asyncio
from amniscient_detection import DetectionClient

async def main():
    client = DetectionClient(
        endpoint="https://api.amniscient.com/inference",
        api_key="your-api-key",
        organization_id="your-org-id",
    )

    result = await client.detect_async("image.jpg", model_id="model-123")
    print(f"Found {len(result)} objects")

    client.close()

asyncio.run(main())
```

### Different Image Inputs

```python
import numpy as np
from PIL import Image

# From file path
result = client.detect("image.jpg", model_id="model-123")

# From bytes
with open("image.jpg", "rb") as f:
    result = client.detect(f.read(), model_id="model-123")

# From numpy array (requires opencv-python or Pillow)
array = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
result = client.detect(array, model_id="model-123")

# From PIL Image
pil_image = Image.open("image.jpg")
result = client.detect(pil_image, model_id="model-123")
```

### Filtering Results

```python
result = client.detect("image.jpg", model_id="model-123")

# Filter by confidence threshold
high_conf = result.filter_by_confidence(0.8)

# Filter by class id
people_only = result.filter_by_class_id("person", "pedestrian")
```

### Raw Responses

By default `detect()` parses the server reply into a `DetectionResult`. Pass
`raw=True` to skip parsing and get back the exact server dict instead — use
this when you need the unparsed response (e.g. for logging/debugging) or when
downstream code already depends on the original `xywh` contract and you don't
want to introduce the `DetectionResult` shape:

```python
payload = client.detect("image.jpg", model_id="model-123", raw=True)
# payload == {"inference_id": "...", "detection_found": True,
#             "xywh": [{"x": ..., "y": ..., "width": ..., "height": ...,
#                        "confidence": ..., "id": "...", "tag": {...}}, ...]}
for det in payload["xywh"]:
    print(det["id"], det["confidence"])
```

`raw` works the same way on `detect_async()`. Retries, error mapping, and auth
validation are unchanged — only the return value skips parsing.

### Barcode & QR reading

Pass `detect_barcodes=True` to decode barcodes/QR codes alongside object
detection, in the same request:

```python
from amniscient_detection import DetectionClient

client = DetectionClient(endpoint, api_key, organization_id, transport="http")
result = client.detect("shelf.jpg", model_id="<model>", detect_barcodes=True)
# object detection is unchanged; barcodes ride alongside:
for code in result.barcodes:
    print(code["type"], code["value"])   # e.g. qr https://…  |  upc-a 0012345678905
```

It's off by default — pass `detect_barcodes=True` explicitly (works the same
on `detect_async()`). `result.barcodes` is a list of `{"type", "value"}` dicts,
and the raw response (`raw=True`) has the same top-level `barcodes` array:

```json
{ "xywh": [ ... ], "detection_found": true, "inference_id": "...",
  "barcodes": [ { "type": "qr", "value": "..." } ] }
```

Supported symbologies (wire `type` values): `qr`, `upc-a`, `upc-e`, `code-39`.

Notes:

* `barcodes` is `[]` when `detect_barcodes` is off, no codes are found, or the
  decode step fails or times out — it never raises and never affects
  `result.detections` / `xywh`.
* Numeric 1D codes (UPC-A/UPC-E) decode to their expanded digit string (e.g.
  `"0012345678905"`), so don't expect an exact round-trip of a shorter printed
  value. `code-39` returns the encoded text as-is.

### Pinning a model version

Models are versioned. By default a request runs whichever version is currently
**active**, so results follow promotions automatically:

```python
# active version — follows promotions
result = client.detect("shelf.jpg", model_id="<model>")

# pinned — same weights and thresholds regardless of later promotions
result = client.detect("shelf.jpg", model_id="<model>", version=3)
```

Versions are 1-based. `version` works the same on `detect_async()` (sent as a
`version` form field).

Notes:

* Pin when you need reproducible output — a benchmark, a regression suite, or a
  customer held to a specific model revision. Leave it unset to get improvements
  as they're promoted.
* The version also selects that version's confidence/IoU thresholds and
  detection-item tags, not just its weights.
* A version that doesn't exist raises `ModelNotFoundError`.
* A malformed value (non-numeric, or less than 1) is rejected up front rather
  than silently falling back to the active version.

### Multi-tenant / per-request credentials

The client normally holds one credential for its lifetime. If you're a
**multi-tenant proxy** whose
credentials arrive with each request and are never stored server-side, build
**one shared client with no stored credentials** and pass `api_key` /
`organization_id` per `detect()` call — they ride each request, and nothing is
kept between calls:

```python
# one shared, long-lived client — no credentials stored
client = DetectionClient(endpoint, transport="http")

# each request carries its own tenant's credentials
result = client.detect(
    image, model_id="<model>",
    api_key=tenant_key, organization_id=tenant_org,
)
```

Per-call credentials ride each request and are never stored between calls — use
this for the per-request, never-store proxy pattern. **HTTP only:** a NATS
client is bound to one key at connect time.

### Validating credentials (login gate)

To accept an API key + org **before** a model is chosen (e.g. a web login
step), use `validate()` — it needs no model id and no image:

```python
if not client.validate(organization_id=org, api_key=key):
    raise Unauthorized("Invalid API key or organization")
```

`validate()` returns `True`/`False` — it calls `GET /models`, so an **accepted
key gets a 200** (a real positive signal) and a **rejected key (401/403) →
`False`**. Anything else (an unexposed route, a 5xx) **propagates** rather than
masquerading as valid, so you never get a false "looks valid" from a route that
simply isn't there — inspect `status_code` on the raised `DetectionError` if you
need to tell those apart. `get_default_model()` returns the org's default model
dict (or `None` when none is set), so you can pre-fill the model at a later
step. Both are HTTP only and authenticate through the API gateway.

### Listing models (model picker)

To let a user pick a model instead of pasting a UUID, `list_models()` returns
the org's selectable models — no model id or image needed:

```python
for m in client.list_models(organization_id=org, api_key=key):
    dropdown.add(value=m["id"], label=m["name"])   # gray out non-TRAINED
# each m == {"id": "<uuid>", "name": "<label>", "status": "TRAINED"}
```

It returns top-level models (specialist sub-models are excluded), each with
`id`, `name`, and `status`. HTTP only, authenticated through the gateway;
`organization_id` / `api_key` override the client's defaults.

### Putting it together — a typical flow

The methods above compose into one flow: **validate the key at login → list
models to pick one → detect on each capture.** How you hold the client depends
on your architecture.

**Single-tenant app** (one org, credentials known at startup) — one client for
its lifetime:

```python
client = DetectionClient(endpoint, api_key=key, organization_id=org, transport="http")

# 1. Login gate — reject bad credentials before showing the app
if not client.validate():
    raise Unauthorized("Invalid API key or organization")

# 2. Let the user choose a model (dropdown), instead of pasting a UUID
models = client.list_models()                 # [{"id","name","status"}, ...]
model_id = choose_from(models)                # e.g. first TRAINED model

# 3. Run detection on each capture
for frame in capture_stream():
    result = client.detect(frame, model_id=model_id)
    handle(result.detections)
```

**Multi-tenant proxy** (e.g. Iris — credentials arrive per request, nothing
stored) — **one shared client**, credentials passed per call:

```python
client = DetectionClient(endpoint, transport="http")   # no stored credentials

def handle_request(req):
    # validate this tenant's key at their login step...
    if not client.validate(organization_id=req.org, api_key=req.key):
        return http_401()
    # ...then serve detections, credentials riding each call
    return client.detect(
        req.image, model_id=req.model_id,
        organization_id=req.org, api_key=req.key,
    )
```

Distinguish a real credential failure from a gateway **routing** error at the
login gate — see [Error Handling](#error-handling) for the `401` vs `403`
pattern.

### Annotating detections (draw boxes)

Turn a `/detect` response + the source image into draw-ready boxes, or a
ready-to-display PNG. Pass the **raw** payload (`detect(raw=True)`), unchanged:

```python
payload = client.detect("shelf.jpg", model_id="<model>", raw=True)
result = client.annotate_detections("shelf.jpg", payload)   # label by name (default)

for b in result.boxes:            # draw-ready specs — render however you like
    print(b["label"], b["box"])   # box = {"x1","y1","x2","y2"} in pixels; also id/name/confidence/color

png = result.to_png()             # ready-to-display PNG bytes (needs the [draw] extra)
```

Options: `label="name"` (default) or `label="id"`; `color` as a hex string
(default `#00FF00`); `render=True` to render eagerly. Box thickness + font size
scale with the image resolution. Install the drawing dependency for `to_png()`:

```bash
pip install "amniscient-detection[draw]"
```

**Handle the no-detections case** — an empty payload raises `NoDetectionsError`
instead of returning a blank image:

```python
from amniscient_detection import NoDetectionsError

try:
    result = client.annotate_detections(image, payload)
except NoDetectionsError:
    ...  # nothing detected — show the original image / a "no detections" state
```

Name labels are pulled by id from the model's `detection-items` (the metadata
endpoint, a sibling of your inference endpoint), cached per model and falling
back to the id. The SDK derives that URL from an `https://…/inference` endpoint;
if yours differs, pass `metadata_endpoint=` when constructing the client — or
pass a `names={id: name}` map to `annotate_detections` to skip the network call.

### Legacy methods

`health_check()` and `load_model()` remain for backward compatibility but run
over the **deprecated gRPC transport** — they need a gRPC `host:port` endpoint
and are not part of the `detect()` path. New integrations should
not depend on them.

## Connecting

The SDK talks to the API over **HTTPS** by default, or over **NATS** (WebSocket)
when you ask for it. Over HTTPS you only supply the endpoint, key, and org (as
in every example above); no `transport=` needed:

```python
client = DetectionClient(
    endpoint="https://api.amniscient.com/inference",
    api_key="YOUR_KEY", organization_id="YOUR_ORG",
)
```

> gRPC was the original connection method; it is deprecated and no longer on
> the `detect()` path.

### Over NATS (WebSocket)

For lower latency and a persistent connection, use the NATS transport. It's
included in the standard install; pass `transport="nats"` and the NATS
endpoint. **Your API key is all you need**: the SDK resolves your organization
from the key, and the NATS server confines the connection to that organization.

```python
client = DetectionClient(
    endpoint="wss://nats-server.amniscient.com",
    api_key="YOUR_KEY",
    transport="nats",
)
result = client.detect("image.jpg", model_id="MODEL_ID")
```

Everything else is the same as HTTP: `detect()`, `detect_async()`,
`list_models()`, `validate()`, `get_default_model()`, barcodes, version pinning
and the same exceptions (an unknown model is `ModelNotFoundError`, a rejected
key is `AuthenticationError`).

Differences to know:

* **One tenant per client.** NATS binds the key to the connection, so per-call
  `api_key` / `organization_id` overrides raise `ValueError`. For a
  multi-tenant proxy, use HTTP (see *Multi-tenant / per-request credentials*).
* **`http_detect_path` doesn't apply** (HTTP only).

### HTTP detect path

Over HTTP the SDK posts to the customer detect route `/detect` with `model_id`
(or `inference_point_id`) in the form body. To hit a route that carries the
model in the URL path instead (e.g. the internal `/detectwithModel/{model_id}`),
override it with `http_detect_path` (only used when `transport="http"`):

```python
client = DetectionClient(
    endpoint="https://api.amniscient.com/inference",
    api_key="YOUR_KEY", organization_id="YOUR_ORG",
    transport="http",
    http_detect_path="/detect-with-model/{model_id}",
)
```

The path must contain a `{model_id}` placeholder; it's filled in per request.

## Configuration Options

| Parameter          | Default                                             | Description                                                                                                                 |
| ------------------ | --------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------- |
| `endpoint`         | -                                                   | `https://api.amniscient.com/inference` (HTTP) or `wss://nats-server.amniscient.com` (NATS)                                  |
| `api_key`          | -                                                   | Your application API key                                                                                                    |
| `organization_id`  | -                                                   | Organization identifier (optional over NATS: resolved from the key)                                                         |
| `transport`        | `"http"`                                            | `"http"` (HTTPS) or `"nats"` (WebSocket); see [Connecting](#connecting)                                                     |
| `compression`      | `True`                                              | Auto-compress images                                                                                                        |
| `high_compression` | `False`                                             | High compression mode (800x600, 250KB max)                                                                                  |
| `jpeg_quality`     | 70                                                  | JPEG compression quality (1-100)                                                                                            |
| `max_dimension`    | 3840                                                | Maximum image dimension in pixels (4K UHD long side)                                                                        |
| `timeout`          | 30.0                                                | Request timeout in seconds                                                                                                  |
| `max_retries`      | 3                                                   | Retries after the first attempt for transient failures (network, timeout, 5xx); `0` means one attempt                       |
| `max_workers`      | 4                                                   | Thread pool size backing `detect_async()` (one blocking `detect()` call runs per worker)                                    |
| `http_detect_path` | `None` (transport default `/detect`, model in form) | Override the HTTP detect path (e.g. `/detectwithModel/{model_id}` puts model in the URL); only used when `transport="http"` |
| `keep_alive`       | `True`                                              | Legacy gRPC path only (no effect on `detect()`)                                                                             |
| `use_tls`          | `True`                                              | Legacy gRPC path only (`detect()` uses the endpoint's scheme)                                                               |

## Error Handling

Every failure surfaces as a typed exception. Authentication is always
validated: a missing or invalid API key raises `AuthenticationError` — the
call never silently succeeds or hangs.

| Exception                   | Meaning                                                                                                  |
| --------------------------- | -------------------------------------------------------------------------------------------------------- |
| `AuthenticationError`       | Missing/invalid API key or organization ID                                                               |
| `EndpointNotAvailableError` | The route isn't deployed on this gateway/stage (a 403 `MISSING_AUTHENTICATION_TOKEN`) — the key is valid |
| `ModelNotFoundError`        | The requested model does not exist                                                                       |
| `ImageTooLargeError`        | Payload exceeds the server's size limit                                                                  |
| `RateLimitError`            | Too many requests                                                                                        |
| `ConnectionError`           | Transport unavailable after retries                                                                      |
| `ServerError`               | Server-side (5xx) failure                                                                                |
| `DetectionError`            | Base class for all of the above                                                                          |

Every exception carries `status_code` and `response_body` (the server's HTTP
status and parsed body) when the failure came from an HTTP response — both
`None` for client-side errors.

```python
from amniscient_detection import DetectionClient
from amniscient_detection.exceptions import (
    AuthenticationError,
    ModelNotFoundError,
    ImageTooLargeError,
    RateLimitError,
    ConnectionError,
    ServerError,
)

try:
    result = client.detect("image.jpg", model_id="model-123")
except AuthenticationError:
    print("Invalid API key or organization ID")
except ModelNotFoundError as e:
    print(f"Model not found: {e.model_id}")
except ImageTooLargeError as e:
    print(f"Image too large: {e.size_bytes} bytes (max: {e.max_bytes})")
except RateLimitError:
    print("Rate limited — back off and retry")
except ServerError as e:
    print(f"Server error ({e.status_code})")
except ConnectionError:
    print("Failed to connect to server")
```

**A missing route is not a rejected key.** An API Gateway returns `403
MISSING_AUTHENTICATION_TOKEN` for an **unmatched route** even when the key is
valid. The SDK recognizes that signature and raises `EndpointNotAvailableError`
(a distinct type — **not** `AuthenticationError`), so a route that simply isn't
deployed never masquerades as "bad credentials":

```python
from amniscient_detection import EndpointNotAvailableError, AuthenticationError

try:
    models = client.list_models(organization_id=org, api_key=key)   # or detect(...)
except EndpointNotAvailableError:
    # the key is fine — the route just isn't exposed on this gateway/stage
    raise ConfigError("Inference route not available for this environment")
except AuthenticationError:
    raise Unauthorized("Invalid API key or organization")   # genuine 401/403
```

`validate()` returns `True`/`False` for a genuine accept/reject, and **raises**
`EndpointNotAvailableError` when the route isn't deployed — it never answers a
confident `False` for a route that isn't there.

## Development

### Setup

Requires Python 3.10+ (use an explicit `python3.12` — macOS `python3` is often
3.9, and there's no bare `python`):

```bash
git clone https://github.com/amniscient/inference-api.git
cd inference-api/sdk/python
python3.12 -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
```

### Testing

**Unit tests** — fast, no network (the connection layer is mocked). Run these
while developing:

```bash
pytest tests/
```

**Live sandbox test** — proves a real `detect()` succeeds and a bad key is
actually rejected, end to end against the sandbox environment. One script sets
up credentials, tunnels, and env, then runs the integration suite:

```bash
bash scripts/run-sandbox-integration.sh
```

(Requires an AWS SSO refresh first; the script fetches a fresh cluster token, so
re-run it if you hit an auth error.)

Building, releasing to PyPI, and the full contributor workflow are documented
in [BUILD.md](BUILD.md).

## License

MIT License - see LICENSE file for details.