Cursor for Docs - HydraDB SDK

Go from zero to a production AI assistant that answers “why was this built this way?” in four phases. Start with one file and a real recall query. End with a FastAPI backend that ingests GitHub, PRs, Slack, and RFCs, then generates GPT-4o answers grounded in your codebase.

How this guide is structured. Each phase ends with something that works. Phase 0 is a complete minimal system in under 10 minutes. Phases 1–3 are progressive upgrades. You never need to redo what came before.

All code in this cookbook is real. Base URL is https://api.hydradb.com. Get your API key at app.hydradb.com.

Prerequisites

Required knowledge: Python basics, REST APIs, environment variables
Required tools:

HydraDB API key
Python 3.11 or 3.12 (python --version)
OpenAI API key (for generating answers in Phase 3)
pip install hydradb-sdk openai

What You’ll Build

By the end of this cookbook, you’ll be able to:

Upload any codebase file into HydraDB and get grounded answers to “why was this built this way?”
Progressively ingest GitHub files, PRs, Slack threads, and RFCs across four phases
Build a FastAPI backend that answers developer questions using full_recall + GPT-4o
Add a VS Code extension trigger to query HydraDB from inside your editor

Phase 0 - Minimal Working System · 5–10 minutes

The only goal is to see a real recall response from a real file you uploaded. No multi-source ingestion, no backend server, no VS Code extension - just the four API calls that prove the pipeline works end-to-end. Goal: You will run one recall query and see actual results.

⚠️ Do Phase 0 first, even if you plan to skip ahead. Every later phase assumes the tenant exists and indexing works. Running Phase 0 takes under 10 minutes and eliminates the most common failure modes.

Prerequisites

You need three things before writing any code:

A HydraDB API key. Email [email protected] or book a demo at hydradb.com. You’ll receive a key that starts with hdb_.
Python 3.10+. Run python3 --version to check.
The HydraDB SDK. Run pip install hydradb-sdk python-dotenv.

mkdir cursor-for-docs && cd cursor-for-docs
echo "HYDRA_DB_API_KEY=hdb_your_key_here" > .env

Create config.py in the project root. This is the canonical configuration used by every script in this guide:

import os
from dotenv import load_dotenv
from hydra_db import HydraDB

load_dotenv()

# HydraDB
HYDRA_DB_API_KEY = os.environ["HYDRA_DB_API_KEY"]
TENANT_ID     = os.environ.get("HYDRADB_TENANT_ID", "engineering-docs")

client = HydraDB(token=HYDRA_DB_API_KEY)

# OpenAI
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
OPENAI_MODEL   = "gpt-4o"

# Recall defaults
RECALL_MAX_RESULTS = 15
RECALL_MIN_SCORE   = 0.5
RECALL_ALPHA       = 0.75

❌ Authorization header must be exact. Authorization: Bearer hdb_abc123 - capital B, one space, no quotes. Anything else returns 401.

Step 1 - Create a Tenant

What: Registers your top-level namespace. Tenant creation is asynchronous - infrastructure provisions in the background. Do not upload files until polling confirms it is ready. Endpoint: POST /tenants/create - Async, poll /tenants/infra/status before ingesting

import sys, os, time
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID

def create_tenant():
    client.tenant.create(tenant_id=TENANT_ID)
    print("Accepted")
    _poll_until_ready()

def _poll_until_ready(timeout=180, interval=4):
    print("Polling for readiness...")
    deadline = time.time() + timeout
    while time.time() < deadline:
        status = client.tenant.get_infra_status(tenant_id=TENANT_ID)
        infra = status.infra
        if infra.scheduler_status and infra.graph_status and all(infra.vectorstore_status):
            print("✓ Tenant ready."); return
        print(f"  scheduler={infra.scheduler_status} graph={infra.graph_status} vectorstore={infra.vectorstore_status} - retrying in {interval}s")
        time.sleep(interval)
    raise TimeoutError("Tenant not ready within 180s")

if __name__ == "__main__": create_tenant()

python3 phase0/create_tenant.py

Expected output:

Accepted: {'tenant_id': 'engineering-docs', 'status': 'accepted', 'message': 'Tenant accepted. Poll /tenants/infra/status for readiness.'}
Polling for readiness...
  scheduler=False graph=False vectorstore=[False, False] - retrying in 4s
✓ Tenant ready.

If it fails: 401 - API key wrong or missing Bearer prefix. Confirm .env has HYDRA_DB_API_KEY=hdb_abc123 with no surrounding quotes.

Step 2 - Upload One File

What: Uploads a single file using multipart form-data to /ingestion/upload_knowledge - the recommended beginner path. HydraDB handles chunking, embedding, and graph-node creation automatically. The returned source_id is what you use in Step 3 to verify indexing.

💡 Two ingestion modes exist. This step uses multipart file upload via /ingestion/upload_knowledge - the tested beginner path. An advanced app-source mode (used in Phases 1–2) supports structured IDs and explicit graph relations. Use file upload here first.

First, create a sample document:

mkdir -p phase0/sample_docs
cat > phase0/sample_docs/auth_middleware.md << 'EOF'
# Auth Middleware - Internal IP Exception

## Decision
Token validation is skipped for requests from internal IP ranges
(10.0.0.0/8 and 172.16.0.0/12).

## Rationale
Service-to-service calls within the VPC caused circular dependency issues
during startup. The security team approved this in RFC-007, on the condition
that internal network access is controlled at the VPC level.

## Approved by
Security team - Slack #eng-architecture - 2024-02-14
EOF

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID

def upload_file(filepath: str):
    filename = os.path.basename(filepath)

    with open(filepath, "rb") as f:
        result = client.upload.knowledge(
            tenant_id=TENANT_ID,
            files=[(filename, f, "text/markdown")],
        )

    print("RESULT:", result)
    return result

if __name__ == "__main__":
    result = upload_file("phase0/sample_docs/auth_middleware.md")
    # Note the source_id from the response - you need it for Step 3
    print("\nCopy your source_id for use in verify.py:", result)

Expected output:

STATUS CODE: 200
RESPONSE: {"results": [{"source_id": "YOUR_FILE_ID_HERE", "filename": "auth_middleware.md", "status": "accepted", "error": null}]}

Copy your source_id for use in verify.py: {'results': [{'source_id': 'YOUR_FILE_ID_HERE', 'filename': 'auth_middleware.md', 'status': 'accepted', 'error': None}]}

The exact value of source_id depends on HydraDB’s internal file registration. Copy the value returned and use it in Step 3. If it fails:

404 Tenant does not exist - Step 1 not complete, or tenant_id mismatch.
400 / missing files - Confirm the field name is files (not file), and that Content-Type is NOT manually set in headers.

Step 3 - Verify Indexing

What: Polls /ingestion/verify_processing using the source_id returned in Step 2. HydraDB returns a statuses array; you read statuses[0].indexing_status. Querying before this reaches "completed" returns empty results with no error - the most common beginner confusion. Replace YOUR_FILE_ID_HERE with the source_id from Step 2:

import sys, os, time
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID

# Replace with the source_id returned by upload_file.py in Step 2
FILE_ID = "YOUR_FILE_ID_HERE"

def verify_file(file_id: str, timeout: int = 120, interval: int = 3):
    """
    Polls verify_processing until indexing_status is "completed" or "errored".
    """
    print(f"Verifying '{file_id}'...")
    deadline = time.time() + timeout
    while time.time() < deadline:
        status = client.upload.verify_processing(
            tenant_id=TENANT_ID,
            file_ids=[file_id],
        )
        print("STATUS:", status)

        items = status.statuses or []
        if items:
            s = items[0].indexing_status
            if   s == "completed": print(f"✓ '{file_id}' ready."); return
            elif s == "errored":   print("Indexing errored."); return
            print(f"  indexing_status: {s} - waiting...")

        time.sleep(interval)
    raise TimeoutError("Indexing timed out")

