strands.experimental.steering.context_providers.ledger_provider

Ledger context provider for comprehensive agent activity tracking.

Tracks complete agent activity ledger including tool calls, conversation history, and timing information. This comprehensive audit trail enables steering handlers to make informed guidance decisions based on agent behavior patterns and history.

Data captured:

- Tool call history with inputs, outputs, timing, success/failure
- Conversation messages and agent responses
- Session metadata and timing information
- Error patterns and recovery attempts

Usage

Use as context provider functions or mix into steering handlers.

logger = logging.getLogger(__name__) module-attribute

AfterToolCallEvent dataclass

Bases: HookEvent

Event triggered after a tool invocation completes.

This event is fired after the agent has finished executing a tool, regardless of whether the execution was successful or resulted in an error. Hook providers can use this event for cleanup, logging, or post-processing.

Note: This event uses reverse callback ordering, meaning callbacks registered later will be invoked first during cleanup.

Attributes:

Name	Type	Description
selected_tool	Optional[AgentTool]	The tool that was invoked. It may be None if tool lookup failed.
tool_use	ToolUse	The tool parameters that were passed to the tool invoked.
invocation_state	dict[str, Any]	Keyword arguments that were passed to the tool
result	ToolResult	The result of the tool invocation. Either a ToolResult on success or an Exception if the tool execution failed.
cancel_message	str | None	The cancellation message if the user cancelled the tool call.

Source code in strands/hooks/events.py

@dataclass
class AfterToolCallEvent(HookEvent):
    """Event triggered after a tool invocation completes.

    This event is fired after the agent has finished executing a tool,
    regardless of whether the execution was successful or resulted in an error.
    Hook providers can use this event for cleanup, logging, or post-processing.

    Note: This event uses reverse callback ordering, meaning callbacks registered
    later will be invoked first during cleanup.

    Attributes:
        selected_tool: The tool that was invoked. It may be None if tool lookup failed.
        tool_use: The tool parameters that were passed to the tool invoked.
        invocation_state: Keyword arguments that were passed to the tool
        result: The result of the tool invocation. Either a ToolResult on success
            or an Exception if the tool execution failed.
        cancel_message: The cancellation message if the user cancelled the tool call.
    """

    selected_tool: Optional[AgentTool]
    tool_use: ToolUse
    invocation_state: dict[str, Any]
    result: ToolResult
    exception: Optional[Exception] = None
    cancel_message: str | None = None

    def _can_write(self, name: str) -> bool:
        return name == "result"

    @property
    def should_reverse_callbacks(self) -> bool:
        """True to invoke callbacks in reverse order."""
        return True

should_reverse_callbacks property

True to invoke callbacks in reverse order.

BeforeToolCallEvent dataclass

Bases: HookEvent, _Interruptible

Event triggered before a tool is invoked.

This event is fired just before the agent executes a tool, allowing hook providers to inspect, modify, or replace the tool that will be executed. The selected_tool can be modified by hook callbacks to change which tool gets executed.

Attributes:

Name	Type	Description
selected_tool	Optional[AgentTool]	The tool that will be invoked. Can be modified by hooks to change which tool gets executed. This may be None if tool lookup failed.
tool_use	ToolUse	The tool parameters that will be passed to selected_tool.
invocation_state	dict[str, Any]	Keyword arguments that will be passed to the tool.
cancel_tool	bool | str	A user defined message that when set, will cancel the tool call. The message will be placed into a tool result with an error status. If set to True, Strands will cancel the tool call and use a default cancel message.

Source code in strands/hooks/events.py

