Python — lbb

lbb is the Python client. The HTTP client (LbbClient / AsyncLbbClient) is the integration path for applications. It talks to lbb-server with a stack API key or a single-mode token as a bearer credential, the same surface the TypeScript SDK and MCP server use. It is built on httpx and pydantic.

Source and issues: github.com/littlebigbrains/lbb-python (Apache-2.0).

Install

pip install littlebigbrain   # imports as `lbb`

Sync client

from lbb import LbbClient, LbbError

with LbbClient(
    "https://0abc1def--production.db.eu.littlebigbrain.com",
    api_key="lbb_sk_live_...",
    graph="main",
) as lbb:
    # Load RDF. The graph is created on first write if it does not exist.
    imported = lbb.graph("main").facts.import_rdf(
        turtle_document,
        format="turtle",
        base_iri="https://example.org/data/",
        idempotency_key="catalog-2026-07",
    )

    # Query the published snapshot. `min_indexed_seq` waits for it to cover
    # the sequence the load committed.
    results = lbb.sparql(
        """
        SELECT ?s ?label WHERE {
            ?s <http://www.w3.org/2000/01/rdf-schema#label> ?label
        } LIMIT 10
        """,
        min_indexed_seq=imported["committed_commit_seq"],
    )
    for row in results:
        print(row["s"], row["label"])

import_rdf accepts ntriples, turtle, nquads, and trig. Keep each request under the 64 MiB body cap. N-Triples and N-Quads are line-oriented, so a large file can be split on line boundaries; Turtle and TriG are document grammars, so each request must be a complete parseable document. Pass blank_node_scope when separate source documents reuse labels such as _:b0; omit it to preserve blank labels exactly across chunks and retries.

Write statements

lbb.commit({
    "triplets": [{
        "source": {"type": "SERVICE", "name": "auth-service"},
        "relation": "WRITES_TO",
        "target": {"type": "DATABASE", "name": "user-db"},
        "confidence": 0.93,
        "evidence": "auth-service writes identity records to user-db",
    }],
}, idempotency_key="import-2026-06-13")

commit_dry_run(body) runs the same ontology and shape validation a real commit would and reports the would-be effect without writing anything.

create_graph() explicitly creates the scoped graph with the built-in ontology. For a custom ontology, call lbb.ontology.define(...) before the first commit. Defining the ontology creates the graph head, and the ontology cannot be replaced after graph creation.

Durable bulk import

submit_import_ndjson consumes a sync or async iterable of entity and edge records lazily and requires an explicit content-bound idempotency key:

accepted = lbb.submit_import_ndjson(
    ({"type": "DOCUMENT", "name": row.title, "key": row.id,
      "properties": row.as_properties()} for row in source_rows()),
    idempotency_key="sharepoint:run-2026-07-29",
)
status = lbb.wait_for_import_job(accepted.job_id)
if status.state.value != "succeeded":
    raise RuntimeError(status.failure.message if status.failure else status.state)
print(status.committed_commit_seq, status.publication_job)

AsyncLbbClient exposes the same methods as coroutines and accepts async iterables. Both clients verify durable_import_jobs_v1 and never silently invoke synchronous import on a server that does not report it. Empty iterables fail locally before a POST. Strong SPARQL can read each acknowledged commit directly; the optional publication waiter follows base reconciliation through its own deadline.

For several RDF documents, graph.facts.import_rdf_many(...) defers every intermediate reconciliation and triggers one final pass. Pass its final_sequence to graph.wait_for_published(...) only when the immutable RDF base itself must cover the whole import.

SPARQL

client.sparql(query) runs SPARQL 1.1 text (SELECT or ASK) through the conformant engine and returns a parsed SparqlResults, so no manual json.loads is needed:

results = lbb.sparql("""
    SELECT ?service ?db WHERE {
        ?service <https://littlebigbrain.com/r/writes_to> ?db
    } LIMIT 10
""")

print(results.vars)           # ['service', 'db']
for row in results:           # iterates flat {var: value} dicts
    print(row["service"], "->", row["db"])

answer = lbb.sparql("ASK { ?s ?p ?o }").boolean   # True / False

SparqlResults exposes .vars, .boolean (ASK), .bindings (raw typed terms), .rows() (flattened dicts, also what iteration yields), and .row_page. The default is asserted-only (entailment="none", reason=False). Pass entailment="subclass", "rdfs", or "owl" to apply that regime at query time (see reasoning); reason=True returns a typed error because stored rules already run at publish time.

