Skip to content

reflexr.otel

The otel extra. See Observability.

OpenTelemetry for applications, in one call: the [otel] extra.

This package is an adapter (ADR-0025). reflexr records through the OpenTelemetry API, its port; this package wires the SDK behind it, with OTLP exporters, the open instrumentations and reflexr's metric cardinality policy::

telemetry = configure_telemetry(service_name="oncall", environment="production")
app = create_app()
telemetry.instrument_app(app)
agent = Agent(..., capabilities=[EventContext(), telemetry.capability()])

An application that runs artifactr too passes artifactr's contribution, and configures telemetry once for both (ADR-0040)::

telemetry = configure_telemetry(artifactr.otel.telemetry(), service_name="app")

Nothing in reflexr requires it, and nothing inside reflexr imports it.

Configuring OpenTelemetry

configure_telemetry 

configure_telemetry(
    *contributions: TelemetryContribution,
    service_name: str,
    service_version: str | None = None,
    environment: str | None = None,
    otlp_endpoint: str | None = None,
    otlp_headers: Mapping[str, str] | None = None,
    instrument: Collection[Instrumented] | None = None,
    engines: Sequence[AsyncEngine | Engine] = (),
    include_content: bool = False,
    metrics_detail: MetricsDetail = "workspace",
    logs: bool = True,
    span_exporter: SpanExporter | None = None,
    metric_reader: MetricReader | None = None,
    log_exporter: LogRecordExporter | None = None,
    span_processors: Sequence[SpanProcessor] = (),
    langfuse: LangfuseMode | None = None,
    langfuse_options: Mapping[str, Any] | None = None,
    set_global: bool = True,
) -> TelemetryHandle

Set up OpenTelemetry for an application: providers, OTLP export and instrumentation.

Call it once, as early as the application starts, before it creates its FastAPI app and agents, with the contributions of the other libraries the application uses, such as artifactr.otel.telemetry(); reflexr's own is always included. artifactr's configure_telemetry takes the same contributions and does the same. It sets up:

  • tracer, meter and logger providers, with service.name, service.version and deployment.environment.name on their resource
  • OTLP over HTTP to otlp_endpoint, or to the endpoint the OTEL_EXPORTER_OTLP_* environment variables name (a local Collector by default)
  • every contribution's views, which apply each library's metric cardinality policy at metrics_detail
  • a baggage processor that copies a run attempt's session, its causal chain, onto every span in it
  • the open instrumentations the contributions advise (FastAPI, SQLAlchemy, httpx and httpx2), or those instrument names, with the stable HTTP semantic conventions (OTEL_SEMCONV_STABILITY_OPT_IN=http, unless it is set already)
  • pydantic-ai's instrumentation settings, which TelemetryHandle.capability wraps for an agent
  • with langfuse, a Langfuse client on the same tracer provider (reflexr.langfuse, the [langfuse] extra), which sends Langfuse whole traces, or only scores

Nothing in reflexr requires it: it is one way to configure the SDK, which reflexr only ever records to through the API.

Parameters:

Name Type Description Default
*contributions TelemetryContribution

Other libraries' contributions (TelemetryContribution), or the application's own. One is kept per name, the first given, and reflexr's own, telemetry, is added unless one is named reflexr.

 ()
service_name str

The service's name, as it appears in every backend.

 required
service_version str | None

The service's version.

 None
environment str | None

The deployment environment, such as production.

 None
otlp_endpoint str | None

The OTLP/HTTP base URL, such as http://collector:4318. Defaults to the OTEL_EXPORTER_OTLP_ENDPOINT environment variable.

 None
otlp_headers Mapping[str, str] | None

Headers for every OTLP request, such as a backend's credentials.

 None
instrument Collection[Instrumented] | None

Which libraries to instrument. Defaults to those the contributions advise that are installed.

 None
engines Sequence[AsyncEngine | Engine]

SQLAlchemy engines that already exist, to trace their queries.

 ()
include_content bool

Whether pydantic-ai records prompts, completions and tool arguments.

 False
metrics_detail MetricsDetail

How much tenancy detail the libraries' metrics keep.

 'workspace'
logs bool

Whether to export the logging module's records through OTLP.

 True
span_exporter SpanExporter | None

Where spans go instead of OTLP, such as an in-memory exporter in tests.

 None
metric_reader MetricReader | None

How metrics are read instead of a periodic OTLP export.

 None
log_exporter LogRecordExporter | None

Where log records go instead of OTLP.

 None
span_processors Sequence[SpanProcessor]

More span processors to add, such as a backend's own.

 ()
langfuse LangfuseMode | None

What the application sends Langfuse itself (LangfuseMode): "traces", with a span filter that keeps a span any contribution keeps, and Langfuse's own LLM spans; or "scores", when the Collector sends Langfuse the traces. Its keys come from the LANGFUSE_* environment variables or langfuse_options.

 None
langfuse_options Mapping[str, Any] | None

Passed to Langfuse(...); a should_export_span here replaces the filter langfuse chooses.

 None
set_global bool

Whether to make the providers the global ones, which reflexr, pydantic-ai and the instrumentations default to.

 True

Returns:

Type Description
TelemetryHandle

A handle that shuts it all down.

TelemetryHandle 

TelemetryHandle(
    *,
    tracer_provider: TracerProvider,
    meter_provider: MeterProvider,
    logger_provider: LoggerProvider | None,
    instrumentation: InstrumentationSettings,
    instrument: Collection[Instrumented] = (),
    engines: Sequence[AsyncEngine | Engine] = (),
    log_handler: Handler | None = None,
    langfuse: Langfuse | None = None,
)