@dataclass
class BeforeToolCallEvent(HookEvent, _Interruptible):
    """Event triggered before a tool is invoked.

    This event is fired just before the agent executes a tool, allowing hook
    providers to inspect, modify, or replace the tool that will be executed.
    The selected_tool can be modified by hook callbacks to change which tool
    gets executed.

    Attributes:
        selected_tool: The tool that will be invoked. Can be modified by hooks
            to change which tool gets executed. This may be None if tool lookup failed.
        tool_use: The tool parameters that will be passed to selected_tool.
        invocation_state: Keyword arguments that will be passed to the tool.
        cancel_tool: A user defined message that when set, will cancel the tool call.
            The message will be placed into a tool result with an error status. If set to `True`, Strands will cancel
            the tool call and use a default cancel message.
    """

    selected_tool: Optional[AgentTool]
    tool_use: ToolUse
    invocation_state: dict[str, Any]
    cancel_tool: bool | str = False

    def _can_write(self, name: str) -> bool:
        return name in ["cancel_tool", "selected_tool", "tool_use"]

    @override
    def _interrupt_id(self, name: str) -> str:
        """Unique id for the interrupt.

        Args:
            name: User defined name for the interrupt.

        Returns:
            Interrupt id.
        """
        return f"v1:before_tool_call:{self.tool_use['toolUseId']}:{uuid.uuid5(uuid.NAMESPACE_OID, name)}"

LedgerAfterToolCall

Bases: SteeringContextCallback[AfterToolCallEvent]

Context provider for ledger tracking after tool calls.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

class LedgerAfterToolCall(SteeringContextCallback[AfterToolCallEvent]):
    """Context provider for ledger tracking after tool calls."""

    def __call__(self, event: AfterToolCallEvent, steering_context: SteeringContext, **kwargs: Any) -> None:
        """Update ledger after tool call."""
        ledger = steering_context.data.get("ledger") or {}

        if ledger.get("tool_calls"):
            last_call = ledger["tool_calls"][-1]
            last_call.update(
                {
                    "completion_timestamp": datetime.now().isoformat(),
                    "status": event.result["status"],
                    "result": event.result["content"],
                    "error": str(event.exception) if event.exception else None,
                }
            )
            steering_context.data.set("ledger", ledger)

__call__(event, steering_context, **kwargs)

Update ledger after tool call.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

def __call__(self, event: AfterToolCallEvent, steering_context: SteeringContext, **kwargs: Any) -> None:
    """Update ledger after tool call."""
    ledger = steering_context.data.get("ledger") or {}

    if ledger.get("tool_calls"):
        last_call = ledger["tool_calls"][-1]
        last_call.update(
            {
                "completion_timestamp": datetime.now().isoformat(),
                "status": event.result["status"],
                "result": event.result["content"],
                "error": str(event.exception) if event.exception else None,
            }
        )
        steering_context.data.set("ledger", ledger)

LedgerBeforeToolCall

Bases: SteeringContextCallback[BeforeToolCallEvent]

Context provider for ledger tracking before tool calls.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

class LedgerBeforeToolCall(SteeringContextCallback[BeforeToolCallEvent]):
    """Context provider for ledger tracking before tool calls."""

    def __init__(self) -> None:
        """Initialize the ledger provider."""
        self.session_start = datetime.now().isoformat()

    def __call__(self, event: BeforeToolCallEvent, steering_context: SteeringContext, **kwargs: Any) -> None:
        """Update ledger before tool call."""
        ledger = steering_context.data.get("ledger") or {}

        if not ledger:
            ledger = {
                "session_start": self.session_start,
                "tool_calls": [],
                "conversation_history": [],
                "session_metadata": {},
            }

        tool_call_entry = {
            "timestamp": datetime.now().isoformat(),
            "tool_name": event.tool_use.get("name"),
            "tool_args": event.tool_use.get("arguments", {}),
            "status": "pending",
        }
        ledger["tool_calls"].append(tool_call_entry)
        steering_context.data.set("ledger", ledger)

__call__(event, steering_context, **kwargs)

Update ledger before tool call.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

