kori-brain 0.1.0

Official Python SDK for the kBrain memory API

kBrain Python SDK

Thin HTTP client for the kBrain memory API. Authenticate with a per-user kb_live_… API key; the SDK only talks to user-scoped /v1/* routes.

pip install kori-brain

from kbrain import KBrain

client = KBrain(api_key="kb_live_...")  # or set KBRAIN_API_KEY

job = client.memories.add(
    text="I live in Berlin and love espresso.",
    container_tag="org:myorg:user:me",  # see user_tag()
)
print(client.memories.status(job["job_id"]))

hits = client.search("Where do I live?", limit=5)
print(hits["results"])

reply = client.chat("What coffee do I like?")
print(reply["reply"])

Note: PyPI name is kori-brain; import is still kbrain. Do not install this package and the internal kbrain-server editable package in the same virtualenv.

Install & auth

Setting	Env var	Default
API key	KBRAIN_API_KEY	— (required)
Base URL	KBRAIN_BASE_URL	https://api.kbrain.kori.one

Obtain a kb_live_… key from Kori Memory settings (or your provisioning flow). Pass it to the constructor or via the environment.

from kbrain import KBrain, AsyncKBrain, user_tag

# Sync
with KBrain() as client:
    print(client.health())

# Async
# async with AsyncKBrain() as client:
#     await client.health()

tag = user_tag("my-org", "user-123")  # → org:my-org:user:user-123

The SDK always sends X-API-Key only. It does not support bootstrap/global keys, X-User-Id impersonation, or /v1/admin/*.

API reference

SDK	REST
client.memories.add(...)	POST /v1/memories
client.memories.add_batch(...)	POST /v1/memories/batch
client.memories.status(job_id)	GET /v1/memories/{job_id}/status
client.memories.stream(job_id)	GET /v1/memories/{job_id}/stream (SSE)
client.memories.wipe(...)	POST /v1/memories/wipe
client.search(...)	POST /v1/search
client.chat(...)	POST /v1/chat
client.chat_stream(...)	POST /v1/chat/stream (SSE)
client.profile(tag)	GET /v1/profile/{tag}
client.graph(...)	GET /v1/graph
client.graph_stream(...)	GET /v1/graph/stream (SSE)
client.entity_graph(...)	GET /v1/entity-graph
client.usage.get / client.usage.clear	GET|DELETE /v1/usage
client.health()	GET /health

Memories

from kbrain import KBrain, user_tag

tag = user_tag("org", "user")
with KBrain() as client:
    job = client.memories.add(text="…", container_tag=tag, title="note", effort="medium")
    # or url=...
    client.memories.add_batch(
        [{"text": "chunk one"}, {"text": "chunk two", "title": "t"}],
        container_tag=tag,
    )
    client.memories.status(job["job_id"])
    for event in client.memories.stream(job["job_id"]):
        print(event)

Search, chat, profile, graph, usage

hits = client.search("coffee", container_tags=[tag], limit=5, mode="memories")
reply = client.chat("What do I like?", history=[{"role": "user", "content": "hi"}])
for ev in client.chat_stream("Tell me more"):
    if ev.get("type") == "token":
        print(ev.get("text"), end="")

client.profile(tag)
client.graph(container_tag=tag, limit=500)
for ev in client.graph_stream(container_tag=tag):
    print(ev.get("type"))
client.entity_graph(tag, hops=2)
client.usage.get(container_tag=tag)
client.usage.clear(container_tag=tag)  # writable tag only

Security model

• Each kb_live_… key is bound to a user/org. The server ACL only allows tags that key can read/write.
• Admin routes (/v1/admin/*), bootstrap/global keys, and global wipe are not exposed by this SDK.
• client.memories.wipe() is destructive but user-scoped: it deletes memories/documents for one writable container (defaults to the key’s own user tag when omitted). It cannot wipe other users or the whole deployment.

See examples/wipe_own_data.py (requires KBRAIN_CONFIRM_WIPE=yes).

Errors & timeouts

Exception	When
AuthenticationError	Missing key or HTTP 401
PermissionError	HTTP 403 (ACL)
NotFoundError	HTTP 404
RateLimitError	HTTP 429 (after light retries)
APIError	Other 4xx/5xx
KBrainError	Base class

Default timeout: 60s (connect 10s). Override with timeout= on the client. Transient 429 / 5xx are retried a couple of times.

from kbrain import KBrain, AuthenticationError, APIError

try:
    with KBrain() as client:
        client.search("hi")
except AuthenticationError:
    print("Check KBRAIN_API_KEY")
except APIError as e:
    print(e.status_code, e)

Async

import asyncio
from kbrain import AsyncKBrain

async def main():
    async with AsyncKBrain() as client:
        print(await client.health())
        print(await client.search("hello"))

asyncio.run(main())

Examples

Runnable scripts under examples/:

• quickstart.py — add → poll status → search
• ingest_and_search.py — batch + SSE + usage
• chat.py — sync + streaming chat
• wipe_own_data.py — user-scoped wipe (opt-in)

export KBRAIN_API_KEY=kb_live_...
python examples/quickstart.py

Development

cd kbrain-sdk
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest
python -m build

Maintainer publish checklist: PUBLISH.md.