BrainClient SDK

BrainClient is the recommended way to interact with ContextBrain from Python. It provides a typed, mixin-based API that works in both gRPC mode (production) and local mode (development).

Quick Start

from contextunity.core import BrainClient

# gRPC mode (connects to running Brain service)
client = BrainClient(host="localhost:50051")

# Search
results = await client.search(
    tenant_id="my_project",
    query_text="How does RAG work?",
    limit=5,
    source_types=["document", "article"],
)
for r in results:
    print(f"Score: {r.score}, Content: {r.content}")

Mixin Architecture

BrainClient is composed of mixins, each providing a domain of operations:

Knowledge Operations (Semantic Search)

The Knowledge Store is designed for Retrieval-Augmented Generation (RAG). It allows you to store documents, index them using vector embeddings (pgvector), and relate them using a semantic Knowledge Graph.

The BrainClient abstracts away the complex gRPC protobuf construction, allowing you to easily perform semantic search, upsert chunks, and traverse Knowledge Graph relations natively in Python.

# Search
results = await client.search(
    tenant_id="my_project",
    query_text="How does PostgreSQL handle concurrency?",
    limit=10,
    source_types=["document"],
)

# Upsert
await client.upsert(
    tenant_id="my_project",
    content="PostgreSQL uses MVCC for concurrency...",
    source_type="document",
    metadata={"source": "docs", "page": 42},
    doc_id="doc_123",  # optional, for updates
)

# Knowledge Graph relation
await client.create_kg_relation(
    tenant_id="my_project",
    source_type="concept",
    source_id="PostgreSQL",
    relation="USES",
    target_type="concept",
    target_id="MVCC",
)

# Graph traversal
graph_result = await client.graph_search(
    tenant_id="my_project",
    entrypoint_ids=["PostgreSQL"],
    max_hops=2,
    allowed_relations=["USES", "REQUIRES"],
    max_results=200,
)

Memory Operations (Tracking the User)

While the Knowledge Store holds static, tenant-level information (like product catalogs or company policies), the Memory Store is designed for dynamic, user-specific data.

It is split into two concepts:

1. Episodic Memory: A rolling transcript of past interactions/chats (Session history).
2. User Facts: Extracted, long-term traits about a specific user (e.g., “User prefers Ukrainian language”, “User is allergic to peanuts”).

# Add episode
await client.add_episode(
    tenant_id="my_project",
    session_id="conv_abc123",
    role="user",
    content="What is RAG?",
)

# Get recent episodes
episodes = await client.get_recent_episodes(
    tenant_id="my_project",
    session_id="conv_abc123",
    limit=20,
)

# Upsert entity fact
await client.upsert_fact(
    tenant_id="my_project",
    entity_id="user_123",
    fact_type="preference",
    key="language",
    value="Ukrainian",
)

# Get user facts
facts = await client.get_user_facts(
    tenant_id="my_project",
    entity_id="user_123",
)

Dual Mode

Every method supports both gRPC and local modes:

Mode	When	How
gRPC	Production, multi-service	BrainClient(host="brain:50051")
Local	Development, testing	BrainClient(local_store=my_store)

In local mode, calls go directly to the storage layer without network overhead.