Examples

Python (FastAPI)

Build a knowledge base manager with FastAPI and the Aether Python SDK.

This guide walks through a complete FastAPI application that exposes a REST API for document management, semantic search, and RAG-powered chat. The async Aether client fits naturally into FastAPI's async-first architecture, so every request is non-blocking end to end.

Quick start

Three commands to get the app running:

Bash
# 1. Install dependencies
pip install aether-ai fastapi uvicorn anthropic

# 2. Run the server
uvicorn kb_manager:app --reload --port 8000

The API is now live at http://localhost:8000. The chat endpoint requires an ANTHROPIC_API_KEY environment variable — every other feature works without it.

Project setup

Install the dependencies into a virtual environment:

Bash
python -m venv .venv
source .venv/bin/activate
pip install aether-ai fastapi uvicorn anthropic

The project is a single file:

text
kb-manager/
  kb_manager.py

API key required

Set AETHER_API_KEY in your environment (or a .env file) before starting the server — the app loads it and passes it to the client. See the Installation guide for details.

Server setup

Start with imports, the Aether client lifecycle, and the FastAPI app. The asynccontextmanager lifespan ensures the async client is properly initialized at startup and closed on shutdown.

Python
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel

from aether import AsyncAetherClient


aether: AsyncAetherClient


@asynccontextmanager
async def lifespan(app: FastAPI):
    global aether
    aether = AsyncAetherClient(api_key=os.environ["AETHER_API_KEY"])
    yield
    await aether.close()


app = FastAPI(title="Knowledge Base Manager", lifespan=lifespan)

The global aether client is shared across all request handlers. AsyncAetherClient uses connection pooling under the hood, so a single instance handles concurrency safely.

Document endpoints

These endpoints cover the full document lifecycle: listing, inserting, viewing, deleting, and restoring.

Python
class InsertRequest(BaseModel):
    text: str
    title: str


@app.get("/api/documents")
async def list_documents(offset: int = 0, limit: int = 50):
    docs = await aether.list(offset=offset, limit=limit)
    return {
        "documents": [
            {
                "id": doc.id,
                "filename": doc.filename,
                "created_at": doc.created_at,
                "status": doc.status,
            }
            for doc in docs
        ]
    }


@app.post("/api/documents", status_code=201)
async def insert_document(req: InsertRequest):
    filename = f"{req.title}.txt"
    doc = await aether.insert_text(text=req.text, filename=filename)
    return {"id": doc.id, "filename": doc.filename, "created_at": doc.created_at}


@app.get("/api/documents/{doc_id}")
async def get_document(doc_id: str):
    try:
        doc = await aether.get(doc_id)
    except Exception:
        raise HTTPException(status_code=404, detail="Document not found")
    content = await aether.download_text(doc_id)
    return {
        "id": doc.id,
        "filename": doc.filename,
        "created_at": doc.created_at,
        "status": doc.status,
        "content": content,
    }


@app.delete("/api/documents/{doc_id}", status_code=204)
async def delete_document(doc_id: str):
    try:
        await aether.delete(doc_id)
    except Exception:
        raise HTTPException(status_code=404, detail="Document not found")
    return None


@app.post("/api/documents/{doc_id}/restore")
async def restore_document(doc_id: str):
    try:
        await aether.restore(doc_id)
    except Exception:
        raise HTTPException(status_code=404, detail="Document not found")
    doc = await aether.get(doc_id)
    return {"id": doc.id, "filename": doc.filename, "status": doc.status}

A few things worth noting:

  • insert_document derives the filename from the user-supplied title, so documents are easy to identify when browsing.
  • get_document calls both get() for metadata and download_text() for the full content, returning everything the frontend needs in one round trip.
  • delete_document performs a soft delete. The document can be brought back with the restore endpoint.

Search and chat

The search endpoint wraps Aether's semantic search. The chat endpoint adds a RAG layer: it retrieves relevant passages with retrieve(), then sends them to Claude as grounding context.

Python
class ChatRequest(BaseModel):
    message: str


@app.get("/api/search")
async def search_documents(
    q: str = Query(..., description="Search query"),
    k: int = Query(default=10, ge=1, le=100),
):
    results = await aether.search(query=q, k=k)
    return {
        "results": [
            {
                "doc_id": r.doc_id,
                "score": r.score,
                "passage": r.passage,
            }
            for r in results
        ]
    }


@app.post("/api/chat")
async def chat(req: ChatRequest):
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        raise HTTPException(
            status_code=503,
            detail="Chat is unavailable. Set the ANTHROPIC_API_KEY environment variable to enable it.",
        )

    # Retrieve grounding context from Aether
    passages = await aether.retrieve(query=req.message, k=5)
    context = "\n\n".join(
        f"[Source {i + 1}] {p.content}" for i, p in enumerate(passages)
    )

    # Generate a grounded answer with Claude
    import anthropic

    claude = anthropic.AsyncAnthropic(api_key=api_key)
    response = await claude.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=(
            "You are a helpful assistant. Answer the user's question using "
            "only the following context. Cite sources by number.\n\n"
            f"{context}"
        ),
        messages=[{"role": "user", "content": req.message}],
    )

    return {
        "answer": response.content[0].text,
        "sources": [
            {"doc_id": p.doc_id, "content": p.content} for p in passages
        ],
    }

The anthropic import is inside the handler so the module loads even when the package is not installed. If the API key is missing, the endpoint returns a 503 with a clear message instead of crashing.

Swapping LLM providers

The RAG pattern here is provider-agnostic. Replace the Anthropic call with any LLM client and the rest of the code stays the same. See the Integrations section for examples with OpenAI, Azure, and others.

Stats endpoint

Expose a usage stat so monitoring tools or a dashboard can check the system at a glance. list() returns pagination metadata with every page, so a single call with limit=1 is enough to read the total document count.

Python
@app.get("/api/stats")
async def stats():
    page = await aether.list(limit=1)
    return {"total_documents": page.total}

That's the complete application. Run it with uvicorn kb_manager:app --reload --port 8000 and point any HTTP client at http://localhost:8000.

Next steps