kurious 0.8.4

Python SDK for the AIntropy Kurious Engine API — RAG, NL2SQL, knowledge-graph, and agentic search.

AIntropy Python SDK

Official Python client for the AIntropy Kurious Engine — a RAG platform with NL2SQL, knowledge-graph search, video understanding, and agentic deep-think.

Installation

pip install kurious

Requires Python ≥ 3.9. Pin a minor series in production:

kurious>=0.5,<0.6

Fastest start — kurious init

Installing the SDK also installs the kurious CLI. kurious init is a fully headless onboarding wizard — no browser required: it verifies your email with a one-time code, creates your account, logs in, mints an API key, and saves it to ~/.kurious/config.toml.

$ kurious init
Enter your email: [e-mail]
Verification code sent to [e-mail].
Enter the verification code (or 'r' to resend): 123456
Email verified.
Choose a password: ********
✅ Account ready. API key saved to ~/.kurious/config.toml

Then load the saved credentials from any script:

from kurious import AIntropy

client = AIntropy.from_config()   # reads ~/.kurious/config.toml
print(client.whoami())

Flags: --email, --full-name, --company-name, --base-url, --config (custom path), --force (replace an existing key), --yes (skip prompts), --skip-otp (only works where the server doesn't enforce OTP). For non-interactive runs set KURIOUS_PASSWORD + KURIOUS_OTP_CODE; override the config location with KURIOUS_CONFIG_PATH. The minted key is account-level and read_write — scope a narrower key per integration below.

Verifying email programmatically (without the CLI): r = client.auth.request_email_otp(email) then v = client.auth.verify_email_otp(email, code) and pass client.auth.signup(..., email_verification_token=v.verification_token).

Getting a project API key

We recommend issuing one project-scoped API key per integration — that way a leaked key only sees data inside that project. Two ways to get one:

Option A — Web UI (preferred). Sign in to the AIntropy dashboard, open Settings → API Keys → New key, pick a project, and copy the key value. It is shown once and cannot be retrieved later; store it in your secrets manager immediately (we recommend the env var name AINTROPY_API_KEY).

Option B — HTTP (CI, scripts, automation). Export your credentials as environment variables first — never paste them into the request body directly. The username is your work email (the address on your AIntropy account).

export AINTROPY_USERNAME="[e-mail]"   # your work email
read -rsp "AIntropy password: " AINTROPY_PASSWORD && export AINTROPY_PASSWORD
export AINTROPY_HOST_URL="https://api.aintropy.ai"

# 1. Exchange email + password for a JWT.
TOKEN=$(curl -s -X POST "$AINTROPY_HOST_URL/users/auth/login" \
  -H "Content-Type: application/json" \
  -d "{\"username\": \"$AINTROPY_USERNAME\", \"password\": \"$AINTROPY_PASSWORD\"}" \
  | jq -r '.access_token')

# 2. Pick a project to scope the key to.
PROJECT_ID=$(curl -s "$AINTROPY_HOST_URL/api/v1/projects" \
  -H "Authorization: Bearer $TOKEN" | jq -r '.[0].id')

# 3. Mint a key scoped to that project.
curl -s -X POST "$AINTROPY_HOST_URL/api/v1/api-keys" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d "{
    \"name\": \"production-rag-service\",
    \"access_type\": \"read_write\",
    \"project_id\": \"$PROJECT_ID\",
    \"expiry_days\": 90
  }" | jq -r '.api_key'

Save the returned value to your secrets manager as AINTROPY_API_KEY. The plain key is shown once and cannot be retrieved again.

The full step-by-step (account creation, listing keys, revoking keys) lives in the documentation.

Pick a host (host_url / base_url)