To pin a query to a past commit, use the native /sparql endpoint’s ?as_of_commit_seq= and ?as_of_valid_time= parameters through raw_request, or see the HTTP API.

Lowercase IRIs

Entity types, relations, and properties created by little big brain are addressed as <https://littlebigbrain.com/{r,class,p}/name> with a lowercase local name. IRIs you loaded through import_rdf keep their original form.

Structured query

client.sparql_select(body) is the JSON form of a SELECT/ASK: a basic graph pattern with typed filters, GROUP BY (entity keys, scalar property keys, or calendar date buckets), aggregates, HAVING, and ORDER BY. It returns the typed vars/solutions/groups response.

entities.filter_by_attributes(...) builds that structured body when you already have relation patterns and need typed attribute predicates:

lbb.entities.filter_by_attributes(
    patterns=[{"subject": {"var": "service"}, "predicate": "WRITES_TO", "object": {"var": "db"}}],
    where=[{"field": "slo", "op": "ge", "value": 0.99}, {"var": "db", "field": "tier", "value": "prod"}],
    select=["service"],
)

Bodies may be plain dicts or instances of the generated Pydantic models in lbb.models.

Read a bounded type sample from the Base family pinned by the published generation:

sample = lbb.entities.sample(type="SERVICE", limit=20)
print(sample.total_count)
for entity in sample.entities:
    print(entity.entity.name)

There is no whole-graph entity iterator in the published-snapshot architecture. Use bounded SPARQL for precise relation-bound selection.

Ontology, shapes & validation

ontology = lbb.ontology.view(counts=True)
report = lbb.ontology.conformance()          # published SHACL report
print(report.conforms, report.result_count, report.validated_at_seq)

bundle = lbb.schema.view()
lbb.schema.publish({
    "desired_mode": "warn",
    "shapes": {"source": shapes_ttl, "format": "turtle"},
})                                           # shapes, atomically

There is no request-time run_shacl() or ShaclQueryRequest. Publishing changes the active schema; conformance reads the immutable report produced for that schema and never accepts inline shapes.

Generated ontology operations have stable public names and an op discriminator (AddEntityTypeOp, AddRelationOp, AddPropertyOp, WidenRelationOp, and the subtractive counterparts). Use AdditiveOntologyEvolveRequest when a structured-output system must be unable to propose narrowing or removal operations. client.ontology.evolve(request, dry_run=True) shows the exact typed diff, predicted ontology version, data conflicts, publishable, and no_op before the same request is published.

Connector onboarding can stay isolated from production data with client.ontology.draft_create(...). The returned OntologyDraft persists its sample evidence references, competency-question analyses, query outlines, additive operations, coverage, pitfalls, and confidence at a pinned snapshot. Continue with draft_validate, then draft_promote (retry-safe with an idempotency key) or draft_reject.

Branches

A branch is a copy-on-write snapshot. Scope a client to the new branch, fork it, then merge it back as one validated commit:

review = LbbClient(base_url, api_key=key, graph="main", branch="review")
review.raw_request("POST", "/v1/graph/branch", body={"from_branch": "main"})

# …load and validate on `review`…

lbb.merge_branch({"from_branch": "review"}, idempotency_key="merge-review-1")
lbb.graph("main", branch="review").delete_branch(confirm="review")

The final live branch cannot be deleted.

Published maintenance and deletion

Writes enqueue complete published-generation maintenance automatically. Clients observe its watermark and lag rather than submitting builds or index garbage collection.

published = lbb.read_snapshot_model()
print(published.snapshot.served_at_seq, published.generation_lag)

lbb.graph("main").delete(confirm="main")  # whole graph, every branch

# Fork the whole graph into a new one (durable job; poll the destination).
fork = lbb.fork_graph("main", "main-copy")  # confirm is pinned to the dst

# Declarative full-state replace: make the graph match this dataset exactly.
# dry_run=True previews the delta with zero durable changes; the response's
# prior_commit_seq / prior_snapshot_token are the rollback anchor.
preview = lbb.reload(records, confirm="main", dry_run=True)

Deletion first retires the graph’s publication epoch, then cancels its active maintenance and removes the old head last. Concurrent retries are idempotent; an old retry cannot cross into a same-name recreated graph.

fork_graph is create-only and safe to retry. reload performs one atomic cutover and keeps prior state queryable through ?as_of_commit_seq=.

Time travel & lineage

state = lbb.current_state({"entity": {"entity_type": "DATABASE", "name": "user-db"}})
timeline = lbb.history({"entity": {"entity_type": "SERVICE", "name": "auth-service"}})
support = lbb.why({...})   # which observations support this statement

