Skip to main content

Python SDK

Planned

The Python SDK is planned. The planned package name is mantika-tessera; it is reserved on PyPI by a 0.0.1 placeholder that is not the SDK. The bare name tessera on PyPI is an unrelated third-party project, so do not install either of them. Send events with HTTPS ingestion — the guide below documents the SDK's contract and its samples run in CI against the repository's own copy.

Ingestion ergonomics (buffering, batching, retries, flush guarantees) plus the full typed API surface, generated from the same OpenAPI document as the API reference. Supported Pythons: 3.10–3.13; the only dependencies are httpx and pydantic.

Authenticate​

export TESSERA_API_KEY="tsk_test_..."

Tessera() reads TESSERA_API_KEY; pass api_key= explicitly if you manage secrets differently. Key hygiene: every key is bound to one project and one environment — develop and run CI on tsk_test_… keys (an isolated data plane that never touches live models); reserve tsk_live_… for production ingestion. Base URLs are env-configurable (base_url=/ingest_url= or TESSERA_BASE_URL/TESSERA_INGEST_URL) for single-tenant deployments — which is also how the samples on this page run against the in-repo test stack in CI.

Track events (buffering and flush())​

01-track-and-flush.py
"""Track events with buffering, flush explicitly, read them back."""

import os
import time

from tessera import Tessera

# The API key comes from TESSERA_API_KEY; base URLs are env-driven so the
# same script runs against any deployment.
client = Tessera(
    base_url=os.environ["TESSERA_API_BASE"],
    ingest_url=os.environ["TESSERA_INGEST_BASE"],
)

# track() never performs network I/O: it validates locally and buffers.
# A background thread flushes at 100 events or every 2 seconds.
for index in range(5):
    client.events.track(
        {"type": "user", "id": "sdk_sample_1"},
        "page.viewed",
        {"path": f"/products/{index}", "referrer": "newsletter"},
    )

# flush() drains the buffer and returns cumulative delivery counters.
result = client.events.flush()
print(f"accepted={result.accepted} rejected={result.rejected} dropped={result.dropped}")

# Read the events back from the entity's timeline.
for _ in range(30):
    timeline = client.api.list_entity_events(entity_type="user", entity_id="sdk_sample_1")
    if len(timeline.data) >= 5:
        break
    time.sleep(1)
print(f"timeline events: {len(timeline.data)}")

client.close()  # close() (and an atexit hook) flushes any tail

Download sdks/python/01-track-and-flush.py — this exact file runs in CI on every change (hosts shown are for this environment).

track() never performs network I/O: it validates the event locally (the same shape rules the platform enforces) and buffers it; a background thread flushes at 100 events or 2 seconds, gzip-compressed, with one Idempotency-Key per batch. Retries follow the platform contract: 429 waits the server's retry_after_seconds, 5xx/connection errors back off exponentially with jitter, other 4xx fail without retry.

Delivery guarantees, precisely:

  • flush() blocks until drained and returns cumulative counters (accepted, rejected, dropped).
  • close() — and an atexit hook — flush pending events on clean exit (bounded by 10 s); the flusher is a daemon thread, so a hung network never prevents interpreter exit.
  • Crash loss is bounded by one buffer: a hard kill loses at most the events since the last flush. Tighten flush_at/flush_interval if you need a smaller bound.
  • If the buffer fills (max_buffer, default 10,000), track() applies bounded backpressure by default; overflow="drop_oldest" prefers losing the oldest events instead. Every undelivered event reaches your on_reject(event, code, message) callback.

Idempotent ingestion​

02-idempotent-ingestion.py
"""Idempotent ingestion: the same event_id can never land twice."""

import os
import time

from tessera import Tessera

client = Tessera(
    base_url=os.environ["TESSERA_API_BASE"],
    ingest_url=os.environ["TESSERA_INGEST_BASE"],
)

event = {
    "entity": {"type": "user", "id": "sdk_sample_2"},
    "event_type": "order.placed",
    "attributes": {"order_value": 19.99},
    "timestamp": "2026-01-15T09:30:00Z",
    "event_id": "3f8a6c1e-9b24-4d75-a0c3-5e1d7b2f4a90",  # your idempotency id
}

# Send it twice — a retry, a replayed queue, a double-clicked backfill.
client.events.track(**event)
client.events.flush()
client.events.track(**event)
result = client.events.flush()
print(f"second send: accepted={result.accepted}")

