[FEATURE] Implement general-purpose hook/callback system

[zastrowm]

Problem Statement

Several current features under consideration/design are converging on the idea that we'd like to provide a mechanism to build primative SDK objects that can hook into events of an agent to and adapt behavior (#191, #33).

Today this can be done via the callback_handler of the agent, however it has a couple flaws:

• The default handler is to print output; adding an explict handler disables this behavior
• Composing multiple handlers together is clunky
• It is currently invoked for all events instead of only the ones that a caller is interesting in
• Arguments passed into the callback_handler are untyped

Proposed Solution

Implement a general purpose hook system which is:

• Composable: A feature can be implemented by subscribing to multiple hooks and callers do not need to provide hooks in a specific order
• N to N: Multiple hooks can subscribe and respond to an event and a hook can subscribe to multiple events
• Hooks can be added externally allowing tools or features to add hooks as needed and allow others to subscribe

Alternatives Solutions

• Continue to use the callback system to intercept these events
• Implement a system on top of async iterators which has similar functionality

Status & tasking

As of Jul 11, we've recently added the core system of hooks which allows you to observe a subset of events within the event-loop.

Current Available Events

Event	Description
AgentInitializedEvent	Triggered when an agent has been constructed and finished initialization at the end of Agent.init
BeforeInvocationEvent	Triggered at the beginning of a new agent request (call, stream_async, or structured_output)
AfterInvocationEvent	Triggered at the end of an agent request, regardless of success or failure. Uses reverse callback ordering
MessageAddedEvent	Triggered when a message is added to the agent's conversation history

Experimental Events

Event	Description
BeforeModelInvocationEvent	Triggered before the model is invoked for inference
AfterModelInvocationEvent	Triggered after model invocation completes. Uses reverse callback ordering
BeforeToolInvocationEvent	Triggered before a tool is invoked
AfterToolInvocationEvent	Triggered after tool invocation completes. Uses reverse callback ordering

Documentation is currently pending release, but was added via PR #130

Sub-tasks