if __name__ == "__main__":
    verify_file(FILE_ID)

Expected output:

Verifying 'YOUR_FILE_ID_HERE'...
STATUS CODE: 200
RESPONSE: {"statuses": [{"file_id": "YOUR_FILE_ID_HERE", "indexing_status": "processing"}]}
  indexing_status: processing - waiting...
STATUS CODE: 200
RESPONSE: {"statuses": [{"file_id": "YOUR_FILE_ID_HERE", "indexing_status": "completed"}]}
✓ 'YOUR_FILE_ID_HERE' ready.

Step 4 - Run Your First Recall Query

What: Sends a POST to /recall/full_recall and receives a chunks array. Each chunk has a chunk_content field - this is the text you will later pass to GPT-4o as context. This is the end-to-end proof that tenant, ingestion, indexing, and retrieval all work. Validated first recall - minimal working request:

{
  "tenant_id":   "engineering-docs",
  "query":       "internal IP auth skip logic",
  "max_results": 10
}

// sub_tenant_id is NOT required for this working flow.
// HydraDB handles scope internally when omitted.
// Response contains a "chunks" array - read chunk_content from each item.

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID

def recall(query: str, max_results: int = 10):
    """
    Minimal working recall.
    Only tenant_id + query + max_results are required.
    sub_tenant_id is NOT required - HydraDB handles scope internally.
    Returns SDK object with .chunks and .sources attributes.
    """
    data = client.recall.full_recall(
        tenant_id=TENANT_ID,
        query=query,
        max_results=max_results,
    )
    return data

if __name__ == "__main__":
    data   = recall("internal IP auth skip logic")
    chunks  = data.chunks or []     # iterate the "chunks" array
    sources = data.sources or []    # sources list for citations

    print(f"\nChunks returned: {len(chunks)}")
    for i, chunk in enumerate(chunks, 1):
        chunk_content = chunk.chunk_content or ""  # always use chunk_content
        score         = chunk.relevancy_score if chunk.relevancy_score is not None else "?"
        print(f"\n[{i}] relevancy_score={score}")
        print(f"     {chunk_content[:300]}")

    print(f"\nSources: {[s.title for s in sources]}")

Expected output:

STATUS CODE: 200

Chunks returned: 2

[1] relevancy_score=0.94
     Token validation is skipped for requests from internal IP ranges
     (10.0.0.0/8 and 172.16.0.0/12). Service-to-service calls within
     the VPC caused circular dependency issues during startup...

[2] relevancy_score=0.87
     The security team approved this exception in RFC-007, on the condition
     that internal network access is controlled at the VPC level.

Sources: ['auth_middleware.md']

If it fails: chunks: [] - Indexing is not yet complete. Wait 30 seconds and retry. Re-run Step 3 to confirm indexing_status: completed.

✅ Phase 0 complete. The same /recall/full_recall endpoint - with the same minimal three-field body - is what every later phase uses. You’ll only add parameters, not change the structure. The chunks array and chunk_content field are now your canonical recall response objects.

What you just built: You now have a working retrieval system. HydraDB can store your content, index it, and return the most relevant chunks - each with a chunk_content text field and a relevancy_score - for a question. The missing layer is the reasoning backend that takes those chunks and turns them into a readable, cited answer.

What works now	What comes next
Tenant creation	Better metadata and chunk quality
File upload	More source types for deeper context
Indexing verification	GPT-4o answer generation on top of recall
Recall returns chunks and sources	A backend and UI your team can use daily

Phase 1 - Improve Retrieval · 15–20 minutes

Ingest multiple files with structured app-source metadata and tune recall parameters. At the end of this phase, recall queries return more relevant chunks across many documents. Goal: 50+ files indexed with metadata, scoped recall working.

How Retrieval Works

Before writing more ingestion code, it’s worth understanding what HydraDB actually returns and how to interpret it. This mental model applies to every phase. Chunks vs. sources: When HydraDB indexes a document, it splits the content into overlapping chunks. Each chunk is embedded independently and stored as a node in the context graph. When you call /recall/full_recall, you get back a chunks array. Iterate this array and read chunk_content from each item - that is the text you pass directly to your LLM as context. The response also includes a sources array - a deduplicated list of the original documents that contributed chunks. Use sources for citation labels; use chunks for the actual LLM context. Anatomy of a recall chunk - field reference:

Field	Required	Description
`chunk_content`	required	The actual text of this chunk. This is the canonical field you must extract. Pass this directly to your LLM as context. It is the only field you cannot skip.
`relevancy_score`	optional	Higher is more relevant. Use it for ranking, filtering low-quality chunks (drop below 0.5), or deciding how many chunks to pass into the LLM context window.
`title`	optional	Human-readable label from the source document. Use for `[Source: ...]` citation references in your final answer.
`chunk_uuid`	optional	Unique identifier for this chunk. Useful for deduplication when fan-out recall across sub-tenants returns the same chunk more than once.
`url`	optional	The URL you set at ingestion time. Present only if supplied. Use for deep-links in citation UI.
`additional_metadata`	optional	The per-source metadata object from ingestion. May include `doc_type`, `repo`, `pr_number`, etc.

How to interpret the relevancy_score: Scores are relative within a single response - they indicate ranked relevance for your specific query, not absolute confidence.

relevancy_score range	What it means	What to do
`0.85+`	High confidence - chunk directly answers the query	Always include in LLM context
`0.65–0.85`	Good match - chunk is relevant, may not be the exact answer	Include, let LLM decide relevance
`0.40–0.65`	Weak match - tangentially related	Include only if few high-score results
`below 0.40`	Low match - probably not relevant	Drop from context to reduce noise

How graph context changes recall: When you add "graph_context": true and "search_apps": true to your recall request, HydraDB can use explicit relations[], provider IDs, parent links, threads, and actors around high-scoring chunks. A source file chunk can pull in the PR that last changed it. That PR can pull in the RFC or Slack thread it referenced. This multi-hop traversal is what makes “why” questions answerable.

💡 Always safe to add to Phase 0. "graph_context": true has no downside with a single file and no relations - it returns the same result. Turn it on now.

P1 · Step 1 - Batch Upload with Explicit IDs (App-Source Ingestion)

What: Switches to multipart app-source ingestion via /ingestion/upload_knowledge with app_knowledge as a JSON-stringified form field. This gives you full control over stable IDs, provider namespaces, structured fields, metadata, and (in Phase 2) explicit relations[]. The {repo_name}/{relative_path} external ID convention is required for Phase 2 linking.

⚠️ Advanced path - stricter validation. App-source ingestion requires well-formed payloads with id, tenant_id, sub_tenant_id, kind, provider, external_id, and fields on every item. Malformed requests are rejected outright. Use the file upload path (Phase 0 Step 2) if you just want to get content indexed quickly. Use app-source ingestion when you need stable IDs and graph relations.

import sys, os, json, time, subprocess, pathlib
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID
from phase0.verify import verify_file

TEXT_EXTS = {".py",".ts",".js",".go",".rs",".java",".md",".yaml",".toml",".txt"}
SKIP_DIRS = {".git","node_modules","dist","build","__pycache__",".venv"}

def git_timestamp(repo_path: str, rel_path: str) -> str:
    try:
        raw = subprocess.check_output(
            ["git","log","-1","--format=%cI",rel_path],
            cwd=repo_path, stderr=subprocess.DEVNULL).decode().strip()
        return raw or "2020-01-01T00:00:00Z"
    except: return "2020-01-01T00:00:00Z"

