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

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

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

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

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

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

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

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

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

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.

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

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

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.

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.

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

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

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.