Programmatic access & API keys

This page covers how to drive CRTX from code today, and — importantly — a concrete design for the API-support features that make CRTX usable inside other products and your customers' environments: long-lived API keys, SDKs, webhooks, rate limiting, and an embeddable chat widget.

Status legend

• ✅ Available today — works against the current backend.
• 🧭 Proposed — a design/roadmap sketch with an implementation path. Not yet in the codebase; included so you can prioritize and build it.

---

#✅ What works today

The full REST + SSE API (reference) is callable from any language using a Supabase-issued JWT (authentication). You can already:

• Automate ingestion (upload files / submit URLs, poll jobs).
• Run streaming queries and consume SSE.
• Manage collections, sessions, shares, and read eval metrics.
• Generate a typed client from Base URL/openapi.json.

A minimal automation loop:

import httpx, time
from supabase import create_client

sb = create_client(SUPABASE_URL, SUPABASE_ANON_KEY)
token = sb.auth.sign_in_with_password({"email": E, "password": P}).session.access_token
api = httpx.Client(base_url=BASE_URL, headers={"Authorization": f"Bearer {token}"})

col = api.post("/collections/", json={"name": "Nightly Import"}).json()
job = api.post(f"/ingest/url", json={"url": SRC, "collection_id": col["id"]}).json()

while True:                                   # poll until terminal
    s = api.get(f"/ingest/jobs/{job['job_id']}").json()
    if s["status"] in ("succeeded", "partial", "failed"):
        break
    time.sleep(2)

The gap: JWTs expire and represent a human. For headless integrations and customer-facing access you want API keys.

---

#🧭 API keys (the core of "API support")

Goal: issue long-lived, revocable, scoped keys that authenticate as a principal without a login flow — the standard way to let other systems and customers call your platform.

#Design

Key format. crtx_sk_<random> (secret, shown once) — plus a short public prefix (crtx_sk_a1b2…) stored in cleartext for identification in dashboards.

Storage. A new api_keys table:

Verification slots in beside the existing JWT dependency. The current get_current_user returns a dict keyed by sub; a key-aware dependency returns the same shape, so routers don't change:

# app/auth.py  (proposed)
async def get_principal(request: Request,
                        creds: HTTPAuthorizationCredentials = Security(_bearer)) -> dict:
    api_key = request.headers.get("X-API-Key")
    if api_key:
        row = lookup_api_key(sha256(api_key))          # constant-time compare
        if not row or row["revoked_at"] or expired(row):
            raise HTTPException(401, "Invalid API key")
        touch_last_used(row["id"])
        return {"sub": row["user_id"], "scopes": row["scopes"],
                "collection_scope": row["collection_id"], "auth": "api_key"}
    if creds:                                           # fall back to JWT
        return {**decode_token(creds.credentials), "scopes": ["*"], "auth": "jwt"}
    raise HTTPException(401, "Not authenticated")

Swap Depends(get_current_user) → Depends(get_principal) in the routers, and add a small require_scope("ingest") guard where mutations happen. Because access is already enforced from user["sub"], a key inherits exactly its owner's collections — optionally narrowed by collection_scope.

Management endpoints (JWT-authenticated, for the dashboard):

POST   /keys/            { name, scopes, collection_id?, expires_at? } → { id, key }  # key shown once
GET    /keys/            → [{ id, name, prefix, scopes, last_used_at, ... }]
DELETE /keys/{id}        # revoke

Usage (customer/integration side):

curl "$BASE_URL/query/" \
  -H "X-API-Key: crtx_sk_live_…" \
  -H "Content-Type: application/json" \
  -d '{ "question": "…", "collection_id": "…" }'

This unlocks: CI/cron ingestion, server-to-server queries from a customer's backend, and scoped read-only keys you can hand out safely.

---

#🧭 Rate limiting & usage metering

Once keys exist, meter them:

• Per-key rate limits via Redis token buckets (Redis is already a dependency). A dependency increments ratelimit:{key_id}:{window} and returns 429 with Retry-After when exceeded.
• Usage records written per request (key id, endpoint, tokens, latency) — you already log tokens per query, so extend query_logs/ingest_logs with the key id. This gives you per-customer billing and quota dashboards.
• Tiering: store limits on the key/plan and enforce in the same dependency.

---

#🧭 Webhooks (push, don't poll)

Ingestion is asynchronous, so today integrations poll /ingest/jobs/{id}. Webhooks let CRTX call you instead — the natural fit for pipelines.

Design.

• webhooks table: user_id, url, secret, events (ingest.succeeded, ingest.failed, ingest.partial, …), active.
• The ingest worker already computes terminal status in ingest_job.py — emit an event there. Sign the payload: X-CRTX-Signature: sha256=HMAC(secret, body).
• Deliver via the Arq worker with retry/backoff (Tenacity is already used).

// POST to the customer's URL
{
  "event": "ingest.succeeded",
  "job_id": "…", "collection_id": "…", "source": "report.pdf",
  "chunks_total": 512, "occurred_at": "2026-07-04T12:00:00Z"
}

Consumers verify the HMAC, then react (mark a document ready, trigger a re-index, notify a user). This is the listed roadmap item "webhook notifications on ingestion completion."

---

#🧭 Official SDKs

The API is small and regular — a thin SDK dramatically improves adoption. Two paths:

1. Generate from the OpenAPI spec (Base URL/openapi.json) with openapi-typescript/openapi-python-client for the REST surface.
2. Hand-write the streaming helper — the SSE parsing for /query/ is the one part worth wrapping ergonomically:

// @crtx/sdk (proposed) — ergonomic streaming
const crtx = new CRTX({ apiKey: "crtx_sk_…", baseUrl });
for await (const evt of crtx.query({ question, collectionId })) {
  if (evt.type === "metadata") renderSources(evt.sources);
  if (evt.type === "token") append(evt.token);
}

# crtx-python (proposed)
crtx = CRTX(api_key="crtx_sk_…", base_url=BASE_URL)
for evt in crtx.query(question="…", collection_id="…"):
    ...

Ship a Python SDK (for data/ingestion pipelines) and a TS/JS SDK (for app integration and the widget below).

---

#🧭 Embeddable chat widget

To let customers drop CRTX into their website/app without building UI:

• Ship a small script that mounts an iframe/web-component chat bound to one collection.
• Authenticate it with a scoped, query-only, collection-restricted API key (never a JWT, never an ingest key).
• For untrusted browsers, mint short-lived widget tokens from the customer's backend (their server exchanges its secret key for a 15-minute widget token), so the long-lived key never ships to the client.

<script src="https://cdn.your-app.com/crtx-widget.js"
        data-collection="COLLECTION_ID"
        data-token-endpoint="https://customer.com/api/crtx-widget-token"></script>

This turns CRTX into an answer engine embeddable in any support page, docs site, or internal tool.

---

#Priority order

If you build the above, this sequence delivers value fastest:

1. API keys — unblocks every other integration and all headless use.
2. Rate limiting + usage metering — required before exposing keys to customers.
3. Webhooks — removes polling; makes ingestion pipeline-friendly.
4. SDKs — lower the integration bar (start with the SSE streaming helper).
5. Embeddable widget — the fastest way for customers to adopt CRTX with no code.

For running CRTX itself inside a customer's infrastructure (single-tenant, VPC, on-prem, bring-your-own-keys, data residency), see Self-hosting & deploying in your environment.