quix-ai-sdk

Prompt in, analysis out. One method launches a Quix.AI agent session: a sandboxed sub-agent writes fresh analysis code for your prompt, queries the lakehouse, and streams conclusions back.

from quix_ai_sdk import run

result = run("Compare sector times between the two fastest laps of the last session")
print(result)              # the agent's conclusions
result.activities          # typed log: tools called, code generated, queries run
result.session_id          # resume handle + audit id

Install

uv add quix-ai-sdk        # or: pip install quix-ai-sdk

Configuration (env only)

Quix__Workspace__Id is only required for run()/stream() (the analysis sandbox) — the agents/knowledge/mcp admin surface doesn't need it.

QUIX_TOKEN must be a user PAT. Quix.AI sessions are user-owned — opening one requires a real user's PAT. The Quix__Sdk__Token that deployments auto-inject is a service token and cannot run Quix.AI sessions. Deployments that run analyses (e.g. automated post-race summaries) must therefore add QUIX_TOKEN as a secret. Missing vars raise MissingConfigError naming the variable.

Acting as a user

Web apps forwarding a logged-in user's token pass token= to run() or stream():

result = run("Summarize", token=request.user.quix_token)

The env QUIX_TOKEN is the default; token= is the only per-call override. When a real user triggers the run, pass their token — fall back to env QUIX_TOKEN only for headless/automated runs (attribution and permissions follow the token). Never log the token value or call args containing it.

Streaming

import asyncio
from quix_ai_sdk import stream, TextChunk, Done

async def main():
    async for event in stream("Find anomalies in brake temperature"):
        if isinstance(event, TextChunk):
            print(event.text, end="")
        elif isinstance(event, Done):
            print(f"\nsession: {event.session_id}")

asyncio.run(main())

Distilled events: TextChunk, ToolCalled, CodeGenerated, QueryRan, Progress, Failed, Done. Pass raw=True for wire-level SSE dicts.

Follow-ups

first = run("Summarize the session")
more = run("Expand on the tyre wear point", resume=first.session_id)

Provisioning (0.3): agents / knowledge / mcp

One-time setup of the AI resources your analyses run on. These live in submodules (agents, knowledge, mcp), not in the root namespace. Every write needs an org-admin user PAT (token= per call or env QUIX_TOKEN).

from quix_ai_sdk import agents, runs

agent_id = agents.ensure_analysis_agent(token=ORG_ADMIN_PAT)  # once per org
result = runs.run(f"Analyze test {test_id} ...", agent_id=agent_id)

ensure_analysis_agent() creates (or updates) an org agent with a curated, versioned system prompt shipped in-package: lakehouse query discipline, the delegate_task sandbox workflow, honest handling of missing data. Append app-specific rules with extra_prompt=. All ensure_* functions are idempotent and support dry_run=True previews. Full guide: docs/admin.md.

Failure model

• Agent outcome → in the result: result.status == "failed", detail in result.error. Never raises.
• Infrastructure → typed exceptions under QuixAIError: MissingConfigError, AuthError, RunTimeout, StreamInterrupted.
• A dropped stream cancels the run server-side (platform behavior). The session survives: StreamInterrupted.session_id feeds resume=. The SDK never auto-retries — a retried run duplicates cost and side effects.

Logging

The SDK logs to stdlib logging under the quix_ai_sdk logger and never configures handlers. Suggested app-side setup:

import logging

handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)-8s] [%(name)s] %(message)s"))
logging.getLogger("quix_ai_sdk").setLevel(logging.INFO)
logging.getLogger("quix_ai_sdk").addHandler(handler)

More runnable examples in examples/. Module docs in docs/.