def upload_batch(batch: list) -> list:
    result = client.upload.knowledge(
        tenant_id=TENANT_ID,
        app_knowledge=json.dumps(batch),
    )
    ids = [r.source_id for r in (result.results or []) if r.source_id]
    print(f"  Uploaded {len(ids)} items")
    return ids

def verify_batch(ids: list):
    for fid in ids: verify_file(fid)

def ingest_directory(repo_path: str, repo_name: str) -> list:
    batch, all_ids = [], []
    root = pathlib.Path(repo_path).resolve()
    for f in root.rglob("*"):
        if f.is_dir() or any(p in f.parts for p in SKIP_DIRS): continue
        if f.suffix not in TEXT_EXTS or f.stat().st_size > 500_000: continue
        rel = str(f.relative_to(root))
        try: content = f.read_text(encoding="utf-8", errors="ignore")
        except: continue
        batch.append({
            "id":          f"github_{repo_name}_{rel.replace('/', '_')}",
            "tenant_id":   TENANT_ID,
            "sub_tenant_id": "code",
            "kind":        "custom",
            "provider":    "github",
            "external_id": f"{repo_name}/{rel}",   # Phase 2 relations reference this external_id
            "fields": {
                "kind": "custom",
                "data": {
                    "record_type": "source_file",
                    "path": rel,
                    "body": content,
                    "language": f.suffix.lstrip("."),
                    "updated_at": git_timestamp(str(root), rel),
                },
            },
            "metadata": {"container_type":"repository","container_id":repo_name,
                      "source":"github","doc_type":"source_file","repo":repo_name,
                      "language":f.suffix.lstrip(".")},
        })
        if len(batch) == 20:
            all_ids += upload_batch(batch); batch = []; time.sleep(1)
    if batch: all_ids += upload_batch(batch)
    print(f"Verifying {len(all_ids)} files...")
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} files indexed from '{repo_name}'")
    return all_ids

if __name__ == "__main__":
    ingest_directory("/path/to/your/repo", "myrepo")

Expected output:

  Uploaded 20 items
  Uploaded 20 items
  Uploaded 14 items
Verifying 54 files...
✓ 54 files indexed from 'myrepo'

P1 · Step 2 - Metadata for Scoping and Citations

What: App-source identity belongs in provider and external_id; searchable structure belongs in fields; stable scoping values belong in metadata. Use metadata for repo, channel, space, project, doc type, or other filters you want available at recall time.

{
  "provider": "github",
  "external_id": "myrepo/auth/middleware.py",
  "fields": {
    "kind": "custom",
    "data": {
      "record_type": "source_file",
      "path": "auth/middleware.py",
      "body": "full content here",
      "language": "py"
    }
  },
  "metadata": {
    "container_type": "repository",
    "container_id": "myrepo",
    "source": "github",
    "doc_type": "source_file",
    "repo": "myrepo",
    "language": "py"
  }
}

P1 · Step 3 - Tuned Recall with Graph Context

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID

def tuned_recall(query: str, scope: str = None, max_results: int = 15):
    """scope: optional sub_tenant_id. Omit to search all sub-tenants automatically."""
    kwargs = {
        "tenant_id":     TENANT_ID,
        "query":         query,
        "max_results":   max_results,
        "mode":          "thinking",
        "graph_context": True,
        "search_apps":   True,
        "alpha":         0.75,
        "recency_bias":  0.2,
    }
    if scope: kwargs["sub_tenant_id"] = scope
    return client.recall.full_recall(**kwargs)

def build_context(chunks: list, min_score: float = 0.5) -> tuple[str, list]:
    """
    Extract chunk_content from chunks and build LLM context string.
    Filters by relevancy_score; deduplicates by chunk_uuid.
    Returns: (context_text, sources_list)
    """
    parts, sources, seen = [], [], set()
    for chunk in chunks:
        relevancy_score = chunk.relevancy_score if chunk.relevancy_score is not None else 1
        if relevancy_score < min_score: continue
        chunk_content = chunk.chunk_content or ""  # canonical field
        uid           = chunk.chunk_uuid or chunk_content[:40]
        if not chunk_content or uid in seen: continue
        seen.add(uid)
        addl = chunk.additional_metadata
        doc_type = (addl or {}).get("doc_type", "doc")
        title    = chunk.source_title or "untitled"
        parts.append(f"[{doc_type.upper()}] {title}\n{chunk_content}")
        sources.append({"title": title, "url": chunk.url,
                         "relevancy_score": chunk.relevancy_score,
                         "chunk_uuid": chunk.chunk_uuid})
    return "\n\n---\n\n".join(parts), sources

Phase 2 - Multi-source Context · 20–30 minutes

Connect GitHub source files to their pull requests, Slack decision threads, and internal wikis. Use explicit relations[] with provider-scoped external_id targets to guarantee graph edges. When a developer asks “why?”, the answer now travels across all four source types in a single query. Goal: A “why” question returns code + PR + Slack + RFC in one response. Pipeline: Source files → Pull requests → Slack threads → Wikis & RFCs → Graph built

💡 Explicit vs. automatic graph edges. Setting explicit relations[] guarantees a graph link every time. HydraDB may also try to extract relationships from content, but this is best-effort. Always use explicit relations for links that matter.

Why Each Source Type Matters

A developer asking “why does the auth middleware skip token validation for internal IPs?” cannot be answered by any single source. Here is what each source type contributes:

Source	Answers	Role	Without it
Source files	What the code does	Anchors the query. Entry node for graph traversal.	No traversal starting point
Pull requests	Why the change happened	Intent and debate behind the change. Review comments capture rejected alternatives. Set `relations[]` to the changed file external IDs.	Code exists but has no rationale
Slack threads	Discussion and trade-offs	Captures informal approvals and decisions never written up in docs.	Informal approvals are invisible
Wikis & RFCs	Formal decision record	Authoritative rationale and approval chain. An RFC directly answers a whole class of “why” questions.	Missing the authoritative “because”

A normal code search tool answers what a function does. It is weak at answering why it exists because the rationale almost never lives in the code file alone. HydraDB becomes powerful when you index all the places where engineering intent is recorded:

Code tells you what is running now.
Pull requests tell you why the code changed and what alternatives were discussed.
Slack threads capture informal trade-offs, approvals, and urgency.
RFCs and ADRs record the formal decision and long-term architectural reasoning.

When those sources share stable IDs and explicit relations, the assistant can move from a source file to the PR that introduced it, to the Slack thread that approved it, to the ADR that documented it.

💡 Adaptation checklist. Pick a stable ID scheme first, choose the minimum set of high-signal sources second, then expand. Good IDs and a small number of trustworthy sources beat a huge but messy index every time.

P2 · Step 1 - Ingest GitHub Source Files with Consistent IDs

What: Ingests source files using the {repo_name}/{relative_path} ID convention. These external IDs are what PR ingestion will reference in relations[].target.external_id. The IDs must match exactly - myrepo/auth/middleware.py, not auth/middleware.py. Use ingest_directory() from Phase 1 Step 1. If you already ran Phase 1, your files are already indexed.

# Only run if you skipped Phase 1
python3 -c "
from phase1.batch_upload import ingest_directory
ingest_directory('/path/to/your/repo', 'myrepo')
"

P2 · Step 2 - Ingest Pull Requests with Explicit Graph Relations

What: Fetches merged PRs and turns each into one document: title + description + review comments + changed-file list. The relations[] list guarantees graph edges form between the PR and every source file it changed. Before: A GitHub token with repo scope. Create at github.com/settings/tokens → New classic token → check repo → add to .env as GITHUB_TOKEN, GITHUB_OWNER, GITHUB_REPO.

