observability.tracer
Tracing utilities for agent-runtimes.
Provides:
- AgentTracer for creating traced agent operations
- Decorators for tracing agent runs and tool calls
- Context propagation for distributed tracing
AgentTracer Objects
class AgentTracer()
Tracer for agent operations.
Wraps OpenTelemetry tracer with agent-specific functionality.
__init__
def __init__(service_name: str = "agent-runtimes",
agent_id: str | None = None)
Initialize the agent tracer.
Arguments:
service_name- Name of the service for tracing.agent_id- Default agent ID to attach to spans.
span
@contextmanager
def span(name: str,
attributes: dict[str, Any] | None = None,
agent_id: str | None = None,
session_id: str | None = None) -> Generator[Any, None, None]
Create a traced span context.
Arguments:
name- Name of the span.attributes- Additional attributes for the span.agent_id- Agent ID to attach to the span.session_id- Session ID to attach to the span.
Yields:
The span object (or None if tracing unavailable).
async_span
@asynccontextmanager
async def async_span(
name: str,
attributes: dict[str, Any] | None = None,
agent_id: str | None = None,
session_id: str | None = None) -> AsyncGenerator[Any, None]
Create an async traced span context.
Arguments:
name- Name of the span.attributes- Additional attributes for the span.agent_id- Agent ID to attach to the span.session_id- Session ID to attach to the span.
Yields:
The span object (or None if tracing unavailable).
trace_run
def trace_run(agent_id: str | None = None,
session_id: str | None = None) -> Callable[[F], F]
Decorator to trace agent run methods.
Arguments:
agent_id- Agent ID to attach to spans.session_id- Session ID to attach to spans.
Returns:
Decorated function.
trace_tool
def trace_tool(tool_name: str | None = None,
agent_id: str | None = None) -> Callable[[F], F]
Decorator to trace tool calls.
Arguments:
tool_name- Name of the tool (uses function name if not provided).agent_id- Agent ID to attach to spans.
Returns:
Decorated function.
trace_code_execution
def trace_code_execution(agent_id: str | None = None,
session_id: str | None = None) -> Callable[[F], F]
Decorator to trace code execution.
Arguments:
agent_id- Agent ID to attach to spans.session_id- Session ID to attach to spans.
Returns:
Decorated function.
get_tracer
def get_tracer(service_name: str = "agent-runtimes",
agent_id: str | None = None) -> AgentTracer
Get or create the default tracer.
Arguments:
service_name- Name of the service for tracing.agent_id- Default agent ID to attach to spans.
Returns:
The AgentTracer instance.
trace_agent_run
def trace_agent_run(agent_id: str | None = None,
session_id: str | None = None) -> Callable[[F], F]
Decorator to trace agent run methods using default tracer.
Arguments:
agent_id- Agent ID to attach to spans.session_id- Session ID to attach to spans.
Returns:
Decorated function.
trace_tool_call
def trace_tool_call(tool_name: str | None = None,
agent_id: str | None = None) -> Callable[[F], F]
Decorator to trace tool calls using default tracer.
Arguments:
tool_name- Name of the tool.agent_id- Agent ID to attach to spans.
Returns:
Decorated function.