Build your own Cursor

import os
import time

from hydra_db import HydraDB
from hydra_db.errors import ConflictError
from hydra_db.helpers import build_string
from openai import OpenAI

client = HydraDB(token=os.environ["HYDRADB_API_KEY"])
llm = OpenAI()


def create_database(database, schema=None):
    """Create the database if it is new, then wait until it accepts context."""
    try:
        if schema:
            client.databases.create(database=database, database_metadata_schema=schema)
        else:
            client.databases.create(database=database)
    except ConflictError:
        pass  # The database already exists.
    while not client.databases.status(database=database).data.infra.ready_for_ingestion:
        time.sleep(5)


def source_ids(result):
    """Return the IDs an ingest accepted. Raise if it rejected any item."""
    for item in result.data.results:
        if item.error_code:
            raise RuntimeError(f"{item.filename or item.id}: {item.error_code} {item.error}")
    return [item.id for item in result.data.results]


def wait_indexed(database, collection, ids):
    """Block until every source has finished indexing. Indexing runs in the background."""
    while True:
        statuses = client.context.status(
            database=database, collection=collection, ids=ids
        ).data.statuses
        for status in statuses:
            if status.indexing_status == "errored":
                raise RuntimeError(f"{status.error_code}: {status.error_message}")
        if all(s.indexing_status == "completed" for s in statuses):
            return
        time.sleep(2)


def answer(question, result, instructions=""):
    """Send a query result to the model and return its answer."""
    completion = llm.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "Answer using only the context provided. If the answer "
                "is not in the context, say you don't know. " + instructions,
            },
            {
                "role": "user",
                "content": f"Context:\n{build_string(result)}\n\nQuestion: {question}",
            },
        ],
    )
    return completion.choices[0].message.content

import hashlib
import json
from pathlib import Path

from hydra import client, create_database, llm, source_ids, wait_indexed

DATABASE = "code"
REPO = Path("path/to/your/repo")
COLLECTION = REPO.name.replace("-", "_")
EXTENSIONS = {".py", ".ts", ".go", ".rs", ".java", ".c", ".h", ".cpp", ".hpp", ".md"}

create_database(DATABASE)

def source_id(path):
    return f"{COLLECTION}_{hashlib.sha1(str(path).encode()).hexdigest()[:16]}"


files = [
    p for p in REPO.rglob("*")
    if p.is_file() and p.suffix in EXTENSIONS and ".git" not in p.parts and p.stat().st_size
]
ids = []
for start in range(0, len(files), 20):
    batch = files[start : start + 20]
    handles = [open(p, "rb") for p in batch]
    try:
        result = client.context.ingest(
            type="knowledge",
            database=DATABASE,
            collection=COLLECTION,
            documents=[(f"{p.name}.txt", h, "text/plain") for p, h in zip(batch, handles)],
            document_metadata=json.dumps([
                {
                    "id": source_id(p.relative_to(REPO)),
                    "additional_metadata": {"path": str(p.relative_to(REPO))},
                }
                for p in batch
            ]),
        )
    finally:
        for h in handles:
            h.close()
    ids += source_ids(result)

for start in range(0, len(ids), 50):
    wait_indexed(DATABASE, COLLECTION, ids[start : start + 50])

def search_code(question):
    result = client.query(
        database=DATABASE,
        collection=COLLECTION,
        query=question,
        type="knowledge",
        query_by="hybrid",
        mode="thinking",
        graph_context=True,
        max_results=8,
    )
    data = json.loads(result.data.json(exclude_none=True))
    context = {
        "chunks": data.get("chunks") or [],
        "graph_context": data.get("graph_context") or {},
        "additional_context": data.get("additional_context") or {},
    }
    for chunk in context["chunks"]:
        chunk["path"] = (chunk.get("additional_metadata") or {}).get("path", chunk.get("source_title"))
    return json.dumps(context, ensure_ascii=False)

TOOLS = [{
    "type": "function",
    "function": {
        "name": "search_code",
        "description": "Search the repository. Ask one natural-language question.",
        "parameters": {
            "type": "object",
            "properties": {"question": {"type": "string"}},
            "required": ["question"],
        },
    },
}]


def ask(question):
    messages = [
        {
            "role": "system",
            "content": "You answer questions about a codebase. Search before you "
            "answer, and cite the file path for every claim.",
        },
        {"role": "user", "content": question},
    ]
    while True:
        reply = llm.chat.completions.create(
            model="gpt-4o", messages=messages, tools=TOOLS
        ).choices[0].message
        if not reply.tool_calls:
            return reply.content
        messages.append(reply)
        for call in reply.tool_calls:
            found = search_code(json.loads(call.function.arguments)["question"])
            messages.append({"role": "tool", "tool_call_id": call.id, "content": found})


print(ask("How does request authentication work, and where is it enforced?"))