Telemetry

Observability and tracing for RoomKit applications. See the Telemetry guide for usage examples.

Core

TelemetryProvider

Bases: ABC

Abstract base class for telemetry providers.

Providers collect span and metric data from RoomKit operations. The default NoopTelemetryProvider has zero overhead.

name abstractmethod property

name

Provider name for identification.

start_span abstractmethod

start_span(kind, name, *, parent_id=None, attributes=None, room_id=None, session_id=None, channel_id=None)

Start a new telemetry span.

Returns:

Type	Description
str	A unique span ID string.

end_span abstractmethod

end_span(span_id, *, status='ok', error_message=None, attributes=None)

End a previously started span.

set_attribute abstractmethod

set_attribute(span_id, key, value)

Set an attribute on an active span.

record_metric abstractmethod

record_metric(name, value, *, unit='', attributes=None)

Record a metric value.

get_span_context

get_span_context(span_id)

Return an opaque context object for the given span.

Used to propagate backend-specific parent context (e.g. OTel Context) through :func:set_current_span for robust parent linking across async boundaries.

Returns None by default. Override in providers that carry backend-specific context (e.g. OpenTelemetryProvider).

flush

flush()

Flush pending spans/metrics without closing the provider.

Called after ending long-lived spans (e.g. VOICE_SESSION) to ensure they are exported promptly rather than waiting for shutdown.

close

close()

Close the provider and flush any pending data.

reset

reset()

Reset internal state (useful for testing).

span

span(kind, name, **kwargs)

Context manager for span lifecycle.

Yields the span ID. Automatically ends the span on exit, recording error status if an exception occurs.

TelemetryConfig dataclass

TelemetryConfig(provider=None, sample_rate=1.0, enabled_spans=None, metadata=dict(), suppressed_hook_triggers=(lambda: {'on_input_audio_level', 'on_output_audio_level', 'on_vad_audio_level'})())

Configuration for telemetry collection.

Attributes:

Name	Type	Description
provider	TelemetryProvider | None	The telemetry provider to use. Defaults to NoopTelemetryProvider if not set.
sample_rate	float	Fraction of spans to record (0.0 to 1.0). Default 1.0 records all spans.
enabled_spans	set[SpanKind] | None	If set, only these span kinds are recorded. None means all span kinds are enabled.

Span dataclass

Span(id=(lambda: hex[:16])(), kind=CUSTOM, name='', parent_id=None, attributes=dict(), start_time=(lambda: now(UTC))(), end_time=None, status='ok', error_message=None, room_id=None, session_id=None, channel_id=None)

Represents a telemetry span.

duration_ms property

duration_ms

Duration in milliseconds, or None if not yet ended.

Attr

Well-known attribute key constants for telemetry spans and metrics.

SpanKind

Bases: StrEnum

Span classifications for telemetry.

Built-in Providers

ConsoleTelemetryProvider

ConsoleTelemetryProvider(*, level=INFO)

Bases: TelemetryProvider

Logs span start/end and metrics to roomkit.telemetry logger.

Zero external dependencies — uses only Python's built-in logging. Useful for development and debugging.

Example::

import logging
logging.basicConfig(level=logging.INFO)

from roomkit import RoomKit
from roomkit.telemetry import ConsoleTelemetryProvider

kit = RoomKit(telemetry=ConsoleTelemetryProvider())

OpenTelemetryProvider

OpenTelemetryProvider(*, tracer_provider=None, meter_provider=None, service_name='roomkit', metadata=None)

Bases: TelemetryProvider

Bridges RoomKit telemetry to OpenTelemetry SDK.

Requires opentelemetry-api and opentelemetry-sdk::

pip install opentelemetry-api opentelemetry-sdk

Example::

from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter

provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))

from roomkit import RoomKit
from roomkit.telemetry.opentelemetry import OpenTelemetryProvider

kit = RoomKit(telemetry=OpenTelemetryProvider(tracer_provider=provider))

get_span_context

get_span_context(span_id)

Return an OTel Context with the given span set as current.

Overrides :meth:TelemetryProvider.get_span_context to return a native OTel Context for robust parent linking across async boundaries.

NoopTelemetryProvider

Bases: TelemetryProvider

Default telemetry provider that does nothing.

All methods are no-ops with zero overhead. This is the default when no telemetry provider is configured.

MockTelemetryProvider

MockTelemetryProvider()

Bases: TelemetryProvider

Records all spans and metrics in lists for test assertions.

Example::

telemetry = MockTelemetryProvider()
kit = RoomKit(telemetry=telemetry)
# ... run operations ...
assert len(telemetry.spans) > 0
stt_spans = telemetry.get_spans(SpanKind.STT_TRANSCRIBE)
assert stt_spans[0].attributes["stt.text_length"] > 0

spans property

spans

All completed spans.

get_spans

get_spans(kind)

Get completed spans of a specific kind.

get_active_spans

get_active_spans()

Get spans that have been started but not ended.

reset

reset()

Clear all recorded spans and metrics.