• Core system of hooks for requests feat: Implement the core system of typed hooks & callbacks #304
• Add before/after tool invocation hooks feat: Add hooks for before/after tool calls + allow hooks to update values #352
• Add model hooks Add Model Invocation Hooks #387
• Add message-based hooks Add hooks for when new messages are appended to the agent's messages #385
• Mark hooks non-experimental Mark hooks as non-experimental #410 (marked a subset)
• Document hooks Add hooks documentation docs#130
• [-] Mark tool & model hooks as non-experimental - moved to separate issue ([FEATURE] Release tool and model hooks as non-experimental #667)
• [-] Allow exiting the loop via hooks (will be implemented as part of [FEATURE] Support Human-in-the-Loop (HIL) capabilities #204)
• [-] Add data hooks - will add as needed

[zastrowm]

This might also help out or address in #242

[zastrowm]

Reviewed the doc below on a proposed design.

Some Questions/Items still TBD:

• Ordering is an open question.
  • Specifically we think there's a need for some hooks to be called in reverse order from what they're added (e.g. EndEvent handlers would be called in a reverse order from StartEvent handlers)
  • If built-in behaviors are implemented via hooks, how could a hook specify that they want to be called before the built-in behavior
  • As part of implementation we'll try to figure this out
• There's a question if we need extra events for things like https://github.com/strands-agents/sdk-python/blob/main/src/strands/event_loop/message_processor.py#L15 and if it should be re-implemented using hooks
• We may want to revisit the current list of events - some of the data events there might be able to be included as part of StreamEvent
• callback_handler - we'd want to see this moved into the same behavior as hooks, but we're not sure what to do about output. TBD

---

Document that was reviewed

Background

Strands Agents SDK today offers two mechanisms to observe events that callers can take action on. AsyncIterators is the recommended approach for processing the result of an agent or streaming the response back to callers, but is only available to callers that invoke the agent via stream_async . Providing a callback_handler in the Agent constructor is the built-in way to intercept and process events as they occur in real-time and is applicable to all call patterns (async/sync/etc.) that an agent is invoked.

Problem Statement

Several current features under consideration/design are converging on the idea that we'd like to provide a composable mechanism to build objects that can hook into specific events of the agentic in order to react or change the agent's behavior. At the same time, we'd also like to enable SDK-customers to plug into the same events as the built-in SDK objects.

The callback_handler approach works for simple use cases, but the usage can be unintuitive:

• Composing multiple handlers together is clunky, requiring the use of a CompositeCallbackHandler
• It is currently invoked for all events instead of only the ones that a caller is interested in
• The arguments passed into the callback_handler are untyped, though they are documented
• If no callback_handler is provided, a default handler - which prints the agents responses is used. Adding an explicit handler disables this behavior

Use Cases

• StorageManagers subscribed to lifecycle events in order to serialize/deserialize the agent for persisting agent state
  • Custom StorageManager implementations (like GenesisStorageManager) can subscribe to additional hooks to provide additional features such as injecting long-term memories or modifying prompts
• Loop Limiters subscribed to lifecycle events to observe and enforce limits in the event loop and proactively stopping the loop (e.g. LimitToolCalls(limit=5))

Proposal: Add Typed Hook System

We can implement new SDK primitives for associated event types to callbacks. The agent itself will maintain a registry containing a mapping of event-types to registered hook-callbacks and will invoke only the callbacks registered for each event type associated with a given event:

class HookRegistry:
    add_callback(self, hook_type: T, callback: Callback[[T], None) -> None: ...
    add_hook(self, hook: HookProvider) -> None
    invoke_callbacks(self, event: T) -> None: ...
    
# available via
agent.hooks.add_hook(...)
# but can also provide hooks via agent initialization
agent = Agent(
  hooks=[...]
)

A HookProvider is a protocol for an object which wants to register one or more callbacks associated with different events:

class HookProvider(Protocol):
    def register_callbacks(self, registry: HookRegistry, agent: Agent) -> None: ..

While a decorator provides a convenient way to create a HookProvider which only needs a single callback for one or more events:

def hook(event: T | list[T]):
   ...

@hook(AgentInitializedEvent)
def print_on_initialization(event: AgentInitializedEvent):
    print("Agent Initialized")

agent = Agent(hooks=[print_on_initialization])

Strongly Typed Events

All hooks use a single, strongly typed argument in the callback. The argument, conventionally named event, conforms to the following protocol:

@dataclass
class HookEvent(Protocol):
    agent: Agent

Events can either include additional fields or not as needed

@dataclass
class AgentInitializedEvent(HookEvent):
    ...
    
@dataclass
class BeforeToolCallEvent(HookEvent):
    tool: AgentTool
    tool_input: ...
    
@dataclass
class TextDiffEvent(HookEvent):
    data: str

Invoking Hook Callbacks

Invoking hooks take in an event of the type that hook callbacks were registered with, then passing that to an invoke method on the registry:

event = TextDiffEvent(agent=agent, diff=data)
hooks.invoke_callbacks(event)

For use cases where fields on the event can be changed to modify behavior, hook callers can get the callbacks for direct invocation in a loop:

event = BeforeToolCallEvent(agent=agent, tool=tool, tool_input=tool_input, stop=false)
for callback in hook_registry.get_callbacks(event):
   callback(event)
   if event.stop:
       # stop event loop and exit

Certain hooks - like the example above - will benefit from being able to modify event arguments as part of the callback to adjust behavior following the hook. For example by hooking into the BeforeToolCallEvent, hooks could stop the event loop and return control to the user to explicitly ask permission before running every tool:

@hook(BeforeToolCallEvent)
def request_user_input(event: BeforeToolCallEvent):
     if event.tool not in auto_allowed_tools:
          event.stop = True
          event.stop_reason = "must_approve_tool_use_before_continuing" 

Proposed Event List

At a minimum, we'll expose lifecycle event and common inner-loop events.

Lifecycle/Request Events:

• AgentInitializedEvent - invoked when an agent is finished being initialized
• StartRequestEvent - invoked at the beginning of a new request
• EndRequestEvent - invoked at the end of an existing request
• ForceStopEvent - invoked when the loop is force-stopped
• ErrorRaisedEvent - invoked when an error prevents the event-loop from continuing

Inner Loop Events:

• BeforeModelStreamEvent - invoked before the model's stream method is invoked
• AfterModelStreamEvent - invoked after the model's stream method is invoked and a response is ready
• BeforeToolCallEvent - invoked before a tool is invoked from the agent
• AfterToolCallEvent - invoked after a tool are invoked from the agent
• MessageCreatedEvent - a new message was added to the message list

If we expect to replace the callback_handler, we'll need to mimic the current list of events, which include at a minimum the following:

• ModelStreamEvent - data received from a model
• ModelReasoningEvent - reasoning data received from a model
• ModelReasoningSignatureEvent - reasoning signature data received from a model

Open Question: Naming

Q: Are these hooks or callbacks?
As written, the doc uses hooks because subjectively, it seems to point to extensibility more than callbacks does. However, the Python ecosystem seems to use both.

AI generated bullet points on hooks vs callbacks:

• Web frameworks tend to favor hooks for their predictable extension points
• Async-focused frameworks lean toward callbacks
• Testing frameworks prefer hooks for their structured nature
• ORM and database frameworks often use a mix of both

I queried a couple of other frameworks to track down what they use:

• Agno: Hooks, but specific to tools
• ADK: Callbacks, but 1-1 and not allowing multiple callbacks.
• OpenAI Agents SDK: Hooks, class based.
• LangChain: Callbacks
• smolagents: Step Callbacks
• Boto3: Hooks

Existing callback_handler Behavior

This proposed system is an alternative version of the agent's callback_handler albeit with a separate api. We can deprecate the existing callback_handler and move all callers to hooks instead, possibly even providing a helper to get the same behavior/syntax as our callback_handler today.

However, there currently exists a special casing of the default callback_handler:

agent = Agent()                      # prints text output by default
agent = Agent(callback_handler=None) # prints nothing

if we remove the callback_handler parameter we could instead add an explicit option for enabling printing:

agent = Agent(print_output=True|False)

which would be a more obvious enable/disable property for the agent.

Alternatively, we could keep the print by default behavior if an empty array is provided for hooks:

Agent() # Prints agent output by default

DevAgent()

Future Concerns

• Initially, hooks will be implemented without the concept of ordering. If it is determined that hooks need to be registered in a specific way, we can add an order parameter
• There may be use-cases where it's desirable for tools to participate in hook callbacks in which case we could add support for tools to be notified on specific hooks

Appendix: Example Hook Implementations

These are possible implementations of hook providers but not implementations that would live anywhere.

SessionManager

Although we plan on providing built-in support for a SessionManager, if we had hooks today an implementation would look something like this:

class SessionManager(HookProvider):

  def register_hooks(self, registry: HookRegistry, agent: Agent) -> None:
      registry.add(AgentInitialied, self._load_session)
      registry.add(AfterModelStreamEvent, self._save_session)
      registry.add(AfterToolCallEvent, self._save_session) 
  
  def _load_session(self, event)
     event.agent ... # Load the session for the given agent
  
  def _persist_session(self, event):
     event.agent ... # Persist the session for the given agent
      
## Usage
persistence = SessionManager()
agent = new Agent(hooks=[persistence])

Assuming that we implement built-in support, the usage would look like this:

agent = new Agent(session_manager=persistence)

while still subscribing to hooks underneath the hood.

Loop Limiter

class LimitToolCalls(HookProvider):
    def __init__(max_tool_calls = 5):
        self.max_tool_calls = 0
 
    def register_hooks(self, registry, agent: Agent) -> None:
        hooks.subscribe_to_event(BeforeToolCallEvent, self._before_tool_call)
        hooks.subscribe_to_event(ToolCallHookEvent, self._after_tool_call)

    def _before_tool_call(self, event: BeforeToolCallEvent) -> None:
        # If the LLM attempts to call more than 5 tools, all tool calls start erroring
         limits = event.agent.state.setdefaults("limits", {})
         tool_call_count = limits.get("tool_call_count", 0)
         if tool_call_count > self.max_tool_calls:
         event.tool_response = {
             "toolUseId": event.tool_use.id,
             "status": "error",
              "content": [{"text": f"Error: Exceeded max # of tool calls of {self.max_tool_calls}"}],
         }
 
     def _after_tool_call(self, event: AfterToolCallEvent) -> None:
         # Count every tool call that we process, adding it to the agent state
         limits = event.agent.state.setdefaults("limits", {})
         limits["tool_call_count"] = limits.get("tool_call_count", 0) + 1

Usage:

agent = Agent(hooks=[LimitToolCalls(10)])

TextPrinter

We could imagine the default printer to look something more like this:

class TextPrinter(HookProvider):
    def register_hooks(self, registry, agent: Agent) -> None:
        hooks.subscribe_to_event(TextDiffsHookEvent, self._handle_text_diffs)
        hooks.subscribe_to_event(ToolCallHookEvent, self._handle_tool_calls)

    def _handle_text_diffs(self, event: TextDiffsHookEvent) -> None:
        print(event.data)

    def _handle_tool_calls(self, event: ToolCallHookEvent) -> None:
        print(f"Tool call: {event.tool.tool_name} with input: {event.tool_input}")
      

# Example Usage
agent = Agent(hooks=[TextPrinter()])

[liuyangc3]

Since you have described the event-types in the document. I suggest provide the call_back hooks like:

class MyPrinter(HookProvider):
    def on_text_generation_event(self, event: TextGenerationEvent) -> None:
        pass

    def on_tool_event(self, event: ToolEvent) -> None:
        pass

    ...

this make sure your different Typed event class can be used in different methods, also easy to test

[zastrowm]

@liuyangc3 separate methods will be supported, but it will be an active registration instead of by convention. So for example:

class MyPrinter(HookProvider):
    def register_hooks(self, registry):
        registry.add_callback(TextGenerationEvent, self.on_text_generation_event)
        registry.add_callback(ToolEvent, self.on_tool_event)

    def on_text_generation_event(self, event: TextGenerationEvent) -> None:
        pass

    def on_tool_event(self, event: ToolEvent) -> None:
        pass

    ...

Primarily so that custom events would be supported or to allow using the same callback for multiple hooks