Centive Python SDK

Production-quality Python SDK for integrating Centive into your backend services. This SDK provides both synchronous and asynchronous clients with built-in retry logic, comprehensive error handling, WebSocket support for real-time FE SDK communication, and automatic PII redaction for secure logging.

Features

• ✅ Sync & Async Support - Use CentiveClient for synchronous code or AsyncCentiveClient for async/await
• ✅ WebSocket Server - Real-time communication with frontend SDKs
• ✅ Message Accumulation - Capture and save Anam AI conversation messages automatically
• ✅ Page Telemetry - Receive FE SDK page-view events through an on_telemetry callback
• ✅ Automatic Retries - Exponential backoff with jitter for transient failures
• ✅ Type Safety - Full Pydantic validation and type hints throughout
• ✅ Error Handling - Comprehensive typed exceptions for all error scenarios
• ✅ Zero API Changes - Integrate without modifying your existing API responses
• ✅ Production Ready - Designed for high-reliability backend services

Installation

pip install centive-sdk

For development with all dev dependencies:

pip install centive-sdk[dev]

Quick Start

Basic Usage (Tool Mapping + Session Trigger)

from centive_sdk import AsyncCentiveClient, ToolMappingRequest, TriggerSessionRequest

async def main():
    async with AsyncCentiveClient(api_key="sk_live_your_api_key") as client:
        # At login: map user to tool
        mapping = await client.sessions.tool_mapping(
            ToolMappingRequest(
                user_id="user_123",
                user_name="John Doe",
                user_email="john.doe@acme.com",
                company_name="Acme Corp",
                role="admin"
            )
        )
        
        # Trigger session when needed
        session = await client.sessions.trigger_session(
            TriggerSessionRequest(
                user_id="user_123",
                user_trigger=True
            )
        )
        
        if session.session_data:
            print(f"Token: {session.session_data['token']}")

asyncio.run(main())

WebSocket Integration (Recommended)

initialize_websocket() returns a connection token bound to the user. Hand it to your frontend in the websocket URL; the Aria FE SDK connects with it and the socket can only act as that user.

Identity always comes from your authentication (session cookie, JWT, ...) — never from a query parameter or request body an anonymous caller controls.

from centive_sdk import AsyncCentiveClient, ToolMappingRequest

client = AsyncCentiveClient()  # api_key from CENTIVE_API_KEY env var

@app.post("/api/login")
async def login(user = Depends(get_current_user)):  # your existing auth
    await client.sessions.tool_mapping(
        ToolMappingRequest(user_id=user.id, user_name=user.name, user_email=user.email,
                           company_name=user.company, role=user.role)
    )
    token = await client.initialize_websocket(user_id=user.id)
    return {
        "status": "success",
        # FE passes this straight to the Aria FE SDK as websocketUrl
        "aria_websocket_url": f"wss://your-host/ws?token={token}" if token else None,
    }

The SDK automatically:

• Checks if Aria is paused internally
• If paused: no WebSocket server starts and no token is issued - the avatar stays hidden
• If active: starts the WebSocket server and returns the user's connection token
• When the FE connects with the token, the session is triggered and emitted to it

FastAPI Integration

Here's a complete example of integrating the SDK into a FastAPI application:

import os
from fastapi import Depends, FastAPI, HTTPException
from centive_sdk import (
    AsyncCentiveClient,
    ToolMappingRequest,
    AuthError,
    ValidationError,
    RateLimitError,
)

app = FastAPI()
centive = AsyncCentiveClient(
    api_key=os.getenv("CENTIVE_API_KEY"),
    ws_allowed_origins=["https://app.yourcompany.com"],
)
PUBLIC_WS_URL = "wss://app.yourcompany.com/ws"

@app.post("/api/login")
async def login(user = Depends(get_current_user)):
    try:
        await centive.sessions.tool_mapping(
            ToolMappingRequest(user_id=user.id, user_name=user.name, user_email=user.email,
                               company_name=user.company, role=user.role)
        )
    except AuthError:
        raise HTTPException(status_code=401, detail="Centive auth failed")
    except ValidationError as e:
        raise HTTPException(status_code=400, detail=str(e))

    token = await centive.initialize_websocket(user_id=user.id)
    return {
        "status": "success",
        "aria_websocket_url": f"{PUBLIC_WS_URL}?token={token}" if token else None,
    }

