Python SDK

The official Python SDK for the RAG API. Built on httpx for reliable HTTP.

Installation

pip install rag-api

Quick Start

from rag_api import RAGClient

client = RAGClient(api_key="YOUR_API_KEY")

# Create a knowledge base
kb = client.create_knowledge_base(name="My Docs")

# Upload a document
doc = client.upload_document(kb["id"], "guide.pdf")

# Wait for processing
import time
while doc["status"] not in ("ready", "failed"):
    time.sleep(2)
    doc = client.get_document(kb["id"], doc["id"])

# Search
results = client.search(
    kb["id"],
    query="How do I reset my password?",
    top_k=5,
)

print(results["results"][0]["text"])

Configuration

client = RAGClient(
    api_key="YOUR_API_KEY",
    base_url="https://api.useragex.com/api",  # optional, this is the default
    timeout=30.0,                          # optional, seconds
)

Parameter	Type	Default	Description
api_key	str	(required)	Your API key
base_url	str	https://api.useragex.com/api	API base URL
timeout	float	30.0	Request timeout in seconds

The client can be used as a context manager for automatic cleanup:

with RAGClient(api_key="YOUR_API_KEY") as client:
    results = client.search(kb_id, query="test")
# Connection pool is closed automatically

Knowledge Bases

Create

kb = client.create_knowledge_base(
    name="Product Docs",
    description="All product documentation",  # optional
    metadata={"team": "engineering"},          # optional
)
# Returns: dict with id, name, description, metadata, document_count, ...

List

response = client.list_knowledge_bases(
    limit=20,     # optional, default 20
    cursor=None,  # optional, for pagination
)

kbs = response["data"]
meta = response["meta"]

# Paginate
if meta["has_more"]:
    next_page = client.list_knowledge_bases(cursor=meta["cursor"])

Get

kb = client.get_knowledge_base("kb_x1y2z3w4v5")

Update

updated = client.update_knowledge_base(
    "kb_x1y2z3w4v5",
    name="Updated Name",
    description="New description",
    metadata={"team": "platform"},
)

Delete

client.delete_knowledge_base("kb_x1y2z3w4v5")
# Cascades: deletes all documents and chunks

Documents

Upload File

# From a file path
doc = client.upload_document(
    kb["id"],
    "guide.pdf",
    metadata={"department": "engineering", "version": 2},
)
# Returns: dict with id, status ("pending"), ...

Upload Bytes

doc = client.upload_document_bytes(
    kb["id"],
    content=pdf_bytes,
    filename="report.pdf",
    content_type="application/pdf",
    metadata={"source": "api"},
)

Ingest Text

doc = client.upload_text_document(
    kb["id"],
    text="Your text content here...",
    name="notes.txt",                  # optional
    metadata={"source": "scraper"},    # optional
)

List Documents

response = client.list_documents(
    kb["id"],
    status="ready",  # optional: filter by status
    limit=20,
    cursor=None,
)

Get Document

doc = client.get_document(kb["id"], "doc_f6g7h8i9j0")
print(doc["status"])      # "pending" | "parsing" | ... | "ready" | "failed"
print(doc["chunk_count"]) # number of chunks after processing

Update Metadata

updated = client.update_document_metadata(
    kb["id"],
    doc["id"],
    metadata={"version": 3, "reviewed": True},
)

List Chunks

response = client.list_chunks(
    kb["id"],
    doc["id"],
    limit=10,
)

for chunk in response["data"]:
    print(f"[{chunk['chunk_index']}] {chunk['text'][:100]}...")
    print(f"  Tokens: {chunk['token_count']}, Page: {chunk['start_page']}")

Delete Document

client.delete_document(kb["id"], "doc_f6g7h8i9j0")
# Also deletes all chunks for this document

Search

results = client.search(
    kb["id"],
    query="How do I configure SSO?",
    top_k=5,                                               # optional, default 10
    rerank=True,                                            # optional, default True
    score_threshold=0.5,                                    # optional, default 0.0
    filter={"department": {"$eq": "engineering"}},          # optional
    include_metadata=True,                                  # optional, default True
)

for result in results["results"]:
    print(f"[{result['score']:.2f}] {result['document_name']}")
    print(f"  {result['text'][:200]}")

print(f"Reranked: {results['usage']['rerank_applied']}")

Usage

usage = client.get_usage()

print(f"Plan: {usage['plan']}")
print(f"Pages: {usage['usage']['pages_processed']} / {usage['limits']['pages_processed']}")
print(f"Queries: {usage['usage']['search_queries']} / {usage['limits']['search_queries']}")

Error Handling

from rag_api import RAGClient, RAGApiError

try:
    client.search(kb_id, query="test")
except RAGApiError as err:
    print(err.code)        # e.g. "NOT_FOUND"
    print(err.message)     # Human-readable message
    print(err.status_code) # HTTP status code
    print(err.details)     # Optional details dict

See Error Handling for all error codes and retry strategies.

Return Types

All methods return plain dictionaries. The SDK provides type aliases for documentation clarity:

from rag_api.types import (
    KnowledgeBase,   # TypedDict
    Document,        # TypedDict
    SearchResponse,  # TypedDict
    UsageResponse,   # TypedDict
    PaginatedResponse,  # TypedDict
)

Key Response Shapes

# SearchResponse
{
    "query": "...",
    "results": [
        {
            "chunk_id": "chk_...",
            "document_id": "doc_...",
            "document_name": "guide.pdf",
            "text": "...",
            "score": 0.94,
            "metadata": {
                "chunk_index": 3,
                "start_page": 5,
                "section_heading": "Auth Setup",
            },
            "document_metadata": {"department": "engineering"},
        }
    ],
    "usage": {
        "embedding_tokens": 12,
        "rerank_applied": True,
        "chunks_searched": 34,
    },
}