Typed graph | Brain Docs

Every verb on this page is BrainClient-only. Typed-graph writes are checked per-type against the active schema: a write referencing a declared type is accepted; an explicit create referencing an undeclared type is rejected. The seeded brain: system namespace is always active.

Entities

Create, resolve, and manage typed entities.

from brain_db_sdk import BrainClient, Auth, new_id
from brain_db_sdk.wire.types import (
    EntityCreateRequest,
    EntityResolveRequest,
    EntityGetRequest,
)

with BrainClient.connect("127.0.0.1", 9090, Auth.token(b"my-token")) as client:
    created = client.create_entity(EntityCreateRequest(
        entity_type_id=1,
        canonical_name="Ada Lovelace",
        aliases=["Ada"],
        attributes_blob=[],       # opaque bytes, as a list of ints
        request_id=new_id(),
    ))
    entity_id = created.entity_id   # 16-byte id

    got = client.get_entity(EntityGetRequest(entity_id=entity_id))

    # Resolve a candidate name to an entity, minting one on a miss.
    resolved = client.resolve_entity(EntityResolveRequest(
        candidate_name="Ada",
        context="",
        entity_type_hint=1,        # must be non-zero; 0 = no hint is rejected
        allow_create=True,
        request_id=new_id(),
    ))
    print(resolved.outcome, resolved.resolved_entity)

Other entity verbs: update_entity, rename_entity, merge_entities, unmerge_entity (reversible within a grace window), tombstone_entity, and list_entities (paginated — see Pagination).

Statements

A statement asserts a typed claim — (subject, predicate, object) — with confidence, evidence, and a bi-temporal validity window. The object is an externally-tagged StatementObject (EntityRef / MemoryRef / StatementRef / Value); evidence is an EvidenceRef.

from brain_db_sdk.wire.types import (
    StatementCreateRequest,
    StatementObject,
    EvidenceRef,
)

stmt = client.create_statement(StatementCreateRequest(
    kind="Fact",
    subject=entity_id,
    predicate="bornIn",
    object=StatementObject("EntityRef", place_id),
    confidence=0.95,
    evidence=EvidenceRef.inline([source_memory_id]),
    extractor_id=0,
    valid_from_unix_nanos=0,
    valid_to_unix_nanos=0,
    event_at_unix_nanos=0,
    schema_version=0,
    request_id=new_id(),
))
print(stmt.statement_id, stmt.auto_superseded, stmt.chain_root)

The response reports any statement this one auto-superseded and the supersession-chain root. Other statement verbs: get_statement, supersede_statement, tombstone_statement, retract_statement (schedules a hard-zero for a genuine mistake), statement_history (walks the version chain), and list_statements.

Relations

A relation connects two entities with a type, properties, evidence, and validity window.

from brain_db_sdk.wire.types import RelationCreateRequest, EvidenceRef

rel = client.create_relation(RelationCreateRequest(
    relation_type="worksWith",
    from_entity=entity_id,
    to_entity=colleague_id,
    properties_blob=[],
    evidence=EvidenceRef.inline([source_memory_id]),
    extractor_id=0,
    confidence=0.9,
    valid_from_unix_nanos=0,
    valid_to_unix_nanos=0,
    request_id=new_id(),
))
print(rel.relation_id)

Other relation verbs: get_relation (with follow_supersession), supersede_relation, tombstone_relation, traverse_relations (multi-hop walk), list_relations_from, and list_relations_to.

Schema

Upload, validate, fetch, and list schema documents. Uploads merge additively into the active namespace.

from brain_db_sdk.wire.types import SchemaUploadRequest

result = client.upload_schema(SchemaUploadRequest(
    schema_document=my_schema_dsl,
    dry_run=False,
    allow_breaking=False,
    request_id=new_id(),
))

With dry_run=True the server validates without applying and returns any validation errors plus a backward-compatibility verdict. Other schema verbs: get_schema (version == 0 selects the active version), validate_schema, and list_schemas.

Transactions

The client mints the txn_id; subsequent writes carry it to enroll until commit or abort.

from brain_db_sdk import new_id
from brain_db_sdk.wire.types import (
    TxnBeginRequest, TxnCommitRequest, TxnAbortRequest,
)

txn_id = new_id()
client.txn_begin(TxnBeginRequest(txn_id=txn_id, timeout_seconds=30))
try:
    # ... enroll writes by passing txn_id on their requests ...
    client.txn_commit(TxnCommitRequest(txn_id=txn_id))
except Exception:
    client.txn_abort(TxnAbortRequest(txn_id=txn_id))
    raise

Subscriptions

subscribe opens a long-lived change feed. It returns a Subscription the caller drains with next() or by iterating; call unsubscribe() for a clean teardown.

from brain_db_sdk.wire.types import SubscribeRequest, SubscriptionFilter

sub = client.subscribe(SubscribeRequest(
    filter=SubscriptionFilter(contexts=None, kinds=None, similar_to=None, agents=None),
    include_history=False,
    from_lsn=None,
    max_inflight=64,
))
try:
    for event in sub:            # blocks until the next SUBSCRIBE_EVENT
        handle(event)
finally:
    sub.unsubscribe()

The server pushes SUBSCRIBE_EVENT frames until the subscription is torn down — useful for observing the async write-derivation stages (pending_stages on an EncodeResponse) as they land.

Graph export

graph_fetch exports the caller's whole typed graph as a node/edge set, paginated by an opaque cursor. See Pagination.

from brain_db_sdk.wire.types import GraphFetchRequest

nodes, edges = client.graph_fetch(GraphFetchRequest(
    limit=500,
    cursor=b"",
    include_statements=True,
    include_memories=False,
    include_tombstoned=False,
))