import sys, os, time, requests
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID
from phase1.batch_upload import upload_batch, verify_batch

GH_TOKEN = os.environ.get("GITHUB_TOKEN", "")
GH_OWNER = os.environ.get("GITHUB_OWNER", "")
GH_REPO  = os.environ.get("GITHUB_REPO",  "")
GH_HEADS = {"Authorization": f"Bearer {GH_TOKEN}",
            "Accept": "application/vnd.github+json"}

def fetch_merged_prs(max_pages: int = 5) -> list:
    base, prs = f"https://api.github.com/repos/{GH_OWNER}/{GH_REPO}", []
    for page in range(1, max_pages+1):
        r = requests.get(f"{base}/pulls", headers=GH_HEADS,
            params={"state":"closed","per_page":100,"page":page}, timeout=15)
        r.raise_for_status()
        page_prs = [p for p in r.json() if p.get("merged_at")]
        if not page_prs: break
        for pr in page_prs:
            url = pr["url"]
            fr = requests.get(f"{url}/files",   headers=GH_HEADS, timeout=10)
            rr = requests.get(f"{url}/reviews", headers=GH_HEADS, timeout=10)
            cr = requests.get(f"{url}/comments",headers=GH_HEADS, timeout=10)
            pr["files"]   = fr.json() if fr.ok else []
            pr["reviews"] = rr.json() if rr.ok else []
            pr["comments"]= cr.json() if cr.ok else []
            prs.append(pr); time.sleep(0.05)
        print(f"  Page {page}: {len(page_prs)} merged PRs")
    print(f"✓ Fetched {len(prs)} PRs")
    return prs

def ingest_pull_requests(prs: list, repo_name: str) -> list:
    batch, all_ids = [], []
    for pr in prs:
        if not pr.get("merged_at"): continue
        changed  = [f["filename"] for f in pr.get("files", [])]
        reviews  = "\n\n".join(r["body"] for r in pr.get("reviews",[]) if r.get("body"))
        comments = "\n\n".join(c["body"] for c in pr.get("comments",[]) if c.get("body"))
        # External IDs must exactly match those set in ingest_directory()
        source_ids = [f"{repo_name}/{fname}" for fname in changed]
        content = (
            f"PR #{pr['number']}: {pr['title']}\n"
            f"Author: {pr['user']['login']} | Merged: {pr['merged_at']}\n\n"
            f"Description:\n{pr.get('body') or '(none)'}\n\n"
            f"Changed files:\n" + "\n".join(changed) +
            f"\n\nReview comments:\n{reviews or '(none)'}\n\n"
            f"Inline comments:\n{comments or '(none)'}"
        )
        batch.append({
            "id":          f"github_pr_{pr['number']}",
            "tenant_id":   TENANT_ID,
            "sub_tenant_id": "github",
            "kind":        "custom",
            "provider":    "github",
            "external_id": f"{repo_name}/pull/{pr['number']}",
            "fields": {
                "kind": "custom",
                "data": {
                    "record_type": "pull_request",
                    "title": f"PR #{pr['number']}: {pr['title']}",
                    "body": content,
                    "author": pr["user"]["login"],
                    "merged_at": pr["merged_at"],
                    "changed_files": changed,
                },
            },
            "relations": [
                {"predicate": "linked_to", "target": {"external_id": sid, "provider": "github"}}
                for sid in source_ids
            ],
            "metadata": {"container_type":"repository","container_id":repo_name,
                      "source":"github","doc_type":"pull_request","pr_number":pr["number"],
                      "author":pr["user"]["login"]},
        })
        if len(batch)==20: all_ids+=upload_batch(batch); batch=[]; time.sleep(1)
    if batch: all_ids+=upload_batch(batch)
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} PRs indexed")
    return all_ids

if __name__ == "__main__":
    prs = fetch_merged_prs(max_pages=3)
    ingest_pull_requests(prs, "myrepo")

P2 · Step 3 - Ingest Slack Threads

What: Ingests each Slack message as a typed message app source. Messages share thread_id, and replies set parent_id to the root thread timestamp. Architecture channels, post-mortems, and decision threads are the highest-value sources - these capture approvals that never get written up anywhere else. Before: A Slack export ZIP. Go to your-workspace.slack.com/services/export (requires admin), export, and unzip. Structure: {export_dir}/{channel_name}/{YYYY-MM-DD}.json.

import sys, os, json, time, pathlib
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import TENANT_ID
from phase1.batch_upload import upload_batch, verify_batch

def ingest_slack_export(export_dir: str, channels: list[str]) -> list:
    batch, all_ids = [], []
    root = pathlib.Path(export_dir)
    for channel in channels:
        channel_dir = root / channel
        if not channel_dir.is_dir():
            print(f"  Skipping #{channel} - not found"); continue
        all_messages = []
        for day_file in sorted(channel_dir.glob("*.json")):
            try: all_messages.extend(json.loads(day_file.read_text()))
            except: continue
        threads: dict[str, list] = {}
        for msg in all_messages:
            key = msg.get("thread_ts") or msg["ts"]
            threads.setdefault(key, []).append(msg)
        for thread_ts, msgs in threads.items():
            for msg in msgs:
                if not msg.get("text"): continue
                ts_dt = datetime.fromtimestamp(float(msg["ts"])).isoformat() + "Z"
                batch.append({
                    "id":          f"slack_{channel}_{msg['ts']}",
                    "tenant_id":   TENANT_ID,
                    "sub_tenant_id": "slack",
                    "kind":        "message",
                    "provider":    "slack",
                    "external_id": msg["ts"],
                    "fields": {
                        "kind": "message",
                        "body": msg["text"],
                        "author": msg.get("user"),
                        "thread_id": thread_ts,
                        "parent_id": thread_ts if msg["ts"] != thread_ts else None,
                        "created_at": ts_dt,
                    },
                    "metadata": {"container_type":"channel","container_id":channel,
                              "container_name":channel,"source":"slack","doc_type":"slack_message"},
                })
                if len(batch)==20: all_ids+=upload_batch(batch); batch=[]; time.sleep(1)
        print(f"  Processed #{channel}: {len(threads)} threads")
    if batch: all_ids+=upload_batch(batch)
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} Slack threads indexed")
    return all_ids

if __name__ == "__main__":
    ingest_slack_export(
        export_dir="/path/to/unzipped-slack-export",
        channels=["eng-architecture", "incidents"],
    )

P2 · Step 4 - Ingest Wikis & RFCs

What: Ingests Architecture Decision Records, RFCs, Confluence pages, and Notion wikis. Tag each with doc_type so recall can be scoped to decisions-only when that is what a question requires.

import sys, os, time, pathlib
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import TENANT_ID
from phase1.batch_upload import upload_batch, verify_batch

def ingest_wikis(pages: list) -> list:
    """
    pages: list of dicts, each requires:
      id, title, content (str), doc_type, last_updated (ISO 8601)
    Optional: author, url
    doc_type: "rfc" | "adr" | "wiki" | "runbook" | "postmortem"
    """
    batch, all_ids = [], []
    for page in pages:
        provider = page.get("provider", "notion" if "notion" in page.get("url", "") else "confluence")
        batch.append({
            "id":          f"{provider}_page_{page['id']}",
            "tenant_id":   TENANT_ID,
            "sub_tenant_id": "docs",
            "kind":        "knowledge_base",
            "provider":    provider,
            "external_id": page["id"],
            "fields": {
                "kind": "knowledge_base",
                "title": page["title"],
                "body": page["content"],
                "updated_at": page["last_updated"],
                "url": page.get("url", ""),
                "updated_by": page.get("author", ""),
            },
            "metadata": {"container_type":"space","container_id":page.get("space", "engineering"),
                      "source":provider,"doc_type":page["doc_type"],"author":page.get("author","")},
        })
        if len(batch)==20: all_ids+=upload_batch(batch); batch=[]; time.sleep(1)
    if batch: all_ids+=upload_batch(batch)
    verify_batch(all_ids)
    print(f"✓ {len(all_ids)} wiki/RFC pages indexed")
    return all_ids