@app.on_event("shutdown")
async def shutdown():
    await centive.aclose()

See examples/fastapi_integration.py for the full pattern, including refreshing an expired connection token.

Configuration

Client Options

Example with Custom Configuration

import logging

logger = logging.getLogger("centive")
logger.setLevel(logging.INFO)

client = AsyncCentiveClient(
    api_key="sk_live_your_api_key",
    base_url="https://custom-centive-api.example.com",
    tool_mapping_path="/api/v1/tool-mapping",
    timeout_seconds=15.0,
    max_retries=5,
    ws_port=8080,
    logger=logger
)

Error Handling

The SDK provides typed exceptions for different error scenarios:

from centive_sdk import (
    CentiveError,      # Base exception
    AuthError,         # 401/403 - Invalid API key
    ValidationError,   # 400/422 - Invalid request data
    RateLimitError,    # 429 - Rate limit exceeded
    ServerError,       # 5xx - Server-side error
    NetworkError,      # Connection/timeout errors
)

try:
    mapping = await client.sessions.tool_mapping(request)
except AuthError as e:
    print(f"Authentication failed: {e}")
    print(f"Request ID: {e.request_id}")
except ValidationError as e:
    print(f"Invalid data: {e}")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after} seconds")
except (ServerError, NetworkError) as e:
    print(f"Service unavailable: {e}")
except CentiveError as e:
    print(f"Unexpected error: {e}")

Security

WebSocket identity binding (default: ws_auth_mode="token")

Every call to initialize_websocket(user_id) issues a cryptographically random connection token bound to that user. The FE must present it as ?token= on the websocket URL:

• A socket with a valid token is bound to that user for its lifetime. Frames claiming a different user_id are rejected (IDENTITY_MISMATCH), and frames targeting another user's session are rejected (SESSION_OWNERSHIP).
• Connections without a valid token are closed during the handshake (code 4401).
• Tokens expire after ws_token_ttl_seconds (default 1 hour) and stay valid within that window so the FE SDK's automatic reconnect keeps working. Fetch a fresh URL from your backend when a token expires (see examples/fastapi_integration.py).

What the SDK guarantees between users

Each of these is verified by the blackbox suite (blackbox/), which attacks the built wheel from a separate process:

• Sessions are namespaced per user. Session ids arrive from the client, so reusing another user's session id creates a separate session owned by the caller. One user cannot read, corrupt, or destroy another's transcript, and cannot squat a session id to lock someone out of recording their conversation.
• Identity is server-side. user_id in a frame is only accepted if it matches the connection's bound user; otherwise the frame is rejected.
• Upstream error detail never reaches a browser. Failures from the Centive API return a generic message on the websocket; full detail goes to your server log only. This includes the API's HTTP-200-with-status="error" responses, whose text can contain SQL, internal hostnames and organization ids.
• A failed save never destroys a transcript. If the API reports a rollback, messages stay buffered for retry instead of being cleared as if stored.
• Credentials stay out of logs. The API key, Anam session tokens, connection tokens (including in transport debug logging) and message content are never written to your logger.
• One user cannot exhaust the process, or affect anyone else's persistence. Sessions, retained bytes, message size and concurrent connections are bounded per user, and malformed payloads are rejected on arrival so they cannot trip the shared circuit breaker.
• Pause and lifecycle are per-user. Pausing or revoking one user closes only their connections; everyone else keeps their session.
• Malformed input cannot wedge the server. Invalid JSON, oversized frames, deeply-nested payloads and wrong-typed fields are handled without affecting other users' connections.

Revoking access mid-session

Connection tokens stay valid for their TTL, and a valid token can obtain a fresh avatar session. When a user's access ends (logout, deactivation, ban), cut it off explicitly — pause is handled automatically by initialize_websocket():

server = client.websocket_server
if server:
    await server.disconnect_user(user_id)   # revokes tokens + closes their sockets

Sending data to users yourself

Use send_to_user(), not broadcast(), for anything user-specific:

server = client.websocket_server
await server.send_to_user(user_id, {"type": "custom", "payload": ...})  # this user only
await server.broadcast({"type": "maintenance"})  # EVERY connected user

