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.

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

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

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.