def ingest_markdown_folder(folder: str) -> list:
    """Convenience: ingest a local folder of .md files as wiki pages."""
    pages = []
    for f in pathlib.Path(folder).rglob("*.md"):
        pages.append({
            "id":           f.stem,
            "title":        f.stem.replace("-"," ").replace("_"," ").title(),
            "content":      f.read_text(encoding="utf-8",errors="ignore"),
            "doc_type":     "rfc" if "rfc" in f.name.lower() else "wiki",
            "last_updated": "2024-01-01T00:00:00Z",
        })
    return ingest_wikis(pages)

P2 · Step 5 - Run a Multi-source Graph Query

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from config import client, TENANT_ID

def multi_source_recall(question: str):
    return client.recall.full_recall(
        tenant_id=TENANT_ID,
        query=question,
        max_results=15,
        mode="thinking",
        graph_context=True,
        search_apps=True,
        alpha=0.65,
        recency_bias=0.15,
    )

if __name__ == "__main__":
    result = multi_source_recall(
        "Why does the auth middleware skip token validation for internal IPs?")
    for chunk in (result.chunks or [])[:5]:
        score = chunk.relevancy_score if chunk.relevancy_score is not None else 0
        print(f"[{score:.2f}] {chunk.source_title or ''}")
        print((chunk.chunk_content or "")[:240])

Expected output:

[0.88] RFC-007 Internal Service Auth
The auth middleware skips token validation for internal IPs...
[0.82] PR #142 Auth Startup Fix
The change resolved circular dependency issues at startup...
[0.77] #eng-architecture
Security approved the internal-network exception...

✅ Phase 2 complete. A “why” question now returns multi-hop retrieved context. Phase 3 wraps this in a production FastAPI server with your own GPT-4o answer generation layer.

Phase 3 - Backend & Answer Generation · 25–35 minutes

Build a production FastAPI server. HydraDB is the memory layer - it retrieves ranked chunks via full_recall(). GPT-4o is the reasoning layer - it reads those chunks and writes a grounded, cited answer. The two layers are kept strictly separate so they can be debugged and improved independently. Goal: POST /chat and POST /ask → HydraDB chunks → GPT-4o → answer.

Backend Architecture

The flow through the backend is:

Question - POST /chat or /ask
Recall - full_recall() → chunks
Context block - chunk_content assembled
GPT-4o - writes grounded answer
Response - answer + citations

Why split retrieval and reasoning? Because they fail differently. Weak recall means you improve ingestion, metadata, or graph links. Weak answers mean you improve prompt construction or model behavior. Keeping the layers separate means you can isolate which side is the problem. File structure:

backend/
├── __init__.py       - makes backend a package so imports work
├── config.py         - env vars, HYDRA_DB_API_KEY, HydraDB client, OpenAI keys, recall defaults
├── hydra_client.py   - full_recall() wrapper for HydraDB retrieval
├── recall.py         - recall_context() and build_context_block(); extracts chunk_content
├── answer.py         - prompt formatter, GPT-4o streaming call, anti-hallucination rules
└── app.py            - FastAPI server; /chat (streaming) and /ask (sync JSON, easier for Postman)

Architecture summary:

config.py - Loads .env, exposes HYDRA_DB_API_KEY, TENANT_ID, client (HydraDB SDK instance), OPENAI_API_KEY, and recall defaults. Every other file imports from here only.
hydra_client.py - full_recall() is the HydraDB retrieval path. It returns chunks, sources, and graph context; prompt formatting and OpenAI calls stay in answer.py. Imported by → recall.py, app.py.
recall.py - recall_context() calls full_recall() and returns the raw payload. build_context_block() extracts chunk_content from each chunk and assembles a formatted context string for the LLM. Imported by → app.py.
answer.py - Takes the context block from recall.py, formats the system + user prompt, and calls the OpenAI streaming API. Contains all anti-hallucination rules. Never calls HydraDB. Imported by → app.py.
app.py - Two endpoints: POST /chat streams NDJSON token-by-token - ideal for a web or IDE frontend. POST /ask returns a complete JSON response with answer, sources, and chunks - easier for Postman testing. Both use the same recall_context → build_context_block → stream_answer pipeline.

Install dependencies for Phase 3:

pip install fastapi uvicorn[standard] openai python-dotenv hydradb-sdk

Add to .env:

OPENAI_API_KEY=sk-your-key-here

Create the empty package init file:

mkdir -p backend && touch backend/__init__.py

backend/config.py

Centralises all credentials. Use HYDRA_DB_API_KEY as the variable name for your HydraDB key. All other backend files import from here.

import os
from dotenv import load_dotenv
from hydra_db import HydraDB

load_dotenv()

# HydraDB
HYDRA_DB_API_KEY = os.environ["HYDRA_DB_API_KEY"]
TENANT_ID     = os.environ.get("HYDRADB_TENANT_ID", "engineering-docs")

client = HydraDB(token=HYDRA_DB_API_KEY)

# OpenAI
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
OPENAI_MODEL   = "gpt-4o"

# Recall defaults
RECALL_MAX_RESULTS = 15
RECALL_MIN_SCORE   = 0.5   # drop chunks below this before building LLM context
RECALL_ALPHA       = 0.75  # semantic vs keyword balance

backend/hydra_client.py

full_recall() is the HydraDB retrieval path. It returns ranked chunks and source metadata; it does not format prompts or decide routing.

from backend.config import client, TENANT_ID, RECALL_MAX_RESULTS, RECALL_ALPHA


def full_recall(
    query: str,
    max_results: int = RECALL_MAX_RESULTS,
    scope: str = None,
    graph_context: bool = True,
    recency_bias: float = 0.2,
):
    """
    PRIMARY VERIFIED PATH - build and test your backend on this function first.

    Calls /recall/full_recall. scope (sub_tenant_id) is optional - omit to
    search all sub-tenants automatically. Response contains:
      "chunks"  - iterate this array; read chunk_content from each item for LLM context
      "sources" - deduplicated source list; use for citations
    """
    kwargs = {
        "tenant_id":     TENANT_ID,
        "query":         query,
        "max_results":   max_results,
        "mode":          "thinking",
        "graph_context": graph_context,
        "search_apps":   True,
        "alpha":         RECALL_ALPHA,
        "recency_bias":  recency_bias,
    }
    if scope:
        kwargs["sub_tenant_id"] = scope
    return client.recall.full_recall(**kwargs)

backend/recall.py

recall_context() calls full_recall() and returns the raw HydraDB payload. build_context_block() reads the chunks array, extracts each chunk_content field, and assembles them into a single formatted string that is passed to the LLM. The sources array from the payload is used separately by app.py for citation output.

from backend.hydra_client import full_recall


def recall_context(query: str, max_results: int = 10):
    """
    Call full_recall() with sensible defaults for the first backend.
    scope=None means HydraDB searches all sub-tenants automatically.
    Returns SDK object with .chunks and .sources attributes.
    """
    return full_recall(
        query=query,
        max_results=max_results,
        scope=None,            # no sub_tenant_id needed - HydraDB handles scope
        graph_context=True,
        recency_bias=0.2,
    )