What configure_telemetry set up, and the way to shut it down.

Use it as a context manager, or call shutdown when the application stops.

Parameters:

Name Type Description Default
tracer_provider TracerProvider

The SDK's tracer provider.

 required
meter_provider MeterProvider

The SDK's meter provider.

 required
logger_provider LoggerProvider | None

The SDK's logger provider, if logs are exported.

 required
instrumentation InstrumentationSettings

pydantic-ai's instrumentation settings.

 required
instrument Collection[Instrumented]

The libraries to instrument now.

 ()
engines Sequence[AsyncEngine | Engine]

SQLAlchemy engines to instrument now.

 ()
log_handler Handler | None

A handler to add to the root logger until shutdown.

 None
langfuse Langfuse | None

A Langfuse client to shut down with the rest.

 None

tracer_provider instance-attribute 

tracer_provider = tracer_provider

The SDK's tracer provider.

meter_provider instance-attribute 

meter_provider = meter_provider

The SDK's meter provider.

logger_provider instance-attribute 

logger_provider = logger_provider

The SDK's logger provider, when logs are exported.

instrumentation instance-attribute 

instrumentation = instrumentation

pydantic-ai's instrumentation settings over these providers.

langfuse instance-attribute 

langfuse = langfuse

The Langfuse client, when the application sends Langfuse its traces or its scores.

capability 

capability() -> Instrumentation

Return pydantic-ai's Instrumentation capability, for an agent's capabilities.

instrument_app 

instrument_app(app: FastAPI) -> None

Trace and measure a FastAPI application's requests.

FastAPI's instrumentation only reaches applications whose class it patched first; pass an application here to instrument it whenever it was created.

instrument_engine 

instrument_engine(engine: AsyncEngine | Engine) -> None

Trace a SQLAlchemy engine's queries and measure its connection pool.

create_async_engine is imported before telemetry is configured in most applications, so its engines are not instrumented automatically: pass them here, or to configure_telemetry(engines=...).

shutdown 

shutdown() -> None

Remove the instrumentation, and flush and shut down the providers. Idempotent.

LangfuseMode module-attribute 

LangfuseMode = Literal['traces', 'scores']

What an application sends Langfuse itself.

  • "traces": the traces, filtered by the contributions' span filters, as well as each run's and turn's session, user and tags, and scores
  • "scores": the session, user and tags, set on the spans the Collector sends Langfuse, and scores, but no spans: for a Collector that sends Langfuse every trace already

Libraries' contributions

See ADR-0040.

telemetry 

telemetry() -> Contribution

Return reflexr's contribution, for any library's configure_telemetry.

Its views apply reflexr's metric cardinality policy, its span filter keeps the scopes in reflexr.telemetry.TRACE_SCOPES, and it advises the FastAPI, SQLAlchemy and httpx instrumentations.

TelemetryContribution

Bases: Protocol

What configure_telemetry reads from a library's contribution: the port (ADR-0040).

It reads these four fields and nothing else, so any value that has them will do: reflexr's telemetry, artifactr's artifactr.otel.telemetry(), which neither library imports, or an application's own Contribution for its metrics. Both libraries' MetricsDetail is "workspace" | "tenant" | "none": part of the contract.

name property 

name: str

The library's name, which is its instrumentation scope; one contribution per name.

metric_views property 

metric_views: Callable[[MetricsDetail], Sequence[View]]

Return the views that apply the library's metric cardinality policy at a detail.

should_export_span property 

should_export_span: Callable[[ReadableSpan], bool]

Return whether a span belongs in the library's traces: Langfuse's span filter.

instrument property 

instrument: frozenset[str]

The instrumentations the library advises: those that trace its work.

Names configure_telemetry does not know (see Instrumented) are ignored.

Contribution dataclass 

Contribution(
    name: str,
    metric_views: Callable[[MetricsDetail], Sequence[View]],
    should_export_span: Callable[[ReadableSpan], bool],
    instrument: frozenset[Instrumented],
)

A library's contribution to an application's telemetry.

Parameters:

Name Type Description Default
name str

The library's name, which is its instrumentation scope.

 required
metric_views Callable[[MetricsDetail], Sequence[View]]

Returns the views that keep only the attributes the library's metrics may carry at a level of detail.

 required
should_export_span Callable[[ReadableSpan], bool]

Whether a span belongs in the library's traces.

 required
instrument frozenset[Instrumented]

The instrumentations the library advises.

 required

Metric views

metric_views 

metric_views(
    detail: MetricsDetail = "workspace",
) -> list[View]

Return views that keep only the attributes reflexr's metrics may carry at detail.

Pass them to the SDK's MeterProvider(views=...). With "tenant" the workspace is dropped, and with "none" both tenant and workspace are, before aggregation, so a deployment with many workspaces keeps a bounded number of series.

Instrumentations

Instrumented module-attribute 

Instrumented = Literal[
    "fastapi", "sqlalchemy", "asyncpg", "httpx"
]

A library configure_telemetry can instrument. httpx covers httpx and httpx2.

INSTRUMENTED module-attribute 

INSTRUMENTED: tuple[Instrumented, ...] = (
    "fastapi",
    "sqlalchemy",
    "asyncpg",
    "httpx",
)

Every library configure_telemetry can instrument.

installed 

installed() -> set[Instrumented]

Return the instrumentable libraries that are installed.

httpx counts when either httpx or httpx2 is: one instruments the other's clients too.

BAGGAGE_KEYS module-attribute 

BAGGAGE_KEYS: frozenset[str] = frozenset({SESSION_ID})

The baggage entries copied onto every span: the session, which a run attempt places in baggage.