base_url is the SDK kwarg that points the client at your AIntropy instance. The default (https://api.aintropy.ai/api/v1) targets the hosted platform. Override it to point at a self-hosted deployment or a local development stack.

Hitting the bare host returns 404 — use /api/docs (Swagger UI) or /health to verify the host is reachable.

Quick start

Authenticate with an API key (dev — beta)

Always read credentials from the environment — never commit them.

import os
from kurious import AIntropy

client = AIntropy(
    api_key=os.environ["AINTROPY_API_KEY"],
    base_url=os.environ.get("AINTROPY_HOST_URL", "https://api.aintropy.ai") + "/api/v1",
)

summary = client.usage.summary()
print(f"{summary.requests_remaining_today} requests remaining today")

Or set the host via env var too so the same code runs unchanged across prod, dev, and local:

import os
from kurious import AIntropy

client = AIntropy(
    api_key=os.environ["AINTROPY_API_KEY"],
    base_url=f"{os.environ['AINTROPY_HOST_URL']}/api/v1",
)

Or authenticate with email + password

import os
from kurious import AIntropy

client = AIntropy(
    base_url=os.environ.get("AINTROPY_HOST_URL", "https://api.aintropy.ai") + "/api/v1",
)
client.auth.login(
    username=os.environ["AINTROPY_USERNAME"],  # your work email
    password=os.environ["AINTROPY_PASSWORD"],
)

Context-manager usage

import os

with AIntropy(api_key=os.environ["AINTROPY_API_KEY"]) as client:
    result = client.search.intelligent(project_id, query="What did Q3 revenue look like?")
    print(result.answer)
# HTTP pool is closed on __exit__

Discovering what's available

client = AIntropy(api_key=...)
print(client)                       # → AIntropy(base_url='...', auth=api_key, resources=17)
print(client.resources)             # list of every client.<resource> attribute
help(client.projects.ingest)        # full docstring for any method

All paginated list endpoints return a *Page wrapper that iterates natively (for x in page:), supports len() / indexing, and is falsy when empty — so you can write if not client.files.list(pid): ... without surprises.

Examples

First call — verify auth

Before doing anything else, confirm the SDK can reach the host with the credentials it's holding. Two lines, no project required:

me = client.whoami()
print(f"user_id={me.user_id} company_id={me.company_id} access={me.access_type}")

If that prints, you're authenticated. If you get AuthenticationError, the key is wrong / expired. If you get a connection error, the base_url doesn't match the host that issued the key.

Tenant isolation (one API key, one company)

client.company_id is auto-resolved from the API key at init via /whoami and cached on the transport. Every request the SDK makes carries X-Company-ID set to that value. A leaked key only ever sees data inside its own company / project — it cannot probe siblings even if the customer passes a different project_id (the engine refuses cross-company reads with 403).

If a customer-facing search call ever returns rows from another tenant, that is a backend bug — report it. The SDK has no way to override the resolved company.

Create a project, ingest a file, search (one-call ingest)

The recommended path is client.projects.ingest(...) — it handles upload + auto-routing + dispatch in a single call and returns a Job you can poll. The lower-level presign_upload → PUT → files.ingest → files.wait_for_job flow is deprecated; new code should not use it.

Video projects: before ingesting videos, set the project to kg_unstructured — client.projects.update_config(project.id, search_mode="kg_unstructured"). The default structured_unstructured runs an NL2SQL leg that a video project has no schema for; it falls back to a shared global index and can destabilize the search backend. See update_config for details.

# 1. Create (or reuse) a project
project = client.projects.create(name="Q3 Reports", description="Quarterly filings")

# 2. One-call: upload + auto-detect MIME + dispatch the right pipeline.
#    wait=True blocks until the worker chain finishes (preprocess →
#    ingest → kg.run). Pass on_progress to stream worker status:
def _on_tick(job):
    print(f"  {job.status} step={job.step} progress={job.progress_pct}")

job = client.projects.ingest(
    project.id,
    "q3-2025.pdf",
    wait=True,
    on_progress=_on_tick,
    timeout_s=30 * 60,        # PDFs usually finish in a few minutes
)
print("ingest:", job.status, "kind=", job.kind)
print("detected:", job.detected_domain, job.detected_sub_kind,
      f"confidence={job.detected_confidence:.2f}")

# 3. Search
result = client.search.intelligent(project.id, query="Summarise the quarterly results")
print(result.answer)
for src in result.attributed_sources:
    print(f"  - {src.title} ({src.relevance_score:.2f})")

The same call accepts a gs:// URI (skips upload), a local Path, or a file-like object (requires filename=). Pass config=AutoIngestConfig(...) to override per-domain knobs (e.g. disable the post-ingest KG chain).

Recovering a stuck or failed upload

If wait=True raises TimeoutError, or you want to inspect a half-finished ingest from a previous run, use the unified status + resume endpoints (keyed by file_id, not job_id):

status = client.files.get_ingest_status(file_id)
print(status.overall_status)          # pending_upload | uploaded | ingesting | indexed | failed
print(status.current_stage, status.last_error)
for stage in status.stages:
    print(f"  {stage.name}: {stage.status}")

if status.resumable:
    result = client.files.resume_ingest(file_id)
    print(f"re-dispatched: {result.dispatched_kinds}")

# Block until terminal (5-second polls, default 2-hour cap):
status = client.files.wait_for_indexed(file_id)
assert status.overall_status == "indexed"

Streaming search

for chunk in client.search.intelligent_stream_text(
    project.id, query="Walk me through the cash-flow narrative",
):
    print(chunk, end="", flush=True)

For full SSE events (with event / data separation), use client.search.intelligent_stream(...). Server-side event: error frames are translated into AIntropyError mid-stream, so any for chunk in ... loop either completes naturally or raises — no silent truncation.

Deep-think (agentic search)

mode="deep_think" switches the engine into the agentic loop — multi-step reasoning, tool calls, larger context. Slower (~10–60s typical), higher quality, costs more quota. Same return shape as the quick path:

result = client.search.intelligent(
    project.id,
    query="What changed in revenue mix between Q2 and Q3, and why?",
    mode="deep_think",
)
print(result.answer)
print(f"iterations={result.iterations} elapsed_ms={result.elapsed_ms}")
for tc in result.tool_calls_made:
    print(f"  tool={tc.get('name')} latency_ms={tc.get('latency_ms')}")

Conversations and bookmarks

chat = client.conversations.create(
    title="Q3 deep dive", mode="quick", project_id=project.id,
)
client.conversations.add_message(chat.id, content="What's the YoY growth?")
msgs = client.conversations.list_messages(chat.id)
last = msgs.messages[-1]

bm = client.bookmarks.create(
    message_id=last.id, conversation_id=chat.id, project_id=project.id,
)

# Make the conversation publicly readable
share = client.conversations.enable_share(chat.id)
print(f"Shared at: /conversations/shared/{share.slug}")

NL2SQL (natural-language → SQL)

res = client.search.nl2sql(
    project.id,
    query="Top 10 customers by revenue last quarter",
)
print(res.sql)
for row in res.rows:
    print(row)

Per-step timings (where is time going?)

# Aggregate per-step durations across all jobs in a project.
timings = client.projects.get_step_timings(project.id)
print(f"jobs={timings.job_count} total_ms={timings.total_duration_ms}")
for s in timings.steps:
    print(f"  {s.step:14s} count={s.count:4d} avg={s.avg_ms:>8.0f}ms p95={s.p95_ms:>8.0f}ms")

# Filter to one pipeline kind — e.g. just video.preprocess.
video = client.projects.get_step_timings(project.id, kind="video.preprocess")
slow = max(video.steps, key=lambda s: s.p95_ms)
print(f"slowest video stage by p95: {slow.step} @ {slow.p95_ms:.0f}ms")

Use it to drive a "where is time going" panel and spot regressions when the median for a step moves.

Backfill utterances (sliced fan-out)

# Fan out a sliced-scroll utterance backfill across 4 parallel children.
# Each child processes 1/N of the index in parallel via a server-side
# slice-scroll, running asynchronously on the engine's background workers.
fanout = client.video.backfill_utterances(
    project_id="proj_legal_video",
    max_slices=4,
)
print(f"umbrella={fanout.job_id} slices={fanout.slice_count}")

# Poll the umbrella row. `status.children` rolls up the per-slice counts
# (slice_count, completed, failed, running, pending) when present.
status = client.video.wait_for_job(fanout.job_id, poll_interval=10.0)
if status.children:
    print(
        f"done: {status.children.completed}/{status.children.slice_count} "
        f"failed={status.children.failed}"
    )

max_slices is capped at 16.

Knowledge-graph search

kg = client.search.kg(project.id, query="suppliers connected to Acme Corp")
print(kg.answer)

Evals

import json

questions_jsonl = "\n".join(json.dumps(q) for q in [
    {"question_text": "Q3 revenue?", "correct_answer": "$1.2B", "is_mcq": False},
    {"question_text": "Cash position?", "correct_answer": "$300M", "is_mcq": False},
])

eval_set = client.evals.create_set(
    project.id, name="Q3 smoke", questions_jsonl=questions_jsonl,
)
run = client.evals.trigger_run(project.id, eval_set_id=eval_set.id, search_mode="quick")
run = client.evals.wait_for_run(project.id, run.id)
print(f"accuracy={run.accuracy}  p95={run.latency_p95_ms}ms")

Direct LLM access (client.ai)

emb = client.ai.embedding("AIntropy is a retrieval platform.")
print(len(emb.embedding))

chat = client.ai.chat(
    messages=[{"role": "user", "content": "One-sentence pitch for RAG."}],
    max_tokens=64,
)

Search analytics + feedback

log = client.search_log.log(
    question="What was Q3 revenue?",
    kurious_answer="$1.2B (+12% YoY).",
    kurious_latency_ms=820,
    is_correct=True,
)
client.search_log.submit_feedback(log.id, kurious_rating=5)

Entitlements

ent = client.entitlements.get(company_id)
print(f"plan={ent.plan_type} status={ent.status}")

# Self-service trial (org-admin of the same company)
client.entitlements.start_trial(company_id)

# Upgrade request
client.entitlements.request_upgrade(
    company_id, plan_type="standard", notes="moving prod traffic",
)

Error handling

The SDK maps backend status codes to typed exceptions. Every exception inherits from AIntropyError so a single except AIntropyError: catches everything; catch the specific subclass when you want to react differently.

HTTP	Exception	When
401	AuthenticationError	Token expired, API key wrong/disabled, missing X-API-Key header
403	PermissionError	Caller authenticated but not a member / lacks role
404	NotFoundError	Project / job / file id doesn't exist (or is masked as 404 for security)
409	ConflictError	Duplicate (e.g. project name), state conflict (job already terminal)
422	ValidationError	Request body / params failed schema
429	RateLimitError	Quota or rate limit. .retry_after carries the server hint in seconds
5xx	ServerError	Backend transient. SDK already retries up to max_retries automatically
other	AIntropyError	Connection / timeout / parse failure. Base class

Plus two SDK-specific errors that are not HTTP mappings:

Exception	When
JobThrottled	Job.retry() hit the per-job 60s rate-limit. .retry_after_s carries the hint
(any)	Raised when a streaming event: error frame is received mid-stream

from kurious import (
    AIntropyError, AuthenticationError, PermissionError,
    NotFoundError, ConflictError, ValidationError,
    RateLimitError, ServerError, JobThrottled,
)

try:
    client.projects.get("does-not-exist")
except NotFoundError as e:
    print(f"missing: {e.status_code} {e}")
except RateLimitError as e:
    print(f"rate limited; retry after {e.retry_after}s")
except AIntropyError as e:
    # base class — catches every SDK error
    print(f"unexpected: {e}")

Logging

The SDK logs to the kurious logger (e.g. retry/back-off events). Enable it like any other Python logger:

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("kurious").setLevel(logging.DEBUG)

Resources reference

Resource	Purpose
client.auth	Signup, login, refresh, logout, password reset, email verify
client.projects	Project CRUD, config, members, audit log
client.files	Presigned uploads, listing, ingestion, jobs
client.indices	Create / delete / stats for company indices
client.documents	Per-doc CRUD + bulk ndjson
client.search	RAG, NL2SQL, KG, intelligent, raw ES, blended benchmark, streaming
client.evals	Eval sets, runs, results, wait-for-run
client.usage	Quota summary + daily breakdown
client.api_keys	List, create, delete API keys
client.nl2sql	NL2SQL data-pipeline orchestration
client.kg	Knowledge-graph build pipeline
client.video	Video pipeline + search
client.conversations	Chat threads, messages, public sharing
client.bookmarks	Saved messages
client.search_log	Log UI searches, submit feedback
client.ai	Direct embedding + chat completion
client.entitlements	Plan + quota self-service

Development

git clone https://github.com/aintropy-ai/kurious-python-sdk
cd kurious-python-sdk
pip install -e ".[dev,lint]"

pytest                    # unit tests
ruff check src tests      # lint
mypy src                  # type check

See CHANGELOG.md for release history.

License

MIT — see LICENSE.