RustyRAG Python SDK

Python client for the RustyRAG API — fast, minimal RAG-as-a-Service.

Install

pip install rustyrag

Note: the package installs as rustyrag, but the import is capitalized: from RustyRAG import rustyrag.

Authentication

Every request authenticates with your RustyRAG API key (rr_sk_...). Create one in your dashboard at rustyrag.ai — it's the only credential the SDK needs (email verification is just for creating your account; the SDK never uses it).

Pass the key directly:

from RustyRAG import rustyrag

client = rustyrag(api_key="rr_sk_...")

Or set it once in the environment and omit it in code (recommended — keeps secrets out of source):

export RUSTYRAG_API_KEY="rr_sk_..."

client = rustyrag()  # reads RUSTYRAG_API_KEY automatically

The key is sent as Authorization: Bearer rr_sk_... on every request. A missing key raises ValueError; an invalid one raises AuthenticationError (HTTP 401).

Quick start

from RustyRAG import rustyrag

client = rustyrag(api_key="rr_sk_...")

# Upload a document
resp = client.upload("report.pdf")
client.wait_for_upload(resp.task_id)

# Ask a question
answer = client.answer("What is the main conclusion?")
print(answer.message)
print(answer.sources)
print(answer.timing)  # Timing(ttft_ms=148, total_ms=230)

Streaming

for chunk in client.answer("Summarize the key points", stream=True):
    if chunk.text:
        print(chunk.text, end="", flush=True)
    if chunk.timing:
        print(f"\nTTFT: {chunk.timing.ttft_ms}ms")

All methods

client.health()                          # GET  /v1/health
client.usage()                           # GET  /v1/usage
client.models()                          # GET  /v1/llms
client.upload("file.pdf")                # POST /v1/upload
client.poll_upload(task_id)              # GET  /v1/uploads?task_id=...
client.wait_for_upload(task_id)          # Poll until complete
client.answer("question")               # POST /v1/answer
client.answer("question", stream=True)  # POST /v1/answer (SSE)
client.search("query", limit=5)         # POST /v1/search

Options

# Override model/provider per request
client.answer("question", model="llama3.1-8b", provider="groq")

# Skip reranker for faster TTFT (omit to use your account's Search preference)
client.answer("question", skip_reranker=True)

# Augment the answer with live web results (plan-gated)
answer = client.answer("question", web_search=True)
for w in answer.web_sources:
    print(w.title, w.url)

# Ground the answer/search in specific collections
client.answer("question", collection_ids=["<uuid1>", "<uuid2>"])
client.search("query", collection_ids=["<uuid1>"])

# Stream + skip reranker
client.answer("question", stream=True, skip_reranker=True)

Async

from RustyRAG import AsyncRustyRAG

async with AsyncRustyRAG(api_key="rr_sk_...") as client:
    answer = await client.answer("What is the main conclusion?")
    print(answer.message)

    # Async streaming
    async for chunk in await client.answer("Summarize", stream=True):
        if chunk.text:
            print(chunk.text, end="")

Configuration

client = rustyrag(
    api_key="rr_sk_...",                              # or set RUSTYRAG_API_KEY env var
    base_url="https://api.rustyrag.ai",               # or set RUSTYRAG_BASE_URL env var
    default_model="qwen-3-235b-a22b-instruct-2507",   # default
    default_provider="cerebras",                       # default
    timeout=120.0,                                     # seconds
)

Error handling

from RustyRAG import (
    rustyrag,
    AuthenticationError,
    BadRequestError,
    ForbiddenError,
    NotFoundError,
    ConflictError,
    QuotaExceededError,
    ServiceUnavailableError,
    NetworkError,
)

client = rustyrag(api_key="rr_sk_...")

try:
    answer = client.answer("question")
except AuthenticationError:
    print("Invalid API key")             # 401
except ForbiddenError:
    print("No access to that collection") # 403
except NotFoundError:
    print("Collection not found")        # 404
except QuotaExceededError:
    print("Weekly quota exceeded")       # 429
except BadRequestError as e:
    print(f"Bad request: {e.message}")   # 400 — e.message is the clean error string
except ServiceUnavailableError:
    print("Upstream service unavailable") # 502
except NetworkError:
    print("Network error (timeout, DNS, etc.)")

License

Elastic-2.0