SDK reference

The SDK lives in owlgraph_core.sdk. It’s a thin HTTP wrapper around the platform’s retrieve endpoint — the same surface the owlgraph CLI uses, so behavior matches exactly.

pip install owlgraph-core

Python 3.12+. Runtime deps: httpx. No build step.

from owlgraph_core import sdk as owl

db = owl.connect(
    database_id="abc-123-...",
    api_key="sk-owl-...",        # optional if OWLGRAPH_API_KEY is set
    url="https://api.owlgraph.ai",  # default; override for staging
    timeout=120.0,                # seconds
)

Returns a Database. Holds a long-lived httpx.Client — keep the instance around for your app’s lifetime, or use as a context manager:

with owl.connect(database_id="...") as db:
    result = db.retrieve("...")
# sockets closed here

1. Arguments to connect() / Database(...)
2. OWLGRAPH_API_KEY / OWLGRAPH_PLATFORM_URL environment variables
3. (Coming soon) ~/.config/owlgraph/config.toml — written by owlgraph auth login

The core method. Runs an ontology-aware retrieval and returns a structured result.

result = db.retrieve(
    "Which suppliers ship to states where we're licensed?",
    module="aragog",   # override the DB's configured RAG module for this call
    trace="full",      # "summary" (default) or "full" (verbose trajectory)
)

print(result.answer)
print(f"{result.loops} loops, ${result.cost_usd:.4f}")

for p in result.passages:
    print(p.source, p.text[:200])

for step in result.trajectory:
    print(step.tool, step.arguments)

• question (str) — the user-facing question. The inner agent rewrites it as needed.
• module (str, optional) — overrides the database’s configured RAG module for this call. One of naive, owl, arag, aragog. Leave unset to use the DB default.
• trace (“summary” | “full”) — controls trajectory verbosity. Default "summary".

import asyncio
from owlgraph_core import sdk as owl

async def main():
    async with owl.AsyncDatabase(database_id="...") as db:
        result = await db.aretrieve("Who invented Hawaiian pizza?")
        print(result.answer)

asyncio.run(main())

Same arguments, same return type. Use under asyncio when you want non-blocking I/O — particularly handy when fanning out many concurrent queries.

from owlgraph_core.sdk import AuthError, RateLimitError, OWLGraphError

try:
    result = db.retrieve("...")
except AuthError as e:
    # 401 / 403 — refresh credentials, fail fast
    raise
except RateLimitError as e:
    # 429 — back off
    time.sleep(e.retry_after or 30)
except OWLGraphError as e:
    # everything else — e.status_code + e.body have details
    raise

All SDK errors carry status_code and a parsed body (when the server returned JSON).

The SDK doesn’t ship MCP tool definitions yet — for now, wrap retrieve in your own tool call:

def retrieve_tool(question: str) -> str:
    result = db.retrieve(question)
    return result.answer or "No answer found."

Then hand retrieve_tool to your agent framework of choice (LangChain, Pydantic AI, raw Anthropic / OpenAI tool calling). The platform’s hosted MCP endpoint (POST /mcp) is the alternative when your agent already speaks MCP.

from concurrent.futures import ThreadPoolExecutor

def score(q, expected):
    r = db.retrieve(q)
    return q, expected, r.answer, r.cost_usd

with db, ThreadPoolExecutor(max_workers=8) as pool:
    rows = list(pool.map(lambda qe: score(*qe), eval_set))

A single Database instance is thread-safe (its httpx.Client pools connections internally). For asyncio-native concurrency, use AsyncDatabase and asyncio.gather.

The SDK doesn’t expose the SSE chat endpoint yet (Phase B follow-up). For now, hit POST /api/v1/databases/<id>/chat/conversations/<cid>/messages directly — see the HTTP API reference.