def build_context_block(recall_payload) -> str:
    """
    Extract chunk_content from each chunk and assemble the LLM context string.

    chunks  - the main array; iterate this and read chunk_content from each item.
    sources - the deduplicated source list; used for citations in app.py.

    chunk_content is the canonical field to pass into the LLM.
    Always iterate chunks[], never a "results" key - the correct key is "chunks".
    """
    chunks  = recall_payload.chunks or []
    sources = recall_payload.sources or []

    context_parts = []
    for chunk in chunks:
        text = (chunk.chunk_content or "").strip()
        if text:
            context_parts.append(text)

    context_text = "\n\n".join(context_parts)

    source_lines = []
    for source in sources:
        title     = source.title or "Untitled"
        source_id = source.id or ""
        source_lines.append(f"- {title} ({source_id})")

    sources_text = "\n".join(source_lines)

    return f"Context:\n{context_text}\n\nSources:\n{sources_text}"

💡 Production upgrade: score filtering. build_context_block is intentionally simple for the first backend. For production, add a relevancy_score filter to drop low-quality chunks before building the context string - the same pattern used in Phase 1’s build_context() helper.

backend/answer.py

Takes the context block string from recall.py and turns it into a grounded GPT-4o answer. This is the only file that touches OpenAI. If the context string is empty, it returns a safe fallback without calling the model at all - this is the primary hallucination guard.

from typing import Generator
from openai import OpenAI
from backend.config import OPENAI_API_KEY, OPENAI_MODEL

_client = OpenAI(api_key=OPENAI_API_KEY)

SYSTEM_PROMPT = """You are a codebase AI assistant. You answer questions about
code, architecture decisions, pull requests, and engineering discussions.

RULES:
1. Only answer from the provided context.
2. If the context is insufficient, say:
   "I don't have enough context to answer that. Try rephrasing or check the relevant source directly."
3. Cite sources inline using [Source: exact_title_here].
4. If multiple sources support a claim, cite all of them.
5. Distinguish clearly between:
   - what the code does
   - why it was written that way
   - who approved or discussed it
6. Never speculate."""


def format_prompt(question: str, context: str) -> list[dict]:
    """
    Build the messages array for the OpenAI API.
    Context goes in the user message - not the system prompt.
    This keeps the system prompt stable and makes context easy to audit.
    """
    user_content = (
        "CONTEXT FROM CODEBASE (retrieved by HydraDB):\n\n"
        f"{context}\n\n"
        "---\n\n"
        f"QUESTION: {question}\n\n"
        "Answer using only the context above. Cite every source you use."
    )
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user",   "content": user_content},
    ]


def stream_answer(question: str, context: str) -> Generator[str, None, None]:
    """
    Streams GPT-4o tokens. If context is empty, yields a safe fallback
    WITHOUT calling OpenAI - prevents the model from hallucinating
    when HydraDB retrieval returned nothing.
    """
    if not context.strip():
        yield (
            "I couldn't find relevant context in the codebase index for that question. "
            "The index may still be building, or the topic may not be ingested yet. "
            "Try rephrasing, or check that the relevant files were uploaded."
        )
        return

    messages = format_prompt(question, context)
    stream = _client.chat.completions.create(
        model=OPENAI_MODEL,
        messages=messages,
        stream=True,
        max_tokens=1200,
        temperature=0.1,  # low temp = grounded, not creative
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content or ""
        if delta:
            yield delta

backend/app.py

The FastAPI server exposes two endpoints. POST /chat streams NDJSON - best for a web frontend or VS Code extension. POST /ask returns a complete JSON response - easier for Postman testing and best for API consumers that don’t stream. Both use the same recall_context → build_context_block → stream_answer pipeline.

💡 Want WHY_SIGNALS routing? Once this basic backend is working, add a question classifier that adjusts full_recall() parameters for “why” questions, such as lower alpha, higher max_results, and graph_context: true. Keep final answer generation in your app layer.

import json

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse

from backend.recall import recall_context, build_context_block
from backend.answer import stream_answer

app = FastAPI(title="Cursor for Docs API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],         # tighten in production
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)


@app.get("/")
def health():
    return {"status": "ok"}


@app.post("/chat")
async def chat(body: dict):
    """
    Streaming NDJSON endpoint - best for frontends and VS Code extensions.
    Line 1:  {"sources": [...]}
    Line 2+: {"text": "token chunk"}
    """
    question = body.get("question", "").strip()
    if not question:
        raise HTTPException(status_code=400, detail="question is required")

    recall_payload = recall_context(question, max_results=10)
    context_block  = build_context_block(recall_payload)

    async def stream():
        # Emit sources first so the client can render citations immediately
        sources_list = [{"title": s.title, "source_id": s.id} for s in (recall_payload.sources or [])]
        yield json.dumps({"sources": sources_list}) + "\n"
        for token in stream_answer(question, context_block):
            yield json.dumps({"text": token}) + "\n"

    return StreamingResponse(stream(), media_type="application/x-ndjson")


@app.post("/ask")
def ask(body: dict):
    """
    Synchronous JSON endpoint - easier for Postman testing and API consumers
    that don't stream. Returns a complete response object in one round-trip.
    Returns answer, sources, chunks, and graph_context.
    """
    question = body.get("question", "").strip()
    if not question:
        raise HTTPException(status_code=400, detail="question is required")

    recall_payload = recall_context(question, max_results=10)
    context_block  = build_context_block(recall_payload)
    full_answer    = "".join(stream_answer(question, context_block))

    return {
        "answer":        full_answer,
        "sources":       [{"title": s.title, "source_id": s.id} for s in (recall_payload.sources or [])],
        "chunks":        [{"chunk_content": c.chunk_content, "relevancy_score": c.relevancy_score,
                           "source_title": c.source_title} for c in (recall_payload.chunks or [])],
        "graph_context": None,
    }

Answer Generation Flow

Understanding the exact data path prevents debugging confusion. build_context_block() does the following in order:

Read the chunks array from the recall payload. /recall/full_recall returns {"chunks": [...], "sources": [...]} - always iterate chunks, never results.
Extract chunk_content from each chunk object. This is the raw text of the chunk - a few hundred tokens of the original document. It is the only required field. Skip any chunk where chunk_content is absent or empty.
Join all chunk_content values with double newlines to form the context body.
Append the sources list from recall_payload.sources in - title (id) format so the model can cite them.
Return the assembled string to stream_answer(). This becomes the user message body inside the OpenAI prompt.

The assembled context block that GPT-4o receives looks like this:

Context:
def validate_token(request):
    if is_internal_ip(request.remote_addr):
        return True   # See RFC-007
    ...

PR #142: feat: skip token validation for internal IPs
Author: alice | Merged: 2024-02-14
Description: Service-to-service calls in the VPC were failing at startup...
Review comments: [bob]: Approved. Security team signed off on #eng-architecture.

The VPC security group rules restrict internal traffic to approved service
accounts. Token validation for internal IPs is redundant and causes...

[alice]: Proposing we skip token validation for 10.x and 172.x ranges
[security-lead]: Approved as long as VPC security groups stay locked down

Sources:
- auth/middleware.py (myrepo/auth/middleware.py)
- PR #142: feat: skip token validation (pr-142)
- RFC-007: Internal Network Auth Exception (wiki-rfc-007)
- Slack - #eng-architecture - 2024-02-14 (slack-eng-arch-123)

Avoiding Hallucination

LLMs will confidently invent details about your codebase if you don’t actively prevent it. The system prompt in answer.py contains specific rules for each failure mode:

Failure mode	What happens	Prevention
Out-of-context answer	Model uses training data about similar-looking code	Rule 1: “Only answer from the provided context.” Temperature 0.1.
No-context answer	Empty context; model invents anyway	`stream_answer()` checks for empty context before calling OpenAI and returns a safe fallback.
Uncited claims	Confident statements with no traceable source	Rules 3 and 4: “Cite sources inline. If multiple sources support a claim, cite all of them.”
Speculation as fact	Model fills gaps with plausible-but-wrong reasoning	Rule 6: “Never speculate. If you are uncertain, say so explicitly.”

⚠️ Do not raise temperature above 0.2 for this use case. Higher temperature increases creativity, which is the opposite of what you want for accurate, cited answers about specific code.

Running the Server

uvicorn backend.app:app --reload --port 8000

Expected startup output:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process
INFO:     Application startup complete.

Test the streaming /chat endpoint:

curl -s -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"question": "why does auth skip token validation for internal IPs?"}' \
  | python3 -c "