Read consistency & read-your-writes

Reads default to eventual, which serves the immutable F3 base at its explicit watermark. Pass consistency="strong" to apply the branch head’s exact bounded delta suffix, or set a client-level default with LbbClient(..., default_consistency="strong").

To read back a fact you wrote a moment ago, use strong consistency and the min_indexed_seq floor. The accepted commit already carries its queryable F3 delta, so a pipeline does not poll publication before reading it.

commit_seq = lbb.commit({"triplets": triplets})["commit_seq"]
rows = lbb.sparql(
    "SELECT ?s ?p ?o WHERE { ?s ?p ?o }",
    min_indexed_seq=commit_seq,
)

min_indexed_seq and consistency are accepted on sparql, sparql_select, query.structured, query.sparql, and summary.

Async client

The async client mirrors every method:

from lbb import AsyncLbbClient

async with AsyncLbbClient("https://0abc1def--production.db.eu.littlebigbrain.com", api_key="lbb_sk_live_...") as lbb:
    state = await lbb.current_state({"entity": {"entity_type": "SERVICE", "name": "auth-service"}})
    sample = await lbb.entities.sample(type="SERVICE", limit=20)

Errors, retries & typed responses

Regular methods return parsed JSON dictionaries and raise LbbError with status_code, type, code, param, request_id, and doc_url on a non-2xx response. Use raw_request(...) when you need response metadata.

The default per-attempt timeout is 120 seconds. Safe reads and idempotency-keyed writes retry 429, 5xx, and transport failures with full-jitter exponential backoff, bounded by a retry budget (retry_budget_ms, default 60s) rather than a fixed attempt count. max_retries (default 6) is a secondary cap. Retries honor Retry-After, and an error the server marks non-retryable is raised immediately. NDJSON and RDF imports get generated idempotency keys automatically. raw_request(..., options={...}) overrides max_retries, retry_budget_ms, timeout, or headers for one call and reports attempts, retry_count, and elapsed_ms. Typed ontology and query reads accept the same options= keyword and automatically classify read-only POSTs as retry-safe. Pass on_retry to the constructor for an absorbed-retry callback, or native httpx event_hooks for instrumentation.

The dict-returning methods are stable for scripts and notebooks. For application code that needs generated Pydantic models and IDE autocompletion, use the matching *_model or *_page helper. Generated request and response models are available under lbb.models.

from lbb.models import EntityTypeSampleResponse, GraphSummaryResponse

summary: GraphSummaryResponse = lbb.summary_model()

entities: EntityTypeSampleResponse = lbb.entities.sample(type="SERVICE", limit=20)
for row in entities.entities:
    print(row.entity.name)

raw = lbb.raw_request("GET", "/v1/graph/summary")
typed = raw.model(GraphSummaryResponse)

Typed helpers cover the high-use surfaces: commit_model, commit_dry_run_model, graph("main").facts.create_model, graph("main").retract_model, summary_model, metadata_model, list_graphs_model, ontology_view_model, ontology_conformance_model, sparql_select_model, entities.sample, and entities.filter_by_attributes_model. ontology.evolve returns a generated response model directly. Async clients expose the same helpers as awaitables.

Every high-level helper that validates a generated response declares that exact Pydantic model as its return annotation. Raw dictionary access remains explicit on raw_request and the untyped convenience methods, and a contract test fails if a typed helper returns Any.

Method surface

create_graph; fork_graph; reload; graph("main").facts.import_rdf; graph("main").facts.create; commit; commit_dry_run; import_ndjson; submit_import_ndjson; wait_for_import_job; merge_branch; delete_branch; retract; entities.sample; entities.filter_by_attributes; sparql; sparql_select; query.structured; query.sparql; query.analytics; query.conflicts; current_state; history; why; ontology.view; ontology.conformance; ontology.define; ontology.evolve; the ontology draft methods; schema.view; schema.publish; read_snapshot_model; schema_summary / schema_summary_model; status; metadata; and summary.

client.governed_conflicts(request) (also client.query.conflicts) requires a visibility filter, applies it before grouping, and returns only distinct-value conflicts with bounded evidence IDs and snapshot and truncation metadata.

Note

The native SPARQL 1.1 Protocol is served at /sparql for standard clients such as YASGUI, Protégé, and RDFLib’s SPARQLWrapper. client.sparql() returns parsed JSON rows for in-process SELECT/ASK use; CONSTRUCT and DESCRIBE return RDF through the native endpoint.