Tracing

Request tracing utilities.

Quick Example

from mamba_agents.observability import RequestTracer

tracer = RequestTracer()

# Start trace
tracer.start_trace()

# Create spans
with tracer.start_span("operation") as span:
    span.set_attribute("model", "gpt-4o")
    span.set_attribute("tokens", 1500)
    # Do work...

# Nested spans
with tracer.start_span("parent") as parent:
    with tracer.start_span("child") as child:
        child.set_attribute("step", 1)

# End trace
trace = tracer.end_trace()
print(f"Duration: {trace.duration_seconds}s")

Trace Data

trace = tracer.end_trace()

print(f"Trace ID: {trace.trace_id}")
print(f"Duration: {trace.duration_seconds}s")

for span in trace.spans:
    print(f"{span.name}: {span.duration_ms}ms")
    for key, value in span.attributes.items():
        print(f"  {key}: {value}")

API Reference

RequestTracer

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

Tracer for tracking requests through the agent.

Initialize the tracer.

PARAMETER	DESCRIPTION
service_name	Name of the service being traced. TYPE: str DEFAULT: 'mamba-agents'

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

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

    Args:
        service_name: Name of the service being traced.
    """
    self._service_name = service_name
    self._context: TraceContext | None = None
    self._current_span_id: str | None = None

trace_id property

trace_id: str | None

Get current trace ID.

current_span_id property

current_span_id: str | None

Get current span ID.

start_trace

start_trace(
    attributes: dict[str, Any] | None = None,
) -> TraceContext

Start a new trace.

PARAMETER	DESCRIPTION
attributes	Initial trace attributes. TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
TraceContext	New TraceContext.

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

def start_trace(self, attributes: dict[str, Any] | None = None) -> TraceContext:
    """Start a new trace.

    Args:
        attributes: Initial trace attributes.

    Returns:
        New TraceContext.
    """
    self._context = TraceContext(
        trace_id=uuid.uuid4().hex,
        attributes=attributes or {},
    )
    _current_trace.set(self._context)
    return self._context

end_trace

end_trace() -> TraceContext | None

End the current trace.

RETURNS	DESCRIPTION
TraceContext | None	Completed TraceContext or None.

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

def end_trace(self) -> TraceContext | None:
    """End the current trace.

    Returns:
        Completed TraceContext or None.
    """
    if self._context:
        self._context.end_time = time.time()
        ctx = self._context
        self._context = None
        self._current_span_id = None
        _current_trace.set(None)
        return ctx
    return None

start_span

start_span(name: str) -> Span

Start a new span.

PARAMETER	DESCRIPTION
name	Span name. TYPE: str

RETURNS	DESCRIPTION
Span	New Span.

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

def start_span(self, name: str) -> Span:
    """Start a new span.

    Args:
        name: Span name.

    Returns:
        New Span.
    """
    if self._context is None:
        self.start_trace()

    span = Span(self, name, parent_id=self._current_span_id)
    self._current_span_id = span._data.span_id
    return span

get_trace_context

get_trace_context() -> TraceContext | None

Get current trace context.

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

def get_trace_context(self) -> TraceContext | None:
    """Get current trace context."""
    return self._context

get_trace_header

get_trace_header() -> dict[str, str]

Get trace context as headers for propagation.

RETURNS	DESCRIPTION
dict[str, str]	Dictionary with traceparent header.

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

def get_trace_header(self) -> dict[str, str]:
    """Get trace context as headers for propagation.

    Returns:
        Dictionary with traceparent header.
    """
    if not self._context:
        return {}

    span_id = self._current_span_id or "0000000000000000"
    traceparent = f"00-{self._context.trace_id}-{span_id}-01"
    return {"traceparent": traceparent}

from_header classmethod

from_header(traceparent: str) -> RequestTracer

Create tracer from incoming traceparent header.

PARAMETER	DESCRIPTION
traceparent	W3C traceparent header value. TYPE: str

RETURNS	DESCRIPTION
RequestTracer	RequestTracer with context restored.

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

@classmethod
def from_header(cls, traceparent: str) -> RequestTracer:
    """Create tracer from incoming traceparent header.

    Args:
        traceparent: W3C traceparent header value.

    Returns:
        RequestTracer with context restored.
    """
    tracer = cls()

    try:
        parts = traceparent.split("-")
        if len(parts) >= 3:
            trace_id = parts[1]
            parent_span_id = parts[2]

            tracer._context = TraceContext(trace_id=trace_id)
            tracer._current_span_id = parent_span_id
            _current_trace.set(tracer._context)
    except (ValueError, IndexError):
        pass

    return tracer

Span

Span(
    tracer: RequestTracer,
    name: str,
    parent_id: str | None = None,
)

A span representing a unit of work within a trace.

Initialize the span.

PARAMETER	DESCRIPTION
tracer	Parent tracer. TYPE: RequestTracer
name	Span name. TYPE: str
parent_id	Parent span ID if nested. TYPE: str | None DEFAULT: None

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

def __init__(
    self,
    tracer: RequestTracer,
    name: str,
    parent_id: str | None = None,
) -> None:
    """Initialize the span.

    Args:
        tracer: Parent tracer.
        name: Span name.
        parent_id: Parent span ID if nested.
    """
    self._tracer = tracer
    self._data = SpanData(
        name=name,
        span_id=uuid.uuid4().hex[:16],
        parent_id=parent_id,
        start_time=time.time(),
    )

set_attribute

set_attribute(key: str, value: Any) -> None

Set a span attribute.

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

def set_attribute(self, key: str, value: Any) -> None:
    """Set a span attribute."""
    self._data.attributes[key] = value

add_event

add_event(
    name: str, attributes: dict[str, Any] | None = None
) -> None

Add an event to the span.

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

def add_event(self, name: str, attributes: dict[str, Any] | None = None) -> None:
    """Add an event to the span."""
    self._data.events.append(
        {
            "name": name,
            "timestamp": time.time(),
            "attributes": attributes or {},
        }
    )

set_error

set_error(error: str) -> None

Mark span as errored.

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

def set_error(self, error: str) -> None:
    """Mark span as errored."""
    self._data.status = "error"
    self._data.error = error

end

end() -> None

End the span.

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

def end(self) -> None:
    """End the span."""
    self._data.end_time = time.time()
    self._tracer._add_span(self._data)