uniqos 0.6.0

Official uniqOS SDK — Personality & Emotional Intelligence Engine + Relational Memory for AI agents.

uniqos

Official Python SDK for uniqOS — a Personality & Emotional Intelligence Engine + Relational Memory layer for AI agents.

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

Install

pip install uniqos

Quickstart

from uniqos import UniqOS

client = UniqOS(api_key="uniq_test_...")
response = client.respond(personality_id="pers_...", message="hola", mode="stateless")

response is the parsed JSON (snake_case, mirroring the API): response["response"] is the agent's reply.

We pass mode="stateless" explicitly because catalog personalities default to stateful, which keeps per-user memory and therefore requires a user_id. For a stateful conversation, pass mode="stateful" and a user_id; see the stateless vs stateful guide.

Get an API key from the uniqOS dashboard. Keys start with uniq_live_ (production) or uniq_test_ (sandbox); the SDK detects the environment from the prefix and rejects a malformed key immediately.

Async

from uniqos import AsyncUniqOS

async with AsyncUniqOS(api_key="uniq_test_...") as client:
    response = await client.respond(personality_id="pers_...", message="hola")

Both clients expose the identical surface (client.personalities, client.end_users, client.billing, …); the async one just awaits.

Streaming

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. Async uses async for.

Errors

from uniqos import RateLimitError, QuotaExhaustedError

try:
    response = 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}")

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

Configuration

client = UniqOS(
    api_key="uniq_test_...",
    base_url="https://api.uniqos.ai",  # host only; do NOT include /v1
    timeout=90,                         # seconds; default 90s; 0 disables
    max_retries=3,                      # 0 disables retries (504 is never auto-retried)
    log_level="warning",               # debug | info | warning | error | silent
    logger=my_logger,                   # optional: any object with debug/info/warning/error
)

Default timeout is 90s, deliberately above the server's ~75s turn ceiling, so a slow turn surfaces as the server's clean 504 turn_timeout (a TurnTimeoutError) instead of a client-side abort. A 504 turn_timeout is not retried automatically: the turn already ran to the budget, so a blind retry risks a second long turn and — combined with client-vs-server timing — a double bill (ADR-0009). Retrying it is your explicit choice.

The SDK never logs the full API key (only the prefix) nor message content beyond 50 chars.

Examples

Runnable scripts live in examples/: hello_world.py, stateful.py, streaming.py, error_handling.py, and dogfood.py.

With a real (or local) API — set the environment and run:

export UNIQOS_API_KEY="uniq_test_..."
export UNIQOS_PERSONALITY_ID="pers_..."
# optional, for a local backend:
export UNIQOS_BASE_URL="http://localhost:3000"

python examples/hello_world.py

Without a network — the test suite drives every code path against a mocked HTTP layer (respx), so no live API is needed to exercise the SDK:

import httpx, respx
from uniqos import UniqOS

@respx.mock
def test_it():
    respx.post("https://api.uniqos.ai/v1/respond").mock(
        return_value=httpx.Response(200, json={"response": "hi"})
    )
    with UniqOS(api_key="uniq_test_abcd1234") as client:
        assert client.respond(personality_id="p_1", message="hi")["response"] == "hi"

Documentation

Full reference and guides live at docs.uniqos.ai. This README and the in-editor docstrings cover the essentials; the deep documentation is not duplicated here.

License

MIT