Amniscient Detection SDK for Python

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

Docs: User Guide (use & integrate) · Migration Guide (moving off direct HTTP) · Build & Release (contributors)

Installation

pip install amniscient-detection

With optional dependencies for image compression:

pip install amniscient-detection[all]

Quick Start

from amniscient_detection import DetectionClient

# Initialize client (connection persists across requests).
# NATS is the default transport.
client = DetectionClient(
    endpoint="wss://nats-server.<env>.amniscient.com",
    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()

New here? See docs/USER_GUIDE.md for a task-oriented walkthrough.

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).

How the request travels to the server is an internal detail with a sensible default — see 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

from amniscient_detection import DetectionClient

with DetectionClient(
    endpoint="wss://nats-server.<env>.amniscient.com",
    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

import asyncio
from amniscient_detection import DetectionClient

async def main():
    client = DetectionClient(
        endpoint="wss://nats-server.<env>.amniscient.com",
        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

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

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:

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:

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, raw=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()). The raw response gains a top-level barcodes array:

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

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

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.
• 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.

Pinning a model version

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

# 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() and over both transports (HTTP sends a version form field, NATS a version header).

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 over HTTP. Over NATS it currently surfaces as a generic DetectionError (status 500) because that transport does not yet map the server's 404.
• 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 (HTTP)

The client normally holds one credential for its lifetime (and, over NATS, a persistent connection bound to it). 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:

# 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 are HTTP only — the NATS transport binds credentials to the connection at connect time, so a NATS client is single-tenant (passing per-call credentials over NATS raises ValueError). Use HTTP for the per-request, never-store pattern; keep NATS for dedicated single-tenant integrations.

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:

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:

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:

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:

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 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:

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():

pip install "amniscient-detection[draw]"

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

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() (NATS/HTTP) path. New integrations should not depend on them.

Connecting

By default the SDK connects the fastest way available — you only supply the endpoint, key, and org (as in every example above). If you need to connect over plain HTTPS instead, flip one switch and use the HTTPS endpoint:

client = DetectionClient(
    endpoint="https://inference.<env>.amniscient.com",
    api_key="YOUR_KEY", organization_id="YOUR_ORG",
    transport="http",
)

detect(), results, and errors are identical either way.

gRPC was the original connection method. It is deprecated and no longer on the 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"):

client = DetectionClient(
    endpoint="https://gateway.<env>.amniscient.com",
    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	-	Endpoint URL (wss://… by default; https://… with transport="http")
api_key	-	API key for authentication
organization_id	-	Organization identifier
transport	"nats"	Connection method: "nats" (default) or "http"
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	Maximum retry attempts for transient failures
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
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 NATS or client-side errors.

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")

Telling a rejected key from a routing error (401 vs 403). An API Gateway returns 403 MISSING_AUTHENTICATION_TOKEN for an unmatched route even when the key is valid — so don't assume every AuthenticationError means "bad credentials." Inspect status_code / response_body on the calls that raise (detect(), list_models(), get_default_model()) before telling a user their key was rejected:

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

validate() collapses this to a plain True/False for the simple case — but because a 403 (including an unexposed route) reads as False, prefer the exception form above wherever a route may not be exposed on the target gateway/stage.

Development

Setup

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

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:

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 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.

License

MIT License - see LICENSE file for details.