cubepi.tracing

tracing_context

function

tracing_context(*, tags: list[str] | tuple[str, ...] | None = None, metadata: dict[str, Any] | None = None) -> Iterator[None]

Scope tags / metadata onto agent runs started inside this block.

The recorder reads these contextvars on AgentStartEvent and stamps them on the invoke_agent span as:

• cubepi.tags — tuple of strings (OTel attribute type)
• one attribute per metadata key, namespaced under cubepi.metadata.* (e.g. metadata=\{"user_id": "u-42"\} → cubepi.metadata.user_id = "u-42"). The dedicated sub-namespace keeps recorder-owned schema keys (cubepi.run_id, cubepi.turn.index, …) safe from caller-supplied collisions.

The contextvar nature means this works for concurrent agents: each asyncio task tree gets its own value. Nested blocks merge additively (tags concatenate; metadata is union with inner keys winning).

Args

• tags — Tags to apply to runs started in this scope. Stored as a tuple on the span so it round-trips through OTel's attribute serializer.
• metadata — Per-run key/value pairs. Values must be types that OTel attributes accept (str, bool, int, float, or a tuple/list of those); other shapes will be silently dropped by the recorder.

source

JsonlSpanExporter

class

JsonlSpanExporter(self, directory: str | os.PathLike[str] = './cubepi-traces')

Write each ReadableSpan as one JSON line.

Files: <directory>/<YYYY-MM-DD>/<run_id>.jsonl. The date used for the subdirectory is the span's start time (UTC). The run_id comes from the cubepi.run_id attribute set by the cubepi Recorder; spans without that attribute fall back to "unknown-run".

Permissions: files are created mode 0o600 (user-only).

source

Meter

class

Meter(self, *, resource: Resource | None = None, exporters: list[MetricExporter] | None = None, export_interval_millis: int = 60000)

Emit OTel GenAI histograms alongside the cubepi Tracer.

Construct with a list of :class:MetricExporter instances (e.g. the OTLP metric exporter); call :meth:attach to start emitting.

Example::

from cubepi.tracing import Tracer, Meter from opentelemetry.exporter.otlp.proto.http.metric_exporter import ( OTLPMetricExporter, )

tracer = Tracer(service_name="my-bot", exporters=[...]) meter = Meter( resource=tracer.resource, exporters=[OTLPMetricExporter(endpoint="http://collector:4318/v1/metrics")], ) tracer.attach(agent) meter.attach(agent)

source

SCHEMA_URL

attribute

source

Tracer

class

Tracer(self, *, service_name: str | None = None, service_version: str | None = None, service_namespace: str | None = None, service_instance_id: str | None = None, deployment_environment: str | None = None, agent_name: str | None = None, agent_id: str | None = None, agent_description: str | None = None, agent_version: str | None = None, exporters: list[SpanExporter] | None = None, record_content: bool = False, redact: 'Callable[[str, Any], Any] | None' = None, resource: Resource | None = None, atexit_flush: bool = True, atexit_flush_timeout_seconds: float = 5.0)

Attaches OTel-compatible tracing to a cubepi :class:Agent.

Construct once per process (or per service). Each attach(agent) call wires the cubepi recorder to the agent's event stream and provider listener registry; the returned callable detaches.

Example

::

from cubepi.tracing import Tracer from cubepi.tracing.exporters import JsonlSpanExporter

tracer = Tracer( service_name="my-bot", agent_name="coding-agent", exporters=[JsonlSpanExporter(directory="./cubepi-traces")], ) detach = tracer.attach(agent) try: ... finally: detach() await tracer.shutdown()

source