Telemetry Events

TelemetryEvent is the SDK’s primitive for product telemetry — user actions, tool calls, agent decisions, anything you’d send to a product analytics tool. Events are pushed independently of trajectories, and can be correlated to them via trace_id (see Linking Events to Trajectories).

Quickstart

Telemetry ingestion uses the generated Client; no provider initialization is required.

import trajectory_sdk as tj
from trajectory import Client
from trajectory.lib import ingest_events

result = ingest_events(Client(), [
    tj.TelemetryEvent(
        event_type="user.accept",
        session_id="session-123",
        properties={"surface": "chat", "model": "gpt-4o"},
    ),
])

print(result.ingested, result.skipped, result.failures)

Constructing Events

TelemetryEvent is a frozen dataclass. Only event_type and session_id are required; every other field has a sensible default.

event = tj.TelemetryEvent(
    event_type="tool.call",
    session_id="session-123",
    properties={"tool": "search", "query": "refunds policy"},
    user_id="user-abc",
    trace_id="req_01HXYZ",
    trajectory_id="traj_xyz",
    source="my_app",
    metadata={"app_version": "2.4.0"},
)

Field	Type	Default	Description
`event_type`	`str`	required	Dotted event name (e.g. `user.accept`, `tool.call`).
`session_id`	`str`	required	Session grouping. Defaults to `trace_id` when produced via `TraceContext.event(...)`.
`properties`	`dict`	`{}`	Free-form event payload.
`event_id`	`str`	new UUID4 hex	Idempotency key (see below).
`timestamp`	`str`	now (UTC ISO)	Event time. Pass an explicit value to backdate.
`user_id`	`str \| None`	`None`	End-user identifier.
`trajectory_id`	`str \| None`	`None`	Backend trajectory ID once known.
`trace_id`	`str \| None`	`None`	Caller-owned correlation key. Always set when producing events via `TraceContext`.
`source`	`str`	`"sdk"`	Where the event originated.
`metadata`	`dict \| None`	`None`	Additional context not part of `properties`.

Idempotency

event_id is the idempotency key. The default UUID4 means re-runs of the same code produce different IDs (and the backend will accept both as separate events). For at-least-once delivery pipelines — webhooks, retries, replay from a queue — set event_id deterministically from your own primary key so duplicate pushes are coalesced server-side:

import hashlib

def event_id_for(record_id: str) -> str:
    return hashlib.sha256(record_id.encode()).hexdigest()

event = tj.TelemetryEvent(
    event_type="user.accept",
    session_id="session-123",
    event_id=event_id_for(record["id"]),
)

Re-pushing the same event_id is a no-op on the backend.

Ingesting Events

ingest_events(client, events) validates events locally and ingests valid events in bounded chunks through the generated API client.

from trajectory import Client
from trajectory.lib import ingest_events

result = ingest_events(Client(), events, chunk_size=1000)

Parameter	Type	Default	Description
`client`	`Client`	required	Generated client. Reads `TRAJECTORY_API_KEY` and `TRAJECTORY_BASE_URL` from the environment by default.
`events`	`list[Any]`	required	Dataclass, Pydantic-model, or mapping events to ingest.
`chunk_size`	`int`	`1000`	Max events per HTTP request. The SDK splits longer lists automatically.

chunk_size must be an integer greater than zero. Passing another type raises TypeError; zero or a negative integer raises ValueError. The call returns a TelemetryIngestSummary:

Field	Type	Description
`ingested`	`int`	Events accepted by the backend.
`skipped`	`int`	Total events rejected locally or by the backend.
`failures`	`list[BatchItemFailure]`	Structured local and backend failures with original item indices.

Invalid events do not poison a whole batch. Missing fields, invalid fields, non-mapping events, and values that cannot be serialized as JSON are excluded locally and reported in result.failures; backend failures are appended to the same list.

Producing Events via TraceContext

When you’re already inside a TraceContext (the recommended path for SDK-instrumented agent runs), use trace.event(...) so events automatically inherit the trace_id:

trace = tj.start_trace()

trace.event("tool.call", {"tool": "search"})
trace.event("tool.result", {"ok": True}, user_id="user-abc")

# trace.events is the accumulated buffer
result = ingest_events(Client(), trace.events)

TraceContext.event(...) sets session_id = trace_id automatically. Override either via the keyword arguments if you need a different grouping. See Linking Events to Trajectories for the full workflow, including trajectory.lib.upload_trace() which uploads a trajectory and ingests correlated events in one atomic call.

Linking Events to Trajectories

The trace_id correlation story end-to-end.

API Reference

Full signatures for ingest_events, TelemetryEvent, and TelemetryIngestSummary.

Getting Started

SDK

HTTP API

Platform

Quickstart

Constructing Events

Idempotency

Ingesting Events

Producing Events via TraceContext

Linking Events to Trajectories

API Reference

​Quickstart

​Constructing Events

​Idempotency

​Ingesting Events

​Producing Events via TraceContext

​Related

Linking Events to Trajectories

API Reference

Quickstart

Constructing Events

Idempotency

Ingesting Events

Producing Events via TraceContext

Related