kurious 0.8.13

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

Kurious Python SDK

Official Python client for the Kurious Engine — AIntropy's retrieval platform combining hybrid RAG, NL2SQL, knowledge-graph search, video understanding, and agentic deep-think behind one typed, async-friendly API.

Why Kurious? One client to: spin up a project, ingest documents/video, and query it with citations — plus streaming, conversations, evals, and per-step pipeline timings. Tenant isolation is enforced per API key.

from kurious import kurious

client = kurious(api_key="krs_...", base_url="https://kurious-backend-trial-api.westus3.cloudapp.azure.com/api/v1")
result = client.search.intelligent(project_id, "Which NJ agencies report air-quality data?")
print(result.answer)   # routed across RAG · NL2SQL · KG, with citations in result.sources

Installation

pip install kurious

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

kurious>=0.8,<0.9

Step 1 — Sign up and get your API key (no password)

The first thing to do is get an API key. The SDK does this in two steps, using only your email — no password, no browser, no OAuth:

from kurious import kurious

client = kurious()
client.auth.start_signup("[e-mail]")          # emails a one-time code
act = client.auth.complete_signup("[e-mail]", "123456")

print(act.api_key)                                    # store it — shown ONCE
print(act.plan_type, act.requests_per_day_limit)      # your rate limits
print(act.max_size_gb, act.max_index, act.expiry_date)  # data sizes + expiry

complete_signup verifies the emailed code, creates your account, signs in, mints an API key, and hands back a SignupActivation with the key and the limits that apply to it. Save act.api_key to your secrets manager immediately (env var KURIOUS_API_KEY) — it cannot be retrieved later.

Or from the terminal — kurious init

Installing the SDK also installs the kurious CLI. kurious init runs the same password-free flow interactively and saves the key 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
✅ Account ready. API key saved to ~/.kurious/config.toml
   Plan: free_trial · 1000 req/day · up to 1.0 GB / 10 indexes

Then load the saved credentials from any script:

from kurious import kurious

client = kurious.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 where the server doesn't enforce OTP). For non-interactive runs set 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.

Getting a project-scoped API key

The signup above gives you an account-level key. For production, issue one project-scoped key per integration — a leaked key then only sees data inside that project. Three ways:

Option A — SDK (preferred). With any authenticated client:

key = client.api_keys.create(
    name="production-rag-service",
    access_type="read_write",
    project_id=client.projects.list()[0].id,
    expiry_days=90,
)
print(key.api_key)   # shown once — store it now

Option B — Web UI. Sign in to the Kurious 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 (env var KURIOUS_API_KEY).

Option C — 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 Kurious account).

export KURIOUS_USERNAME="[e-mail]"   # your work email
read -rsp "Kurious password: " KURIOUS_PASSWORD && export KURIOUS_PASSWORD
export KURIOUS_HOST_URL="https://kurious-backend-trial-api.westus3.cloudapp.azure.com"

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

# 2. Pick a project to scope the key to.
PROJECT_ID=$(curl -s "$KURIOUS_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 "$KURIOUS_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 KURIOUS_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 Kurious instance. The default (https://kurious-backend-trial-api.westus3.cloudapp.azure.com/api/v1) targets the Trial (evaluation) cluster. Override it to point at production, 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 kurious

client = kurious(
    api_key=os.environ["KURIOUS_API_KEY"],
    base_url=os.environ.get("KURIOUS_HOST_URL", "https://kurious-backend-trial-api.westus3.cloudapp.azure.com") + "/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 kurious

client = kurious(
    api_key=os.environ["KURIOUS_API_KEY"],
    base_url=f"{os.environ['KURIOUS_HOST_URL']}/api/v1",
)

Or authenticate with email + password

import os
from kurious import kurious

client = kurious(
    base_url=os.environ.get("KURIOUS_HOST_URL", "https://kurious-backend-trial-api.westus3.cloudapp.azure.com") + "/api/v1",
)
client.auth.login(
    username=os.environ["KURIOUS_USERNAME"],  # your work email
    password=os.environ["KURIOUS_PASSWORD"],
)

Context-manager usage

import os
from kurious import kurious

with kurious(api_key=os.environ["KURIOUS_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 = kurious(api_key=...)
print(client)                       # → kurious(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 KuriousError 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("Kurious 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",
)

Quota

client.quota is a friendlier front door over usage + entitlements: see your quota and request more from one place, without needing your company_id (it is resolved automatically).

client.auth.login("[e-mail]", "secret")

q = client.quota.get()                 # live quota + plan, no company_id arg
print(f"{q.requests_remaining_today} of {q.requests_per_day_limit} left")

# A quota increase is a plan-tier request reviewed by an admin.
client.quota.request_increase(
    plan_type="standard", notes="moving prod traffic",
)

Password reset

Two flows. The OTP flow completes entirely from the SDK; the link flow hands off to a browser.

# OTP flow (self-service, no browser)
client.auth.request_password_reset_otp("[e-mail]")     # emails a code
# user reads the code from their email
client.auth.reset_password("[e-mail]", "123456", "NewPass123!")
client.auth.login("[e-mail]", "NewPass123!")

# Link flow (Keycloak emails a reset link, completed in a browser)
client.auth.request_password_reset("[e-mail]")

Error handling

The SDK maps backend status codes to typed exceptions. Every exception inherits from KuriousError so a single except KuriousError: 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	KuriousError	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 (
    KuriousError, 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 KuriousError 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
client.quota	Unified view quota + request an increase (auto company_id)

Development

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

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

Changelog

Full history in CHANGELOG.md. Recent releases:

• 0.8.5 — from kurious import kurious lowercase import alias (the Kurious / AIntropy names stay as backward-compatible aliases — existing code keeps working); README, docstrings, and __repr__ updated to the new pattern.
• 0.8.4 — projects.ingest(wait=True) now blocks until the file is actually searchable (follows the chained index job, not just preprocess) and surfaces real documents_indexed / documents_failed counts on job.result["auto_chain"] (#800, #801).
• 0.7.0 — kurious init headless onboarding CLI + email-OTP verify-first signup (#709); installed as the kurious console script.

License

Proprietary — © 2026 AIntropy. All rights reserved. This software and its source code are proprietary and confidential; use is governed by your agreement with AIntropy. See LICENSE.