Python SDK

breeth — official Python client. Sync + async, Pydantic types, httpx transport.

breeth is a thin Python client for the Breeth REST API. Sync + async variants, Pydantic v2 response models, httpx under the hood.

Quickstart

from breeth import BreethClient

client = BreethClient()  # reads BREETH_API_KEY from env

# Write
client.write(
    content="Sridhar prefers async Rust over Go for IO-heavy services.",
    group_id="default",
    extract_intent=True,
)

# Retrieve
res = client.retrieve(
    query="What language does Sridhar prefer?",
    group_id="default",
    limit=5,
)

for edge in res.edges:
    print(edge.fact, edge.confidence)

Async

AsyncBreethClient mirrors the sync API one-to-one, returning awaitables.

import asyncio
from breeth import AsyncBreethClient

async def main():
    async with AsyncBreethClient() as client:
        res = await client.retrieve(query="gRPC", group_id="default", limit=3)
        for edge in res.edges:
            print(edge.fact)

asyncio.run(main())

The async client must be used in an async with context (or call .aclose() manually) so httpx's connection pool gets cleaned up.

Methods

`client.write(...)`

Write an episode.

res = client.write(
    content="Decision: switching to gRPC for inference.",
    group_id="default",
    source_description="api",
    extract_intent=False,
)
res.episode_name        # 'api_1778…'
res.extracted.entities  # int
res.cogram.mode         # 'sync' | 'async'

`client.retrieve(query, group_id, limit)`

Hybrid BM25 + vector + graph search.

res = client.retrieve(query="…", group_id="default", limit=10)
for edge in res.edges:
    print(edge.fact, edge.source_node, edge.target_node)

The response keeps the server's underscore-prefixed metadata fields (_cache, _tier, _source, _note) for parity with the REST shape — Pydantic exposes them as cache, tier, source, note on the response object.

`client.entity(name, mode=None, limit=None)`

"Tell me about X." Name match is substring (case-insensitive). UUID also accepted for exact lookup.

view = client.entity("Sridhar", mode="all", limit=20)
view.narrative   # list[EntityNarrativeRow]
view.edges       # list[EntityEdgeRow]
view.episodes    # list[EntityEpisodeRow]

`client.groups(query=None, limit=None, offset=None)`

List graph partitions in the team scope.

Graph reads

ents = client.graph.list_entities(limit=20)
edges = client.graph.list_edges(limit=20)
eps = client.graph.list_episodes(limit=20)
details = client.graph.node_details(uuid_or_name)

Error handling

Non-2xx responses raise BreethError:

from breeth import BreethError

try:
    client.write(content="")
except BreethError as e:
    print(e.status)  # 422
    print(e.slug)    # 'validation_error'
    print(e.body)    # full server response

Same status code semantics as the Node SDK.

End-user (B2B2C) pass-through

client.write(content="…", end_user_id="user-42")

Forwards X-End-User-Id on the call. Use this when calling Breeth on behalf of your users.

Pydantic models

Every response is a Pydantic v2 model. You can call .model_dump() to get plain dicts, or .model_dump_json() for raw JSON. Import types directly:

from breeth import (
    BreethClient,
    AsyncBreethClient,
    BreethError,
    WriteResponse,
    RetrieveResponse,
    EntityResponse,
    GraphEntityListResponse,
    NodeDetailsResponse,
)

License + source

MIT. Source: github.com/Gramies/cogram-sdk-python. Issues + PRs welcome.