Skip to main content

Python SDK

The official Python client. Sync and async clients (UniqOS / AsyncUniqOS), typed errors, automatic retries with backoff, automatic idempotency keys, SSE streaming as a plain iterator, fully type-hinted (py.typed), one runtime dependency (httpx), Python 3.10+.

The PyPI package is uniqos.

Install​

pip install uniqos==0.4.0

Configure​

from uniqos import UniqOS

client = UniqOS(
    api_key="uniq_test_...",
    base_url="https://api.uniqos.ai",  # host only; do NOT include /v1
    timeout=90,                         # seconds; default 90 (0 disables)
    max_retries=3,                      # 0 disables retries
    log_level="warning",               # debug | info | warning | error | silent
)

:::note Default timeout is 90s The default timeout is 90 seconds, deliberately above the server's turn ceiling (~75s), so a slow turn ends in the server's clean, unbilled 504 turn_timeout rather than the client abandoning a turn the server may still bill. Lowering it below the server budget takes on that risk. See 504 turn_timeout. :::

The SDK detects the environment from the key prefix (uniq_live_ / uniq_test_) and rejects a malformed key immediately. It never logs the full API key (only the prefix) nor message content beyond 50 characters.

Call​

result = client.respond(personality_id="pers_...", message="Hello!")
result["response"]  # the agent's reply (parsed JSON, snake_case, mirroring the API)
result["inferred_emotional_state"]

The namespaces mirror the API: client.personalities, client.catalog, client.vocabularies, client.end_users, client.relationships (+ .memory), client.interactions, client.me, client.organizations, client.api_keys, client.billing. For stateful turns, pass mode="stateful" and a user_id — see Stateless vs stateful.

Async​

from uniqos import AsyncUniqOS

async with AsyncUniqOS(api_key="uniq_test_...") as client:
    result = await client.respond(personality_id="pers_...", message="Hello!")

Both clients expose the identical surface; the async one just awaits.

Stream​

stream = client.respond(personality_id="pers_...", message="tell me a story", stream=True)
for event in stream:
    if event.type == "text":
        print(event.delta, end="", flush=True)
    elif event.type == "completion":
        print(f"\n[{event.latency_ms}ms]")

The five event types (metadata, text, guardrail_modulation, completion, error) are typed dataclasses with attribute access. The async client uses async for.

Errors​

from uniqos import RateLimitError, QuotaExhaustedError, TurnTimeoutError

try:
    result = client.respond(personality_id="pers_...", message="hi")
except RateLimitError as err:
    print(f"Rate limited. Retry after {err.retry_after_seconds}s")
except QuotaExhaustedError as err:
    print(f"Quota exhausted: {err.details}")
except TurnTimeoutError:
    print("Turn timed out (not billed); not auto-retried")

Every error carries code, message, request_id, http_status, and details. Catch UniqOSError for anything the SDK can raise. It retries transient failures (429 rate_limit_exceeded, 500 / 502 / 503, network) automatically with backoff; it never retries quota_exhausted, 401, 403, or 504 turn_timeout. See Errors.