Tracing API Reference

Optional OpenTelemetry tracing integration (istos[otel]).

OpenTelemetry tracing integration (optional dependency).

TracingMiddleware

Creates an OpenTelemetry span per request, linked across hops.

Continues the caller's trace when the request carried a W3C traceparent (see :class:~istos.context.RequestEnvelope), and writes this hop's active traceparent back onto the request context so any outbound query/publish propagates it to the next hop — end-to-end distributed tracing over Zenoh.

Source code in src/istos/observability/tracing.py

class TracingMiddleware:
    """Creates an OpenTelemetry span per request, linked across hops.

    Continues the caller's trace when the request carried a W3C ``traceparent``
    (see :class:`~istos.context.RequestEnvelope`), and writes this hop's active
    ``traceparent`` back onto the request context so any outbound query/publish
    propagates it to the next hop — end-to-end distributed tracing over Zenoh.
    """

    async def __call__(
        self,
        scope: RequestScope,
        call_next: HandlerCallable,
    ) -> Any:
        if _tracer is None:
            return await call_next(scope)

        from opentelemetry.trace.propagation.tracecontext import (
            TraceContextTextMapPropagator,
        )

        from istos.context import get_request_context

        propagator = TraceContextTextMapPropagator()
        incoming = scope.context.traceparent
        parent = propagator.extract({"traceparent": incoming}) if incoming else None

        with _tracer.start_as_current_span(
            f"istos.{scope.operation}",
            context=parent,
            attributes={
                "istos.prefix": scope.prefix,
                "istos.operation": scope.operation,
                "istos.correlation_id": scope.context.correlation_id,
            },
        ):
            # Publish this span as the traceparent for downstream hops.
            carrier: dict[str, str] = {}
            propagator.inject(carrier)
            tp = carrier.get("traceparent")
            if tp:
                scope.context.traceparent = tp
                get_request_context().traceparent = tp
            return await call_next(scope)

configure_tracing(service_name='istos', endpoint=None)

Configure OpenTelemetry tracing. Returns True if configured, False if opentelemetry is not installed.

Source code in src/istos/observability/tracing.py

def configure_tracing(
    service_name: str = "istos",
    endpoint: Optional[str] = None,
) -> bool:
    """
    Configure OpenTelemetry tracing. Returns True if configured, False if
    opentelemetry is not installed.
    """
    global _tracer
    try:
        from opentelemetry import trace
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.resources import Resource
        from opentelemetry.sdk.trace.export import BatchSpanProcessor
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
    except ImportError:
        return False

    resource = Resource.create({"service.name": service_name})
    provider = TracerProvider(resource=resource)
    if endpoint:
        exporter = OTLPSpanExporter(endpoint=endpoint)
        provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    _tracer = trace.get_tracer("istos")
    return True

set_span_attributes(sp, attributes)

Set attributes on sp, skipping None values. Safe when sp is None (tracing off).

Source code in src/istos/observability/tracing.py

def set_span_attributes(sp: Any, attributes: Dict[str, Any]) -> None:
    """Set attributes on ``sp``, skipping ``None`` values. Safe when ``sp`` is
    ``None`` (tracing off)."""
    if sp is None:
        return
    for key, value in _clean(attributes).items():
        sp.set_attribute(key, value)

span(name, attributes=None)

Start a child span when tracing is configured; a no-op otherwise.

Nests under the active request span (the handler's :class:TracingMiddleware span), so agent model/tool work shows up inside the hop that drove it.

Source code in src/istos/observability/tracing.py

@contextlib.contextmanager
def span(name: str, attributes: Optional[Dict[str, Any]] = None) -> Iterator[Any]:
    """Start a child span when tracing is configured; a no-op otherwise.

    Nests under the active request span (the handler's :class:`TracingMiddleware`
    span), so agent model/tool work shows up inside the hop that drove it.
    """
    if _tracer is None:
        yield None
        return
    with _tracer.start_as_current_span(name, attributes=_clean(attributes)) as sp:
        yield sp