import sys, json
for line in sys.stdin:
    obj = json.loads(line)
    if 'sources' in obj:
        print('Sources:', [s.get('title') for s in obj['sources']])
    elif 'text' in obj:
        print(obj['text'], end='', flush=True)
print()
"

Test the synchronous /ask endpoint (easier for Postman):

curl -s -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "what does the auth middleware do?"}' \
  | python3 -m json.tool

✅ Phase 3 complete. You have a production FastAPI backend. HydraDB is the memory layer - full_recall() retrieves ranked chunks. GPT-4o is the reasoning layer - it reads those chunks via stream_answer() and writes a grounded answer. The two layers are intentionally separate so you can debug and improve each independently.

Phase 4 - Productization · Coming next

Ship the assistant to your team. Build the VS Code extension sidebar, set up a daily incremental sync cron job, wire a GitHub Actions workflow for on-push indexing, and monitor recall quality over time. Goal: Team using it daily with a fresh index on every merge. Detailed implementation coming in the next revision. Everything in Phase 4 is packaging around a working Phase 3 server. Planned components: VS Code extension (TypeScript), Webview chat UI, Active-file context injection, Daily cron sync, GitHub Actions on push to main, Incremental re-index, Recall quality monitoring.

If Your Retrieval Quality Is Weak

Fix your IDs and metadata. Weak retrieval often starts upstream. Use stable IDs, clear titles, accurate timestamps, and useful metadata like doc_type, repo, and channel.
Add explicit relations. If “why” answers feel shallow, it usually means the assistant sees the code but not the documents around it. Add explicit relations[] between code, PRs, Slack, and RFCs.
Improve source formatting. A PR with only a title is weak. A PR with title, description, changed files, reviews, and inline comments is strong. Rich documents produce better chunks.
Add relevancy_score filtering. The production upgrade to build_context_block is to drop chunks where relevancy_score is below your threshold before assembling the context string.

How to Extend Beyond the First Version

A good progression: Phase 0 recall in the terminal → FastAPI /ask in Postman → FastAPI /chat with a simple web frontend → VS Code sidebar → streaming, incremental sync, caching, usage analytics.

Troubleshooting

401 - Not Authenticated

Cause: The Authorization header is missing or malformed. Most common: the Bearer prefix is absent, quotes surround the key in .env, or there is a leading/trailing space in the key value. Fix: Run python3 -c "from config import HYDRA_DB_API_KEY; print(repr(HYDRA_DB_API_KEY))". The output should be 'hdb_abc123' - no surrounding spaces. Your header must be exactly Authorization: Bearer hdb_abc123.

404 - Tenant Does Not Exist

Cause: You are attempting to upload or query before the tenant was created, or the tenant_id in your request does not match the one you created. Tenant IDs are case-sensitive. Fix: Run python3 phase0/create_tenant.py and wait for ”✓ Tenant ready.” before proceeding. Confirm TENANT_ID in config.py exactly matches what you passed to create_tenant.

404 - Not Found (on recall or ingestion endpoint)

Cause: The endpoint path has a typo, extra slash, or missing segment. Also occurs when querying a sub_tenant_id that has never had data written to it. Fix: Correct paths: /ingestion/upload_knowledge (file upload or app-source batch with app_knowledge), /recall/full_recall, /ingestion/verify_processing. If using sub_tenant_id, confirm at least one batch was successfully uploaded to that sub-tenant first.

400 - Provide at least one of: files or app_knowledge

Cause: For file upload: the field name is wrong (file instead of files), or Content-Type: application/json was manually set (which breaks multipart). For app-source batch: the array is empty or objects are missing required fields. Fix: For file upload: use files={"files": (filename, f, "text/markdown")} and data={"tenant_id": TENANT_ID}. Do NOT set Content-Type manually. For app-source batch: confirm each item has id, tenant_id, sub_tenant_id, kind, provider, external_id, and fields.

Empty Recall - chunks: []

Cause: Indexing is still in progress, or relevancy_score filtering in build_context_block is too aggressive, or the query does not semantically match any ingested content. Fix: Step 1 - Run verify_file() and confirm "indexing_status": "completed" in the statuses array. Step 2 - Run phase0/recall.py directly and print the raw response - check that the chunks key is present and non-empty before any filtering. Step 3 - Check the relevancy_score values on returned chunks.

429 - OpenAI insufficient_quota

Cause: Your OpenAI account has run out of credits or hit its usage limit. This is an OpenAI-side error - HydraDB retrieval may be working perfectly. The error appears in stream_answer() inside answer.py, not in the HydraDB recall step. Fix: Add billing or upgrade your OpenAI plan at platform.openai.com. While debugging, replace stream_answer in answer.py with this temporary debug fallback - it returns the raw retrieved context without calling OpenAI:

# TEMPORARY DEBUG MODE - replace stream_answer in answer.py to test
# retrieval without OpenAI. Remove this before going to production.
def stream_answer(question: str, context: str):
    yield "=== DEBUG MODE (NO OPENAI) ===\n\n"
    yield f"Question:\n{question}\n\n"
    yield f"Context:\n{context[:1000]}"

💡 General debugging pattern. Add print("STATUS CODE:", resp.status_code) and print("RESPONSE:", resp.text[:500]) immediately after every requests.post() call. See the full error body before doing anything else.

Production Notes

Batch size and rate limits. The app-source batch upload path accepts a maximum of 20 source objects per request. Always sleep 1 second between batches. For large repos (1,000+ files), expect ingestion to take 10–30 minutes total.
Indexing delays. Indexing is async. Never rely on a fixed sleep; always poll verify_processing until the statuses array shows "indexing_status": "completed".
Consistency and upserts. HydraDB upserts by id - re-uploading replaces the existing document. There is a brief window where recall may return stale chunks. Run verify on new IDs before marking a deployment complete.
LLM context window. The basic build_context_block assembles all chunk text. For production, add a relevancy_score filter and a character cap (e.g. 12,000 chars) to prevent exceeding GPT-4o’s context window.
Graph edge consistency. Graph edges only form if both sides exist and are indexed. Always ingest source files first, then PRs. Re-ingest affected PRs if you add new files after initial ingestion.
API key security. The HydraDB API key grants full access to all tenant data. Never commit it to git. Use environment secrets in production. Rotate immediately if you suspect exposure.
OpenAI token costs. Each question sends context + system prompt + question to GPT-4o. For a 50-engineer team asking 200 questions/day, budget $2–4/day. Use gpt-4o-mini for factual lookups to reduce cost.
Recall quality monitoring. Log the relevancy_score value of each chunk included in a response. If median scores are falling over time, new content may be diluting the index - re-ingest with better metadata and app-source fields. Also confirm the chunks array is non-empty before sending context to GPT-4o.

API Reference

All endpoints used in this cookbook. Base URL: https://api.hydradb.com. Header: Authorization: Bearer YOUR_API_KEY

Create Tenant

POST /tenants/create - Async, poll /tenants/infra/status before ingesting

Request:  { "tenant_id": "engineering-docs" }

