Skip to content

Tracing

tracing

Observability — trace context with structured logging, timing, and cost attribution.

Integrates with ADK events: call TraceContext.record_adk_event() to capture token usage from Event.usage_metadata emitted by the ADK runtime.

ToolCallRecord dataclass 

ToolCallRecord(
    tool_name: str,
    agent_name: str,
    input_summary: str,
    output_summary: str,
    duration_ms: float,
    success: bool,
    error: str | None = None,
)

Record of a single tool invocation.

AgentSpan dataclass 

AgentSpan(
    agent_name: str,
    start_time: float = monotonic(),
    end_time: float | None = None,
    input_tokens: int = 0,
    output_tokens: int = 0,
    tool_calls: list[ToolCallRecord] = list(),
)

Timing and cost span for a single agent invocation.

TraceContext 

TraceContext(trace_id: str | None = None)

Trace context that flows through all agent hops in a request.

Source code in libs/ninja-agents/src/ninja_agents/tracing.py 
81
82
83
84
def __init__(self, trace_id: str | None = None) -> None:
    self.trace_id = trace_id or uuid.uuid4().hex
    self.spans: list[AgentSpan] = []
    self._active_spans: dict[str, AgentSpan] = {}

record_adk_event 

record_adk_event(event: Any) -> None

Extract token counts from an ADK Event and attribute them.

If the event has usage_metadata with prompt_token_count / candidates_token_count, the tokens are recorded on the active span for event.author (if one exists).

Source code in libs/ninja-agents/src/ninja_agents/tracing.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def record_adk_event(self, event: Any) -> None:
    """Extract token counts from an ADK ``Event`` and attribute them.

    If the event has ``usage_metadata`` with ``prompt_token_count`` /
    ``candidates_token_count``, the tokens are recorded on the active
    span for ``event.author`` (if one exists).
    """
    author: str | None = getattr(event, "author", None)
    usage = getattr(event, "usage_metadata", None)
    if author and usage:
        span = self._active_spans.get(author)
        if span is None:
            span = self.start_span(author)
        input_tokens = getattr(usage, "prompt_token_count", 0) or 0
        output_tokens = getattr(usage, "candidates_token_count", 0) or 0
        if input_tokens or output_tokens:
            span.record_tokens(input_tokens, output_tokens)