nat.plugins.strands.strands_callback_handler

Attributes

logger

Classes

StrandsToolInstrumentationHook	Hook callbacks for instrumenting Strands tool invocations.
StrandsProfilerHandler	Helper class that provides a standard way to create an ABC using

Module Contents

logger

class StrandsToolInstrumentationHook(handler: StrandsProfilerHandler)

Hook callbacks for instrumenting Strands tool invocations.

This class provides callbacks for Strands’ hooks API to capture tool execution events and emit proper TOOL_START/END spans.

Initialize the hook with a reference to the profiler handler.

Args:

handler: StrandsProfilerHandler instance that manages this hook

handler

_tool_start_times: dict[str, float]

_step_manager

on_before_tool_invocation(event: Any) → None

Handle tool invocation start.

Called by Strands before a tool is executed. Emits a TOOL_START span.

Args:

event: BeforeToolInvocationEvent from Strands

on_after_tool_invocation(event: Any) → None

Handle tool invocation end.

Called by Strands after a tool execution completes. Emits a TOOL_END span.

Args:

event: AfterToolInvocationEvent from Strands

_extract_tool_info(

selected_tool: Any,

tool_use: dict,

) → tuple[str, str, dict]

Extract tool name, ID, and input from event.

Args:

selected_tool: The tool being invoked tool_use: Tool use dictionary from Strands event

Returns:

Tuple of (tool_name, tool_use_id, tool_input)

class StrandsProfilerHandler

Bases: nat.profiler.callbacks.base_callback_class.BaseProfilerCallback

Helper class that provides a standard way to create an ABC using inheritance.

_patched: bool = False

last_call_ts

instrument() → None

Instrument Strands for telemetry capture.

This patches: 1. Model streaming methods (OpenAI/Bedrock) for LLM spans 2. Agent.__init__ to auto-register tool hooks on Agent creation

Tool instrumentation uses Strands’ hooks API, which is automatically registered when an Agent is instantiated.

_instrument_agent_init() → None

Patch Agent.__init__ to auto-register hooks on instantiation.

This ensures that whenever a Strands Agent is created, our tool instrumentation hooks are automatically registered without requiring any user code changes.

_extract_model_info(model_instance: Any) → tuple[str, dict[str, Any]]

Extract model name from Strands model instance.

_wrap_stream_method(

original: collections.abc.Callable[Ellipsis, Any],

) → collections.abc.Callable[Ellipsis, Any]

_extract_text_from_event(ev: dict) → str

Extract text content from a Strands event.

Args:

ev: Event dictionary from Strands stream

Returns:

Extracted text content or empty string

_finalize_tool_call(tool_call: dict[str, Any]) → None

Parse the accumulated input_str JSON and store in the input field.

Args:

tool_call: Tool call dictionary with input_str to parse

_extract_tool_call_from_event(ev: dict) → dict[str, Any] | None

Extract tool call information from a Strands event.

Args:

ev: Event dictionary from Strands stream

Returns:

Dictionary with tool call info (name, input_chunk) or None if not a tool call

_extract_usage_from_event(ev: dict) → dict[str, int] | None

Extract usage information from a Strands event.

Args:

ev: Event dictionary from Strands stream

Returns:

Dictionary with token usage info or None if not found