Response:
{
  "tenant_id": "engineering-docs",
  "status":    "accepted",
  "message":   "Tenant accepted. Poll /tenants/infra/status for readiness."
}

GET /tenants/infra/status?tenant_id=engineering-docs
→ { "status": "ready" }

Upload File - Beginner Path

POST /ingestion/upload_knowledge - Multipart form-data · field name "files" · tenant_id as form field

# Correct multipart file upload
client.upload.knowledge(
    tenant_id=TENANT_ID,
    files=[(filename, f, "text/markdown")],
)

# Response:
{
  "results": [
    {
      "source_id": "<id>",
      "filename":  "auth_middleware.md",
      "status":    "accepted",
      "error":     null
    }
  ]
}

Upload App Sources - Advanced

POST /ingestion/upload_knowledge - Max 20 items · multipart field app_knowledge (JSON string) · supports typed fields and relations

import json

client.upload.knowledge(
    tenant_id="engineering-docs",
    app_knowledge=json.dumps([{
        "id": "github_myrepo_auth_middleware_py",
        "tenant_id": "engineering-docs",
        "sub_tenant_id": "code",
        "kind": "custom",
        "provider": "github",
        "external_id": "myrepo/auth/middleware.py",
        "fields": {
            "kind": "custom",
            "data": {
                "record_type": "source_file",
                "path": "auth/middleware.py",
                "body": "full content here",
                "language": "py",
                "updated_at": "2025-11-14T10:22:00Z"
            }
        },
        "relations": [
            {"predicate": "linked_to", "target": {"external_id": "myrepo/pull/42", "provider": "github"}}
        ],
        "metadata": {"container_type": "repository", "container_id": "myrepo", "source": "github"}
    }]),
)

# Response results include source_id values for verification/deletes.

Verify Indexing

POST /ingestion/verify_processing - Returns statuses[] array · poll until "completed"

// Required: ?file_ids=<id>&tenant_id=engineering-docs

// Response:
{
  "statuses": [
    { "file_id": "<id>", "indexing_status": "completed" }
  ]
}
// indexing_status: "processing" | "completed" | "errored"

Recall - Validated Minimal Request

POST /recall/full_recall - Primary verified path · sub_tenant_id optional · returns chunks[] and sources[]

{
  "tenant_id":   "engineering-docs",
  "query":       "internal IP auth skip logic",
  "max_results": 10
}

// sub_tenant_id is NOT required. HydraDB handles scope internally.
// Response: {"chunks": [...], "sources": [...]}
// Read chunk_content from each item in chunks[] for LLM context.
// relevancy_score on each chunk indicates ranked relevance.

Multi-hop Recall

POST /recall/full_recall - Use for multi-hop retrieval, then generate the answer in your app layer

{
  "tenant_id":     "engineering-docs",
  "query":         "Why does auth middleware skip token validation for internal IPs?",
  "max_results":   15,
  "mode":          "thinking",
  "graph_context": true,
  "search_apps":   true,
  "alpha":         0.65,
  "recency_bias":  0.15
}

Benchmarks

1,200 developer questions across three engineering teams (18–80 engineers, codebases 150k–2.2M lines) compared against naive vector search using identical ingested content.

Query type	Naive vector search	HydraDB with graph_context	Δ
”Why was X built this way?“	22%	87%	+295%
Cross-source recall (code + PR + Slack)	14%	79%	+464%
Factual lookup (“what does X do?“)	71%	91%	+28%
Decision trail completeness (3+ hops cited)	4%	68%	+1,600%
P95 recall latency	180ms	under 200 ms	Sub-second

ℹ️ The 22% accuracy on “why” questions isn’t a tuning problem - it’s structural. Embedding a source file and the RFC that motivated it produces similar-but-unlinked vectors. Without explicit graph edges, the RFC never surfaces when the source file matches. HydraDB makes the connection explicit: every node linked to your query gets traversed, not just nodes that look semantically similar.

AI Travel Planner AI Customer Support Agent with Memory

⌘I

​Prerequisites

​What You’ll Build

​Phase 0 - Minimal Working System · 5–10 minutes

​Prerequisites

​Step 1 - Create a Tenant

​Step 2 - Upload One File

​Step 3 - Verify Indexing

​Step 4 - Run Your First Recall Query

​Phase 1 - Improve Retrieval · 15–20 minutes

​How Retrieval Works

​P1 · Step 1 - Batch Upload with Explicit IDs (App-Source Ingestion)

​P1 · Step 2 - Metadata for Scoping and Citations

​P1 · Step 3 - Tuned Recall with Graph Context

​Phase 2 - Multi-source Context · 20–30 minutes

​Why Each Source Type Matters

​P2 · Step 1 - Ingest GitHub Source Files with Consistent IDs

​P2 · Step 2 - Ingest Pull Requests with Explicit Graph Relations

​P2 · Step 3 - Ingest Slack Threads

​P2 · Step 4 - Ingest Wikis & RFCs

​P2 · Step 5 - Run a Multi-source Graph Query

​Phase 3 - Backend & Answer Generation · 25–35 minutes

​Backend Architecture

​backend/config.py

​backend/hydra_client.py

​backend/recall.py

​backend/answer.py

​backend/app.py

​Answer Generation Flow

​Avoiding Hallucination

​Running the Server

​Phase 4 - Productization · Coming next

​If Your Retrieval Quality Is Weak

​How to Extend Beyond the First Version

​Troubleshooting

​401 - Not Authenticated

​404 - Tenant Does Not Exist

​404 - Not Found (on recall or ingestion endpoint)

​400 - Provide at least one of: files or app_knowledge

​Empty Recall - chunks: []

​429 - OpenAI insufficient_quota

​Production Notes

​API Reference

​Create Tenant

​Upload File - Beginner Path

​Upload App Sources - Advanced

​Verify Indexing

​Recall - Validated Minimal Request

​Multi-hop Recall

​Benchmarks

Prerequisites

What You’ll Build

Phase 0 - Minimal Working System · 5–10 minutes

Prerequisites

Step 1 - Create a Tenant

Step 2 - Upload One File

Step 3 - Verify Indexing

Step 4 - Run Your First Recall Query

Phase 1 - Improve Retrieval · 15–20 minutes

How Retrieval Works

P1 · Step 1 - Batch Upload with Explicit IDs (App-Source Ingestion)

P1 · Step 2 - Metadata for Scoping and Citations

P1 · Step 3 - Tuned Recall with Graph Context

Phase 2 - Multi-source Context · 20–30 minutes

Why Each Source Type Matters

P2 · Step 1 - Ingest GitHub Source Files with Consistent IDs

P2 · Step 2 - Ingest Pull Requests with Explicit Graph Relations

P2 · Step 3 - Ingest Slack Threads

P2 · Step 4 - Ingest Wikis & RFCs

P2 · Step 5 - Run a Multi-source Graph Query

Phase 3 - Backend & Answer Generation · 25–35 minutes

Backend Architecture

backend/config.py

backend/hydra_client.py

backend/recall.py

backend/answer.py

backend/app.py

Answer Generation Flow

Avoiding Hallucination

Running the Server

Phase 4 - Productization · Coming next

If Your Retrieval Quality Is Weak

How to Extend Beyond the First Version

Troubleshooting

401 - Not Authenticated

404 - Tenant Does Not Exist

404 - Not Found (on recall or ingestion endpoint)

400 - Provide at least one of: files or app_knowledge

Empty Recall - chunks: []

429 - OpenAI insufficient_quota

Production Notes

API Reference

Create Tenant

Upload File - Beginner Path

Upload App Sources - Advanced

Verify Indexing

Recall - Validated Minimal Request

Multi-hop Recall

Benchmarks