Skip to content

Using hooks

This guide covers how to register hooks on agents, bundle hooks with plugins, pass context through invocation state, and use hooks with multi-agent orchestrators. It assumes familiarity with hooks concepts and a working agent.

Register a callback for a specific event with agent.add_hook() (Python) or agent.addHook() (TypeScript).

Add a type hint to the callback parameter and the event type is inferred automatically. Or pass the event class explicitly:

from strands import Agent
from strands.hooks import BeforeInvocationEvent, BeforeToolCallEvent
agent = Agent()
# Type-inferred registration (recommended)
def on_invocation(event: BeforeInvocationEvent) -> None:
print(f"Invocation starting for agent: {event.agent.name}")
agent.add_hook(on_invocation)
# Explicit registration with event class
def on_tool_call(event: BeforeToolCallEvent) -> None:
print(f"Tool called: {event.tool_use['name']}")
agent.add_hook(on_tool_call, BeforeToolCallEvent)

Group related hooks into a single unit with the Plugin class. See Plugins for the full plugin API.

Decorate methods with @hook. The event type is inferred from the type hint:

from strands import Agent
from strands.plugins import Plugin, hook
from strands.hooks import BeforeToolCallEvent, AfterToolCallEvent
class LoggingPlugin(Plugin):
name = "logging-plugin"
@hook
def log_before(self, event: BeforeToolCallEvent) -> None:
print(f"Calling: {event.tool_use['name']} with {event.tool_use['input']}")
@hook
def log_after(self, event: AfterToolCallEvent) -> None:
print(f"Completed: {event.tool_use['name']}")
agent = Agent(plugins=[LoggingPlugin()])
agent("Summarize today's news")

Pass request-scoped data (user IDs, database connections, configuration) into hooks via invocation_state. Add custom values as keyword arguments when invoking the agent:

import sqlite3
from strands import Agent
from strands.hooks import BeforeToolCallEvent
def audit_tool_calls(event: BeforeToolCallEvent) -> None:
user_id = event.invocation_state.get("user_id", "unknown")
db = event.invocation_state.get("db")
tool_name = event.tool_use["name"]
print(f"[audit] user={user_id} tool={tool_name}")
if db:
cursor = db.cursor()
cursor.execute(
"INSERT INTO audit_log (user_id, tool_name) VALUES (?, ?)",
(user_id, tool_name),
)
db.commit()
agent = Agent()
agent.add_hook(audit_tool_calls)
db_conn = sqlite3.connect(":memory:")
db_conn.execute("CREATE TABLE audit_log (user_id TEXT, tool_name TEXT)")
result = agent(
"Look up the weather in Portland",
user_id="user-42",
db=db_conn,
)

Register hooks on Graph or Swarm orchestrators for node-level events like BeforeNodeCallEvent and AfterNodeCallEvent.

from strands import Agent
from strands.hooks import BeforeNodeCallEvent, AfterNodeCallEvent
from strands.multiagent import Graph
researcher = Agent(system_prompt="You are a research specialist.")
writer = Agent(system_prompt="You are a writing specialist.")
graph = Graph(
agents={"researcher": researcher, "writer": writer},
edges=[("researcher", "writer")],
)
def log_node_start(event: BeforeNodeCallEvent) -> None:
print(f"Node {event.node_id} starting")
def log_node_end(event: AfterNodeCallEvent) -> None:
print(f"Node {event.node_id} completed")
graph.hooks.add_callback(BeforeNodeCallEvent, log_node_start)
graph.hooks.add_callback(AfterNodeCallEvent, log_node_end)

Layered hooks: agent-level and orchestrator-level

Section titled “Layered hooks: agent-level and orchestrator-level”

Combine agent-level and orchestrator-level hooks. Both fire during execution:

from strands import Agent
from strands.plugins import Plugin, hook
from strands.hooks import (
BeforeToolCallEvent,
BeforeNodeCallEvent,
AfterNodeCallEvent,
HookProvider,
HookRegistry,
)
from strands.multiagent import Graph
# Agent-level plugin: logs every tool call inside each agent
class AgentLoggingPlugin(Plugin):
name = "agent-logging"
@hook
def log_tool(self, event: BeforeToolCallEvent) -> None:
print(f" [agent] tool call: {event.tool_use['name']}")
# Orchestrator-level hook provider: logs node transitions
class OrchestratorLoggingHook(HookProvider):
def register_hooks(self, registry: HookRegistry) -> None:
registry.add_callback(BeforeNodeCallEvent, self.on_node_start)
registry.add_callback(AfterNodeCallEvent, self.on_node_end)
def on_node_start(self, event: BeforeNodeCallEvent) -> None:
print(f"[orchestrator] node {event.node_id} starting")
def on_node_end(self, event: AfterNodeCallEvent) -> None:
print(f"[orchestrator] node {event.node_id} completed")
# Wire both layers together
agent1 = Agent(system_prompt="You research topics.", plugins=[AgentLoggingPlugin()])
agent2 = Agent(system_prompt="You write summaries.", plugins=[AgentLoggingPlugin()])
graph = Graph(
agents={"researcher": agent1, "writer": agent2},
edges=[("researcher", "writer")],
hooks=[OrchestratorLoggingHook()],
)

Create a reusable hook collection (Python)

Section titled “Create a reusable hook collection (Python)”

Create a reusable hook collection by implementing the HookProvider protocol. This is lighter than a full plugin when you only need hooks:

from strands import Agent
from strands.hooks import (
HookProvider,
HookRegistry,
BeforeInvocationEvent,
AfterInvocationEvent,
)
class RequestLogger(HookProvider):
def register_hooks(self, registry: HookRegistry) -> None:
registry.add_callback(BeforeInvocationEvent, self.log_start)
registry.add_callback(AfterInvocationEvent, self.log_end)
def log_start(self, event: BeforeInvocationEvent) -> None:
print(f"Request started for agent: {event.agent.name}")
def log_end(self, event: AfterInvocationEvent) -> None:
print(f"Request completed for agent: {event.agent.name}")
# Pass via hooks parameter
agent = Agent(hooks=[RequestLogger()])
# Or add after construction
agent.hooks.add_hook(RequestLogger())