broadcast() reaches every end-user connected to the process; passing user-specific data to it is a cross-user disclosure.

Recommended hardening

• Serve the websocket over TLS (wss://) behind your reverse proxy, and set ws_host="127.0.0.1" so only the proxy can reach the SDK directly.
• Set ws_allowed_origins to your frontend origin(s) so browsers on other sites cannot even complete the handshake.
• Treat connection tokens like short-lived credentials: don't log them and keep them out of analytics that record URLs.

Legacy mode (ws_auth_mode="open") — deprecated

Pre-2.0 behavior: the first socket to connect is assigned the next registered user and client-supplied user_id is trusted. This is insecure — any client that can reach the port can race for another user's session token — and exists only as a temporary migration escape hatch. The server emits a warning at startup when it is enabled. ws_handshake_secret (a static shared ?token= secret) is honored only in this mode.

Logging

The SDK automatically redacts sensitive information in logs:

• API keys are completely redacted as [REDACTED]
• User names are masked (e.g., "Harrison" → "Ha***")
• Company names are masked (e.g., "Acme Corp" → "Ac***")

import logging

logger = logging.getLogger("my_app")
logger.setLevel(logging.INFO)

client = AsyncCentiveClient(
    api_key="sk_live_secret",
    logger=logger
)

# Logs will show masked values:
# INFO: Mapping user to tool {"user_id": "user_123", "user_name": "Jo***", "role": "admin"}

Pause Status Handling

The SDK automatically handles Aria pause status internally:

@app.post("/api/login")
async def login(user = Depends(get_current_user)):
    # SDK internally checks if Aria is paused
    # If paused: no WebSocket starts, no token issued, avatar stays hidden
    # If active: WebSocket starts and the user's connection token is returned
    token = await client.initialize_websocket(user_id=user.id)

    return {"aria_websocket_url": f"wss://your-host/ws?token={token}" if token else None}

Optional: Check Pause Status for UI Display

If you want to show pause status in your UI (e.g., "Aria is unavailable"), use the helper method:

@app.get("/api/aria/status")
async def aria_status(user = Depends(get_current_user)):
    pause_status = await client.get_aria_status(user.id)
    return {
        "is_paused": pause_status.is_paused,
        "pause_source": pause_status.pause_source,  # 'org_paused', 'account_paused', or 'not_paused'
        "message": pause_status.message,
        "paused_until": pause_status.paused_until,
    }

Optional: Check Last Pause Status

After calling initialize_websocket(), you can check the cached pause status:

await client.initialize_websocket(user_id=user_id)

# Check what happened (without another API call)
if client.last_pause_status and client.last_pause_status.is_paused:
    print(f"Aria is paused: {client.last_pause_status.message}")

WebSocket Flow

1. Backend: Call initialize_websocket(user_id) from an authenticated endpoint
2. SDK: Checks if Aria is paused for this user
3. If paused: No WebSocket server starts, no token is issued, avatar stays hidden
4. If active: WebSocket server starts and a per-user connection token is returned
5. Backend: Returns wss://your-host/ws?token=<token> to the frontend
6. FE SDK: Connects with that URL; the socket is bound to that user, the session is triggered, and the avatar renders

Message Accumulation (Anam AI Integration)

The SDK includes built-in support for accumulating conversation messages from Anam AI video avatar sessions. Messages are collected via WebSocket and automatically saved to the Centive API.

How It Works

1. frontend SDK sends message events to the BE SDK via WebSocket during the Anam AI session
2. BE SDK accumulates messages in memory with automatic deduplication
3. messages are saved via periodic auto-save AND on session end/disconnect

Backend Usage

The message accumulation happens automatically once the WebSocket server is initialized:

from centive_sdk import AsyncCentiveClient

client = AsyncCentiveClient(api_key="sk_live_your_api_key")

@app.post("/api/login")
async def login(user = Depends(get_current_user)):
    token = await client.initialize_websocket(user_id=user.id)
    return {"aria_websocket_url": f"wss://your-host/ws?token={token}" if token else None}

The SDK automatically:

• accumulates messages from incoming WebSocket events
• deduplicates messages using history as source of truth
• periodic auto-save every 5 minutes (configurable)
• saves on session end or early disconnect
• circuit breaker prevents cascading failures if API is down
• graceful shutdown saves all pending sessions when server stops

Features

• no limits: unlimited sessions, lifetime, and messages per session
• incremental save mode: send messages to API immediately as they arrive (real-time)
• batch save mode (default): accumulate and send all messages at session end
• periodic auto-save: saves long-running sessions automatically (default: every 5 min)
• circuit breaker: opens after 5 failures, recovers after 60s (configurable)
• automatic deduplication: backend handles duplicate messages automatically
• early disconnect handling: saves partial messages if connection drops
• graceful shutdown: saves all sessions when server stops
• retry logic: failed saves are retried with exponential backoff

Save Modes

The SDK supports two save modes for message persistence:

Batch Mode (Default)

Messages are accumulated in memory and sent all together when:

• Session ends (via session_end event)
• Periodic auto-save triggers (every 5 minutes)
• Early disconnect occurs
• Server shuts down

client = AsyncCentiveClient(
    api_key="sk_live_your_api_key",
    incremental_save_enabled=False  # default
)

Incremental Mode (Real-time)

Messages are sent to the API immediately as they arrive:

• After each message_history update
• After each finalized stream message (is_final=true)
• Backend handles deduplication automatically

client = AsyncCentiveClient(
    api_key="sk_live_your_api_key",
    incremental_save_enabled=True  # enable real-time saving
)

When to use incremental mode:

• Critical data that must be saved immediately
• Real-time analytics or monitoring requirements
• Lower risk tolerance for data loss
• Shorter, more frequent API calls preferred

When to use batch mode:

• Standard use cases where end-of-session save is sufficient
• Minimize API calls and bandwidth
• Periodic saves (every 5 min) are acceptable

For details on WebSocket event formats, see websocket_messages.md.

Page Tracking (Product Events)

From v2.2.0 the websocket server also accepts type: "telemetry" frames from the browser SDK. These carry page views and page leaves for the connected end user. The SDK validates each frame, queues the events, and forwards them to Centive in batches (every 2 seconds or 50 events, whichever comes first).

Nothing is required from your code beyond what you already do for the avatar: initialize_websocket() starts the queue and aclose() flushes it.

# Optional: emit your own server-side events into the same pipe.
client.telemetry.track(user_id, "invoice_paid", {"amount": 120})

# Optional: inspect the queue.
client.telemetry.stats()
# {'enqueued': 412, 'sent': 410, 'accepted': 398, 'rejected': 12,
#  'rejections_by_reason': {'user_not_mapped_to_account': 12}, 'queue_depth': 2, ...}

Context back to Aria: when a user has a live Aria session, Centive answers the events call with short context lines ("The user just opened the Invoices page."). The SDK pushes each one to that user's browser as a context frame and the browser SDK hands it to the avatar with anamClient.addContext. This is why page tracking exists: Aria knows where the user is while they talk.

What Centive keeps: only events for end users that tool_mapping linked to an account in your organization. Events for other users come back in the 202 response as rejected with reason user_not_mapped_to_account; they are counted in stats() and never retried. If that number is high, the company_name you pass to tool_mapping does not match the account names in Centive.

Safety properties:

• The browser cannot spoof identity: the frame's user_id must match the user bound to the socket, exactly like transcript frames.
• Frames over 100 events or 64 KB, and more than 120 frames per minute per connection, are dropped with a reasoned ack. The socket is never closed for a telemetry violation.
• The queue is bounded (telemetry_max_queue_events, default 5000). When full, the oldest events are dropped and counted. The websocket handler never blocks.
• Telemetry has its own circuit breaker; a failing events endpoint cannot stop transcript saves.
• Set telemetry_enabled=False to turn the feature off entirely.

Wire contract: docs/product-events/CONTRACTS.md in the CentiveAI repository.

Page Telemetry

The FE SDK (@centive/aria-sdk 1.1.0+) sends page-view and page-leave events over the same websocket as the chat traffic. The BE SDK validates them, attributes them to the connection's authenticated user, and hands them to a callback you supply. It stores nothing and forwards nothing to the Centive API — where the events go is entirely your decision.

Your callback receives two arguments: user_id, the identity bound to the websocket connection (never a value the browser claimed), and frame, the validated frame. The events are at frame["events"]; frame["schema"] and frame["sdk"] carry the producer's schema version and SDK version, which are worth storing alongside them.

async def record_page_events(user_id: str, frame: dict) -> None:
    await analytics.insert_many(
        {**event, "user_id": user_id, "sdk": frame.get("sdk")}
        for event in frame["events"]
    )

client = AsyncCentiveClient(
    api_key="sk_live_your_api_key",
    on_telemetry=record_page_events,  # sync callables work too
)

**event is spread first, never last: events are forwarded verbatim with no key filtering, so a client-supplied user_id inside an event would otherwise overwrite the identity the server already verified. Keep the trusted user_id and sdk after the spread.

Each event carries client_event_id, event_type, occurred_at, browser_session_id, detection, and usually a page object (path, pattern, title, referrer_path) plus duration_ms on page_leave. event_type is an open set — new client-side event kinds reach your callback rather than being rejected, so switch on the values you know and ignore the rest. Each of client_event_id, event_type, occurred_at and browser_session_id is capped at 256 characters; over that, the whole frame is rejected with INVALID_TELEMETRY_FORMAT rather than truncated, so keep custom event_type values well under that.

telemetry_ack carries one of five reason values whenever accepted is less than the event count:

Behavior worth knowing:

• Without on_telemetry, events are dropped. The frame is still acknowledged, with accepted: 0 and reason: "no_telemetry_handler", so the frontend never stalls waiting on a sink that does not exist.
• A callback that raises cannot break the connection. The frame is acked as dropped (reason: "handler_error") and the exception goes to your logger.
• A sync callback runs on the event loop. If yours does real work — a database write, an HTTP call — make it async, or hand off to a queue.
• The producer's own caps are enforced here too (100 events per frame, 64 KB per frame, 4 KB of properties per event). An over-cap frame never reaches your callback; it is acknowledged with accepted: 0 and a reason of too_many_events, frame_too_large or properties_too_large. A malformed frame gets the ordinary error response instead (INVALID_TELEMETRY_FORMAT).
• A newer schema is not an error. The SDK does not gate on the version, so a future FE SDK release keeps working without a backend deploy. Branch on frame["schema"] yourself if your storage ever needs to.
• properties._dropped is a counter the browser SDK reports when its own buffer had to evict events. It is relayed to your callback verbatim; the SDK never acts on it.
• properties._properties_dropped replaces the entire properties object with {"_properties_dropped": true} when the browser's own properties blob exceeded 4 KB before it was ever sent. Unlike _dropped, this is not a count alongside intact data — it means the original properties are gone entirely, not truncated.
• There is no rate limiting. The frontend understands a reason: "rate_limited" ack, but the SDK never sends one today.

Advanced Usage

Concurrent Tool Mapping (Async)

import asyncio

async def map_multiple_users(users):
    async with AsyncCentiveClient(api_key="sk_live_your_api_key") as client:
        tasks = [
            client.sessions.tool_mapping(ToolMappingRequest(**user))
            for user in users
        ]
        return await asyncio.gather(*tasks)

users = [
    {"user_id": "1", "user_name": "Alice", "user_email": "alice@acme.com", "company_name": "Acme Corp", "role": "admin"},
    {"user_id": "2", "user_name": "Bob", "user_email": "bob@techco.com", "company_name": "TechCo", "role": "viewer"},
]

mappings = asyncio.run(map_multiple_users(users))

WebSocket Server Status

client = AsyncCentiveClient(api_key="sk_live_your_api_key")
await client.initialize_websocket(user_id="user_123")

# Check server status
if client.websocket_server:
    print(f"Server running: {client.websocket_server.is_running}")
    print(f"Connected clients: {client.websocket_server.connected_clients}")

Development

Running Tests

# install dev dependencies
pip install -e ".[dev]"

# run all tests
pytest

# run with coverage
pytest --cov=centive_sdk --cov-report=html

# run specific test file
pytest tests/test_client_sync.py

# run async tests only
pytest tests/test_client_async.py

# run WebSocket tests
pytest tests/test_websocket_server.py

# run the blackbox security suite against the built wheel
python -m build
python -m venv /tmp/customer-venv
/tmp/customer-venv/bin/pip install dist/centive_sdk-*.whl
BLACKBOX_CUSTOMER_PYTHON=/tmp/customer-venv/bin/python pytest blackbox/ -v

See blackbox/README.md for what that suite covers and why it is structured the way it is.

Code Quality

# format code
ruff format .

# lint code
ruff check .

# fix auto-fixable issues
ruff check --fix .

Examples

See the examples/ directory for complete working examples:

• sync_example.py - Basic synchronous usage
• async_example.py - Asynchronous usage with WebSocket
• fastapi_integration.py - Full FastAPI integration
• websocket_example.py - WebSocket flow demonstration

Run examples:

python examples/sync_example.py
python examples/async_example.py
python examples/fastapi_integration.py
python examples/websocket_example.py

Environments (dev / prod)

There is one package — centive-sdk on PyPI. Which Centive environment it talks to is selected by configuration, never by installing a different package:

# production deployment
export CENTIVE_API_KEY="sk_live_..."
# CENTIVE_ENVIRONMENT unset -> prod

# dev deployment
export CENTIVE_API_KEY="sk_test_..."
export CENTIVE_ENVIRONMENT="dev"

client = AsyncCentiveClient()                    # env-var driven (recommended)
client = AsyncCentiveClient(environment="dev")   # or explicit
client = AsyncCentiveClient(base_url="https://...")  # full override, wins over both

Releasing SDK changes: dev branch → pre-release, prod branch → stable

The SDK follows the same branch model as the Centive FE/BE repos:

Both channels use the real index. pip excludes pre-releases unless they are exact-pinned, so pip install centive-sdk and range pins never resolve to a dev build — while dev consumers still get normal dependency resolution. TestPyPI is deliberately not used: installing from it safely requires --index-url plus --no-deps, which means the dev layer would not be running the SDK's real dependency set.

Flow for a change:

1. PR into dev — ci.yaml runs lint + tests (Python 3.10–3.13).

2. Merge to dev — tests and the blackbox suite run again, then a uniquely-versioned dev build (e.g. 2.1.0.dev42) is published automatically.

3. Test on dev — pin that exact version in the dev consumer with CENTIVE_ENVIRONMENT=dev, and run your integration suite against the dev API:

   pip install centive-sdk==2.1.0.dev42
   

4. Promote — set version in pyproject.toml to the release version, update CHANGELOG.md, and merge dev → prod. The prod workflow publishes, tags vX.Y.Z, and smoke-tests the published package. Pushes to prod without a version bump publish nothing (idempotent), so doc-only changes are safe.

5. Open the next cycle — bump version on dev to the next unreleased version, because X.Y.Z.devN sorts below X.Y.Z. The dev workflow fails with a clear message if you forget.

Production installs are exact-pinned (centive-sdk==2.0.0) from pypi.org with no extra index configured — never add --extra-index-url, which would expose every dependency in the build to substitution (security-audit finding A3). Keep dev and prod API keys separate; a dev deployment should hold no prod credentials.

Maintainers: RELEASING.md has the one-time trusted-publishing setup plus the full release procedure.

Migration to v2.0

v2.0 makes WebSocket identity binding secure by default:

1. initialize_websocket(user_id) now returns a connection token (a str, or None when Aria is paused).
2. Your API must hand the frontend its websocket URL including the token: wss://your-host/ws?token=<token>, and the FE app passes that URL to the Aria FE SDK as websocketUrl. No FE SDK upgrade is required.
3. Frames on a connection are bound to its user: forged user_id values and cross-user session access are rejected.
4. Derive user_id from your authentication, never from an unauthenticated query parameter.

If you cannot plumb the token through yet, ws_auth_mode="open" temporarily restores the old behavior (deprecated, insecure, warns at startup); plan to remove it before exposing the websocket to untrusted networks.

Migration from v1.0.0

If you're upgrading from v1.0.0, note these breaking changes:

• create_session() → Use tool_mapping() + trigger_session() instead
• CreateSessionRequest → Use ToolMappingRequest and TriggerSessionRequest
• SessionCreated → Use ToolMappingResponse and TriggerSessionResponse

See CHANGELOG.md for detailed migration guide.