OpenTelemetry Integration

OpenTelemetry tracing and metrics.

Prerequisites

uv add mamba-agents[otel]

Quick Example

from mamba_agents.observability import get_otel_integration

otel = get_otel_integration()

# Initialize
if otel.initialize():
    print("OpenTelemetry initialized")

# Trace operations
with otel.trace_agent_run(prompt, model="gpt-4o") as span:
    result = await agent.run(prompt)
    span.set_attribute("tokens", result.usage().total_tokens)

Exporters

Jaeger

from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

otel.add_span_processor(BatchSpanProcessor(
    JaegerExporter(agent_host_name="localhost", agent_port=6831)
))

OTLP

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

otel.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint="https://otlp.example.com:4317")
))

Configuration

from mamba_agents.config import ObservabilityConfig

config = ObservabilityConfig(
    enable_tracing=True,
    service_name="my-agent",
)

API Reference

OTelIntegration

OTelIntegration(service_name: str = 'mamba-agents')

OpenTelemetry integration for the agent.

Provides hooks for tracing agent operations when OpenTelemetry is available.

Initialize OpenTelemetry integration.

PARAMETER	DESCRIPTION
service_name	Service name for tracing. TYPE: str DEFAULT: 'mamba-agents'

Source code in src/mamba_agents/observability/otel.py

def __init__(self, service_name: str = "mamba-agents") -> None:
    """Initialize OpenTelemetry integration.

    Args:
        service_name: Service name for tracing.
    """
    self._service_name = service_name
    self._tracer: TracerProtocol | NoOpTracer = NoOpTracer()
    self._initialized = False

is_initialized property

is_initialized: bool

Check if OpenTelemetry is initialized.

initialize

initialize() -> bool

Initialize OpenTelemetry if available.

RETURNS	DESCRIPTION
bool	True if OpenTelemetry was initialized.

Source code in src/mamba_agents/observability/otel.py

def initialize(self) -> bool:
    """Initialize OpenTelemetry if available.

    Returns:
        True if OpenTelemetry was initialized.
    """
    try:
        from opentelemetry import trace
        from opentelemetry.sdk.trace import TracerProvider

        provider = TracerProvider()
        trace.set_tracer_provider(provider)
        self._tracer = trace.get_tracer(self._service_name)
        self._initialized = True
        return True
    except ImportError:
        self._initialized = False
        return False

trace_agent_run

trace_agent_run(
    prompt: str, model: str | None = None
) -> Generator[SpanProtocol | NoOpSpan, None, None]

Create a span for an agent run.

PARAMETER	DESCRIPTION
prompt	The user prompt. TYPE: str
model	Model being used. TYPE: str | None DEFAULT: None

YIELDS	DESCRIPTION
SpanProtocol | NoOpSpan	Span for the agent run.

Source code in src/mamba_agents/observability/otel.py

def trace_agent_run(
    self,
    prompt: str,
    model: str | None = None,
) -> Generator[SpanProtocol | NoOpSpan, None, None]:
    """Create a span for an agent run.

    Args:
        prompt: The user prompt.
        model: Model being used.

    Yields:
        Span for the agent run.
    """
    span = self._tracer.start_span("agent.run")
    try:
        span.set_attribute("agent.prompt_length", len(prompt))
        if model:
            span.set_attribute("agent.model", model)
        yield span
    finally:
        span.end()

trace_tool_call

trace_tool_call(
    tool_name: str, args: dict[str, Any] | None = None
) -> Generator[SpanProtocol | NoOpSpan, None, None]

Create a span for a tool call.

PARAMETER	DESCRIPTION
tool_name	Name of the tool. TYPE: str
args	Tool arguments. TYPE: dict[str, Any] | None DEFAULT: None

YIELDS	DESCRIPTION
SpanProtocol | NoOpSpan	Span for the tool call.

Source code in src/mamba_agents/observability/otel.py

def trace_tool_call(
    self,
    tool_name: str,
    args: dict[str, Any] | None = None,
) -> Generator[SpanProtocol | NoOpSpan, None, None]:
    """Create a span for a tool call.

    Args:
        tool_name: Name of the tool.
        args: Tool arguments.

    Yields:
        Span for the tool call.
    """
    span = self._tracer.start_span(f"tool.{tool_name}")
    try:
        span.set_attribute("tool.name", tool_name)
        if args:
            span.set_attribute("tool.arg_count", len(args))
        yield span
    finally:
        span.end()

trace_model_request

trace_model_request(
    model: str, token_count: int | None = None
) -> Generator[SpanProtocol | NoOpSpan, None, None]

Create a span for a model request.

PARAMETER	DESCRIPTION
model	Model name. TYPE: str
token_count	Input token count. TYPE: int | None DEFAULT: None

YIELDS	DESCRIPTION
SpanProtocol | NoOpSpan	Span for the model request.

Source code in src/mamba_agents/observability/otel.py

def trace_model_request(
    self,
    model: str,
    token_count: int | None = None,
) -> Generator[SpanProtocol | NoOpSpan, None, None]:
    """Create a span for a model request.

    Args:
        model: Model name.
        token_count: Input token count.

    Yields:
        Span for the model request.
    """
    span = self._tracer.start_span("model.request")
    try:
        span.set_attribute("model.name", model)
        if token_count:
            span.set_attribute("model.input_tokens", token_count)
        yield span
    finally:
        span.end()

record_usage

record_usage(
    input_tokens: int,
    output_tokens: int,
    model: str | None = None,
) -> None

Record token usage as metrics.

PARAMETER	DESCRIPTION
input_tokens	Number of input tokens. TYPE: int
output_tokens	Number of output tokens. TYPE: int
model	Model used. TYPE: str | None DEFAULT: None

Source code in src/mamba_agents/observability/otel.py

def record_usage(
    self,
    input_tokens: int,
    output_tokens: int,
    model: str | None = None,
) -> None:
    """Record token usage as metrics.

    Args:
        input_tokens: Number of input tokens.
        output_tokens: Number of output tokens.
        model: Model used.
    """
    if not self._initialized:
        return

    try:
        from opentelemetry import metrics

        meter = metrics.get_meter(self._service_name)

        input_counter = meter.create_counter(
            "agent.tokens.input",
            description="Input tokens used",
        )
        output_counter = meter.create_counter(
            "agent.tokens.output",
            description="Output tokens used",
        )

        attributes = {"model": model} if model else {}
        input_counter.add(input_tokens, attributes)
        output_counter.add(output_tokens, attributes)
    except ImportError:
        pass

get_otel_integration

get_otel_integration() -> OTelIntegration

Get the global OpenTelemetry integration.

RETURNS	DESCRIPTION
OTelIntegration	OTelIntegration instance.

Source code in src/mamba_agents/observability/otel.py

def get_otel_integration() -> OTelIntegration:
    """Get the global OpenTelemetry integration.

    Returns:
        OTelIntegration instance.
    """
    global _otel_integration
    if _otel_integration is None:
        _otel_integration = OTelIntegration()
    return _otel_integration