# The platform dedupes on event_id: exactly one event exists.
for _ in range(30):
    timeline = client.api.list_entity_events(entity_type="user", entity_id="sdk_sample_2")
    if len(timeline.data) >= 1:
        break
    time.sleep(1)
time.sleep(2)  # give a duplicate every chance to appear
timeline = client.api.list_entity_events(entity_type="user", entity_id="sdk_sample_2")
print(f"events with id 3f8a6c1e-9b24-4d75-a0c3-5e1d7b2f4a90: {len(timeline.data)}")
client.close()

Download sdks/python/02-idempotent-ingestion.py — this exact file runs in CI on every change (hosts shown are for this environment).

Supply event_id for any event your producer might retry — replayed queues, at-least-once workers, backfills re-run twice. The platform dedupes on it forever; without one, a derived id makes exact retries no-ops anyway.

Make the first 16 characters of every event id unique: start it with a random or hashed part (for example a UUID or a hash), not a fixed prefix such as a date or an order-number stem. Until the ledger identity upgrade, an event whose first 16 characters match another event of the same entity at the same timestamp is held back, not stored, until the upgrade stores it (spec 002 §8). The first 16 characters must also not contain #: events.track() raises TesseraValidationError for such an id before anything is queued. See choosing event_id values.

Entity ids (the entity's and every secondary entity's) must not contain # anywhere: events.track() raises TesseraValidationError (entity.id: must not contain '#') before anything is queued. See entity ids.

Batch error handling​

03-batch-errors.py
"""Batch error handling: per-item results (003 §5) and the on_reject hook."""

import os

import httpx

from tessera import Tessera
from tessera.errors import TesseraValidationError

# --- 1. the wire contract: per-item results ----------------------------------
# One bad event never poisons the batch: the endpoint answers 207-style with
# an index, code and message per rejected item.
response = httpx.post(
    f"{os.environ['TESSERA_INGEST_BASE']}/events",
    headers={"Authorization": f"Bearer {os.environ['TESSERA_API_KEY']}"},
    json={
        "events": [
            {
                "entity": {"type": "user", "id": "sdk_sample_3"},
                "event_type": "cart.item_added",
                "timestamp": "2026-02-01T10:00:00Z",
                "attributes": {"sku_count": 3},
            },
            {
                "entity": {"type": "user", "id": "sdk_sample_3"},
                "event_type": "cart.item_added",
                "timestamp": "2026-02-01T10:00:01Z",
                # Nested objects are rejected — flatten upstream (002 §2).
                "attributes": {"item": {"sku": "A-1", "qty": 2}},
            },
        ]
    },
)
body = response.json()
print(f"accepted={body['accepted']} rejected={body['rejected']}")
for item in body["results"]:
    print(f"rejected index={item['index']} code={item['code']}")

# --- 2. the SDK catches shape violations before they leave the process --------
client = Tessera(
    base_url=os.environ["TESSERA_API_BASE"],
    ingest_url=os.environ["TESSERA_INGEST_BASE"],
)
try:
    client.events.track(
        {"type": "user", "id": "sdk_sample_3"},
        "cart.item_added",
        {"item": {"sku": "A-1", "qty": 2}},  # nested -> raises, nothing queued
    )
except TesseraValidationError as error:
    print(f"TesseraValidationError: {type(error).__name__} caught locally")

# Server-side rejections (and overflow drops) surface through on_reject and
# the FlushResult counters — see the guide for the callback signature.
client.close()

Download sdks/python/03-batch-errors.py — this exact file runs in CI on every change (hosts shown are for this environment).

Three layers to build against: the SDK raises TesseraValidationError synchronously for shape violations (nothing queued); the wire protocol answers per-item so one bad event never poisons a batch; and post-ingest contract violations park in the dead-letter view, replayable after you fix the cause.

Reading timelines and state​

The samples above read back what they wrote with client.api.list_entity_events(...). The same generated, typed surface covers every operation — one method per OpenAPI operation on client.api, plus handwritten resource groups (client.sources, client.projects, client.usage, …):

state = client.api.get_slot_entity_state(slot="prod", entity_type="user", entity_id="u_42")
print(state.scores, state.freshness.model_fingerprint)

The quickstart shows the SDK form of every lifecycle step, each one executed in CI: uploads, contracts, labels, datasets, runs, slots, and state reads.