def __call__(self, event: BeforeToolCallEvent, steering_context: SteeringContext, **kwargs: Any) -> None:
    """Update ledger before tool call."""
    ledger = steering_context.data.get("ledger") or {}

    if not ledger:
        ledger = {
            "session_start": self.session_start,
            "tool_calls": [],
            "conversation_history": [],
            "session_metadata": {},
        }

    tool_call_entry = {
        "timestamp": datetime.now().isoformat(),
        "tool_name": event.tool_use.get("name"),
        "tool_args": event.tool_use.get("arguments", {}),
        "status": "pending",
    }
    ledger["tool_calls"].append(tool_call_entry)
    steering_context.data.set("ledger", ledger)

__init__()

Initialize the ledger provider.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

def __init__(self) -> None:
    """Initialize the ledger provider."""
    self.session_start = datetime.now().isoformat()

LedgerProvider

Bases: SteeringContextProvider

Combined ledger context provider for both before and after tool calls.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

class LedgerProvider(SteeringContextProvider):
    """Combined ledger context provider for both before and after tool calls."""

    def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
        """Return ledger context providers with shared state."""
        return [
            LedgerBeforeToolCall(),
            LedgerAfterToolCall(),
        ]

context_providers(**kwargs)

Return ledger context providers with shared state.

Source code in strands/experimental/steering/context_providers/ledger_provider.py

def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
    """Return ledger context providers with shared state."""
    return [
        LedgerBeforeToolCall(),
        LedgerAfterToolCall(),
    ]

SteeringContext dataclass

Container for steering context data.

Source code in strands/experimental/steering/core/context.py

@dataclass
class SteeringContext:
    """Container for steering context data."""

    """Container for steering context data.

    This class should not be instantiated directly - it is intended for internal use only.
    """

    data: JSONSerializableDict = field(default_factory=JSONSerializableDict)

SteeringContextCallback

Bases: ABC, Generic[EventType]

Abstract base class for steering context update callbacks.

Source code in strands/experimental/steering/core/context.py

class SteeringContextCallback(ABC, Generic[EventType]):
    """Abstract base class for steering context update callbacks."""

    @property
    def event_type(self) -> type[HookEvent]:
        """Return the event type this callback handles."""
        for base in getattr(self.__class__, "__orig_bases__", ()):
            if get_origin(base) is SteeringContextCallback:
                return cast(type[HookEvent], get_args(base)[0])
        raise ValueError("Could not determine event type from generic parameter")

    def __call__(self, event: EventType, steering_context: "SteeringContext", **kwargs: Any) -> None:
        """Update steering context based on hook event.

        Args:
            event: The hook event that triggered the callback
            steering_context: The steering context to update
            **kwargs: Additional keyword arguments for context updates
        """
        ...

event_type property

Return the event type this callback handles.

__call__(event, steering_context, **kwargs)

Update steering context based on hook event.

Parameters:

Name	Type	Description	Default
event	EventType	The hook event that triggered the callback	required
steering_context	SteeringContext	The steering context to update	required
**kwargs	Any	Additional keyword arguments for context updates	{}

Source code in strands/experimental/steering/core/context.py

def __call__(self, event: EventType, steering_context: "SteeringContext", **kwargs: Any) -> None:
    """Update steering context based on hook event.

    Args:
        event: The hook event that triggered the callback
        steering_context: The steering context to update
        **kwargs: Additional keyword arguments for context updates
    """
    ...

SteeringContextProvider

Bases: ABC

Abstract base class for context providers that handle multiple event types.

Source code in strands/experimental/steering/core/context.py

class SteeringContextProvider(ABC):
    """Abstract base class for context providers that handle multiple event types."""

    @abstractmethod
    def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
        """Return list of context callbacks with event types extracted from generics."""
        ...

context_providers(**kwargs) abstractmethod

Return list of context callbacks with event types extracted from generics.

Source code in strands/experimental/steering/core/context.py

@abstractmethod
def context_providers(self, **kwargs: Any) -> list[SteeringContextCallback]:
    """Return list of context callbacks with event types extracted from generics."""
    ...