Skip to content

Python SDK (telha)

Synchronous REST client mirroring @telha/sdk, built on httpx with pydantic models. Source: sdk/python/ (distribution name telha-sdk, import name telha). Requires Python >= 3.12.

Install

Not published to PyPI yet. Install from a checkout:

pip install -e /path/to/telha-v2/sdk/python

(The connector and format workers in workers/ consume it the same way, via the workspace.)

Client construction

REST only; there is no Python gRPC transport today.

from telha import TelhaClient

client = TelhaClient("http://127.0.0.1:7625", api_key)
# optional: retries=2, backoff_ms=100

# context manager closes the connection pool
with TelhaClient(url, api_key) as client:
    ...

Records and relationships

# create (list in, list of RecordVersion out); per-record keys like
# "id" (client-supplied logical id) and "upsert" pass through verbatim
[risk] = client.create_records([
    {"labels": ["RISK"],
     "properties": {"title": "Supplier delay", "severity": 7},
     "relationships": [{"type": "OWNED_BY", "node": person_id}]},
])

current = client.get_record(risk.id)              # RecordVersion
page = client.get_history(risk.id, limit=50)      # HistoryPage(.versions, .next_cursor)
newer = client.update_record(risk.id, {           # full-snapshot update (new version)
    "labels": ["RISK"],
    "properties": {"title": "Supplier delay", "severity": 8},
})
receipt = client.delete_record(risk.id)           # CascadeReceipt(.edges_tombstoned, ...)

# one edge per call (the REST surface's shape); loops for you
edges = client.create_relationships([
    {"from": risk.id, "to": person_id, "type": "OWNED_BY"},
])

# connector lookup (PR-065): resolve a source item to its projection root
info = client.lookup_record(external_id="site,item-42", connector="sharepoint")
# {"externalId", "connector", "sourceId", "sourceVersionTx", "rootNodeId", "aclJson", "record"}

Upsert for idempotent syncs (create / identical is a no-write / different appends a version / tombstoned is 409, never resurrected):

client.create_records([{"labels": ["PERSON"], "id": person_id, "upsert": True,
                        "properties": {"display_name": "Kim"}}])

Query, snapshot, schema

result = client.query({
    "find": "RISK",
    "where": {"severity": {"$gte": 7}, "status": {"$eq": "open"}},
    "expand": [{"type": "OWNED_BY", "direction": "out", "depth": 1}],
    "limit": 50,
})
# QueryResult: .records, .related, .next_cursor, .trace_id, .stats

as_of = client.snapshot({"find": "RISK", "atValidTime": "2026-06-01T00:00:00Z"})
schemas = client.schema(label="RISK")   # list of inferred schema dicts

Compare (temporal delta)

compare() wraps POST /v1/compare (PR-053): what changed between two bitemporal coordinates. validTime is required per coordinate; txTime ("as known at") defaults to now on the server.

delta = client.compare({
    "baseline": {"validTime": "2026-01-01T00:00:00Z"},
    "comparison": {"validTime": "2026-04-01T00:00:00Z"},
    "find": "RISK",                       # optional label scope
    "where": {"severity": {"$gte": 7}},   # optional, comparison-side only
    "limit": 100,
})
# DeltaPage:
#   .added     - CompareAdded: full comparison-side .properties
#   .removed   - CompareRemoved: .reason is "tombstone" | "validityLapse"
#   .modified  - CompareModified: .versions (baseline/comparison
#                coordinates), .props.added/.removed/.changed
#                ([{key, old, new}]), .labels_added/.labels_removed,
#                .touch (re-versioned with identical content)
#   .stats     - counts incl. unchanged (counted, never emitted),
#                .rows_scanned, .partial, .edges_skipped_by_filter
#   .next_cursor - paginate large deltas

while delta.next_cursor:
    delta = client.compare({..., "cursor": delta.next_cursor})

where filters the comparison-side selection only - swap the coordinates and invert added/removed to filter the baseline side. With any find/where set, edge deltas are skipped (stats.edges_skipped_by_filter). Invalid requests raise ValidationError with code COMPARE_INVALID and a JSON pointer in the message.

Predicates are operator objects ({"$eq": ...} etc.), never bare scalars. The living grammar reference is core-engine/tests/fixtures/query_corpus.json - every valid and invalid shape with its expected JSON-pointer error, pinned by core tests.

Ingest

accepted = client.ingest_text(
    format="json", name="risks.json",
    content='[{"label": "RISK", "title": "Supplier delay", "severity": 7}]',
)
# IngestAccepted: .deduplicated, .source_id, .job_id (None for the
# synchronous json path), .status_url

accepted = client.ingest_bytes(
    format="docling", name="report.pdf", content=pdf_bytes,
    options={"sourceMeta": {...}},   # connector envelope passes through verbatim
)
status = client.ingest_status(accepted.job_id)   # JobStatus: .state, .attempts, .events

ingest_text/ingest_bytes accept options for format-specific settings (JSON nesting policy, tabular header hints, crawl bounds) and the connector sourceMeta envelope (see connectors).

Not in this SDK yet (honest gaps)

  • No wrappers for /v1/generate, /v1/trace/:id, or the clarifications surface. Use plain httpx/requests against those endpoints (shapes in the quickstart).

Errors

telha.errors mirrors the TS taxonomy; everything extends TelhaError(message, code, status, request_id):

  • QueryInvalidError - 400 QUERY_INVALID, message carries the JSON pointer
  • ValidationError - other 4xx
  • AuthError - 401
  • NotFoundError - 404
  • ConflictError - 409 (TOMBSTONED, TYPE_COLLISION, ...)
  • UnimplementedError - 501
  • ServerError - 5xx and transport failures (status 0, code TRANSPORT)

Retry semantics

Identical policy to the TS SDK: only idempotent calls (get_record, get_history, lookup_record, query, snapshot, compare, schema, ingest_status) retry, only on 5xx or transport errors, with jittered exponential backoff (backoff_ms * 2^(attempt-1) * (0.5 + random)), default 2 extra attempts. Writes are never retried. The underlying httpx.Client timeout is 30 s.

See also