How-To: Use hooks to customize agent behavior#
Ragbits provides a hook system that lets you intercept and modify agent behavior at key points in the execution lifecycle. Hooks allow you to validate inputs, modify arguments, mask outputs, enforce guardrails, require user confirmation, and more, all without changing the agent or tool code itself.
The hook system supports five event types:
PRE_TOOL- triggered before a tool is invokedPOST_TOOL- triggered after a tool completesPRE_RUN- triggered before the agent run startsPOST_RUN- triggered after the agent run completesON_EVENT- triggered for each streaming event duringrun_streaming()
Hooks are executed in priority order (lower numbers first) and support chaining. Each hook receives the output of the previous one, enabling composable pipelines of transformations.
How to create and register a hook#
A hook is an instance of Hook that binds an async callback to a lifecycle event. You pass hooks directly to the Agent constructor:
from ragbits.agents import Agent
from ragbits.agents.hooks import EventType, Hook
from ragbits.core.llms import LiteLLM
from ragbits.core.llms.base import ToolCall
async def my_hook(tool_call: ToolCall) -> ToolCall:
"""A simple pre-tool hook that passes all calls through."""
return tool_call # decision defaults to "pass"
agent = Agent(
llm=LiteLLM("gpt-4o-mini"),
tools=[...],
hooks=[
Hook(
event_type=EventType.PRE_TOOL,
callback=my_hook,
tool_names=["my_tool"], # None = apply to all tools
priority=10, # lower runs first (default: 100)
),
],
)
The Hook constructor accepts:
event_type- one ofEventType.PRE_TOOL,POST_TOOL,PRE_RUN,POST_RUNorON_EVENTcallback- an async function matching the corresponding callback protocoltool_names- optional list of tool names this hook applies to. IfNone, the hook runs for every tool. This parameter is only relevant forPRE_TOOLandPOST_TOOLhooks.priority- execution order; lower numbers execute first (default:100)
How to validate and modify tool inputs with pre-tool hooks#
Pre-tool hooks receive a ToolCall and return a ToolCall. The returned ToolCall can have a decision field that controls whether the tool executes:
"pass"- allow the tool to run (optionally with modified arguments). This is the default."deny"- block the tool from running (requires areason)"ask"- request user confirmation before proceeding (requires areason)
Validate inputs#
This hook validates that an email argument has a correct format before allowing the send_notification tool to execute:
users = {
"123": {"name": "John Doe", "email": "john@example.com", "ssn": "123-45-6789", "balance": 5000},
"456": {"name": "Jane Smith", "email": "jane@example.com", "ssn": "987-65-4321", "balance": 3500},
}
return users.get(user_id, {"error": "User not found"})
def send_notification(email: str, message: str) -> str:
"""Send notification email to user."""
return f"Email sent to {email}: {message}"
async def validate_email(tool_call: ToolCall) -> ToolCall:
"""Validate email format before sending."""
if tool_call.name != "send_notification":
return tool_call
If validation fails, the hook returns a ToolCall with decision="deny" which immediately stops tool execution and returns the reason to the LLM.
Modify arguments#
This hook rewrites email domains to an approved list, demonstrating how pre-tool hooks can modify tool arguments:
if not re.match(email_pattern, email):
hook_actions.append({"hook": "validate_email", "action": "denied", "email": email})
return tool_call.model_copy(
update={
"decision": "deny",
"reason": f"Invalid email format: {email}",
}
)
hook_actions.append({"hook": "validate_email", "action": "passed", "email": email})
return tool_call
async def sanitize_email_domain(tool_call: ToolCall) -> ToolCall:
"""Ensure emails only go to approved domains."""
if tool_call.name != "send_notification":
return tool_call
email = tool_call.arguments.get("email", "")
approved_domains = ["example.com", "test.com"]
domain = email.split("@")[-1] if "@" in email else ""
if domain not in approved_domains:
modified_email = email.split("@")[0] + "@example.com"
modified_args = tool_call.arguments.copy()
modified_args["email"] = modified_email
The modified arguments are passed to the next hook in the chain (or to the tool itself if this is the last hook).
Chain multiple pre-tool hooks#
When multiple pre-tool hooks are registered, they execute in priority order and each hook sees the ToolCall modified by the previous one. If any hook returns "deny", execution stops immediately:
hook_actions.append(
{
"hook": "log_notification",
"action": "enhanced",
"original": original_output,
"enhanced": enhanced_output,
}
)
return ToolReturn(enhanced_output)
async def main() -> None:
In this example, validate_email (priority 10) runs first. If it passes, sanitize_email_domain (priority 20) runs next and may modify the email argument before the tool executes.
You can find the complete code example in the Ragbits repository here.
How to modify tool outputs with post-tool hooks#
Post-tool hooks receive the original ToolCall and the ToolReturn, and return a (possibly modified) ToolReturn.
Mask sensitive data#
This hook replaces sensitive fields in search results before they reach the LLM:
{
"hook": "sanitize_email_domain",
"action": "modified",
"original": email,
"modified": modified_email,
}
)
return tool_call.model_copy(update={"arguments": modified_args})
hook_actions.append({"hook": "sanitize_email_domain", "action": "passed", "email": email})
return tool_call
async def mask_sensitive_data(tool_call: ToolCall, tool_return: ToolReturn) -> ToolReturn:
"""Mask sensitive information in user search results."""
if tool_call.name != "search_user":
return tool_return
if isinstance(tool_return.value, dict) and "ssn" in tool_return.value:
Log tool outputs#
This hook logs the output of specific agent tools without modifying the result:
To apply it only to specific tools, use the tool_names parameter:
llm = LiteLLM("gpt-4o-mini")
# Child agent 1: Diet expert
diet_agent = Agent(
name="diet_expert",
description="A nutrition expert who provides diet plans and healthy eating advice",
You can find the complete code example in the Ragbits repository here.
How to validate agent input with pre-run hooks#
Pre-run hooks execute before the agent starts processing. They receive the user input, options, and context as separate arguments, and return the (potentially modified) input directly.
A common use case is integrating guardrails to block unsafe or policy-violating inputs:
fail_reason=f"Input contains blocked topic: {topic}",
)
return GuardrailVerificationResult(
guardrail_name=self.__class__.__name__,
succeeded=True,
fail_reason=None,
)
def create_guardrail_hook(
guardrail_manager: GuardrailManager,
) -> PreRunCallback:
"""Create a pre-run hook that validates input using guardrails."""
async def guardrail_hook(
To block the agent from processing the input, the hook returns a replacement message. This message becomes the agent's final output. No LLM call or tool execution happens.
Register it with the agent:
for result in results:
if not result.succeeded:
return f"I cannot help with that request. Reason: {result.fail_reason}"
return input
Tip
Pre-run hooks also work with streaming via agent.run_streaming(). If a pre-run hook replaces the input, the replacement text is streamed back directly.
You can find the complete code example in the Ragbits repository here.
How to modify agent results with post-run hooks#
Post-run hooks execute after the agent completes its run. They receive the AgentResult, options, and context as separate arguments, and return the (potentially modified) AgentResult.
Use post-run hooks to transform, enrich, or log final results:
from ragbits.agents._main import AgentResult
from ragbits.agents.hooks import EventType, Hook
async def enrich_result(result: AgentResult, options: object, context: object) -> AgentResult:
"""Add metadata to the agent result."""
result.metadata["processed_by"] = "post_run_hook"
return result
hook = Hook(event_type=EventType.POST_RUN, callback=enrich_result)
How to process streaming events with on-event hooks#
On-event hooks intercept every event emitted during agent.run_streaming(). They receive a StreamingEvent and can modify it, suppress it, or expand it into multiple events. This is useful for real-time filtering, transformation, or enrichment of the streaming output.
StreamingEvent is a union of all types that can appear in the streaming output: str (text chunks), ToolCall, ToolCallResult, ToolEvent, DownstreamAgentResult, SimpleNamespace, BasePrompt, Usage, and ConfirmationRequest.
Modify streaming events#
An on-event callback receives a single StreamingEvent and returns the modified event (or the same event unchanged):
from ragbits.agents.hooks import EventType, Hook
from ragbits.agents.hooks.types import StreamingEvent
async def redact_keywords(event: StreamingEvent) -> StreamingEvent | None:
"""Replace sensitive keywords in text chunks."""
if isinstance(event, str):
return event.replace("SECRET", "***")
return event
hook = Hook(event_type=EventType.ON_EVENT, callback=redact_keywords)
Suppress events#
Return None to remove an event from the stream entirely:
from ragbits.agents.hooks.types import StreamingEvent
from ragbits.core.llms.base import Usage
async def suppress_usage(event: StreamingEvent) -> StreamingEvent | None:
"""Remove Usage events from the stream."""
if isinstance(event, Usage):
return None
return event
Expand a single event into multiple events#
An on-event callback can also return an async generator, yielding multiple events from a single input:
from collections.abc import AsyncGenerator
from ragbits.agents.hooks.types import StreamingEvent
async def split_sentences(event: StreamingEvent) -> AsyncGenerator[StreamingEvent, None]:
"""Split text chunks on sentence boundaries."""
if isinstance(event, str) and ". " in event:
for sentence in event.split(". "):
yield sentence.strip() + "."
else:
yield event
When chained with other hooks, each yielded event is passed through subsequent hooks individually. For example, a splitting hook at priority 1 followed by an uppercasing hook at priority 2 will uppercase each split fragment separately.
Note
On-event hooks only run during agent.run_streaming(). They have no effect on agent.run().
How to require user confirmation before tool execution#
Ragbits includes a built-in create_confirmation_hook factory that creates a pre-tool hook requiring user approval before a tool runs:
from ragbits.agents import Agent
from ragbits.agents.hooks import create_confirmation_hook
from ragbits.core.llms import LiteLLM
agent = Agent(
llm=LiteLLM("gpt-4o-mini"),
tools=[delete_file, send_email],
hooks=[
create_confirmation_hook(tool_names=["delete_file", "send_email"]),
],
)
When the agent attempts to call one of the specified tools, the hook returns a ConfirmationRequest containing:
confirmation_id- a deterministic ID based on the hook, tool name, and argumentstool_name- the name of the tool being calledtool_description- a description of why confirmation is neededarguments- the tool arguments
The agent yields the ConfirmationRequest and pauses. You can then resume the agent with the user's decision by including it in the AgentRunContext.tool_confirmations list.
How hook chaining works#
All hook types support chaining: hooks execute in priority order (lower numbers first), and each hook receives the output of the previous one. When hooks share the same priority (the default is 100), they execute in the order they were defined.
For pre-tool hooks, the chained value is the ToolCall (including its arguments). For post-tool hooks, it is the ToolReturn. For pre-run hooks, it is the agent input. For post-run hooks, it is the AgentResult. For on-event hooks, the hooks are composed as a pipeline of async generators. Each hook wraps the previous one, so events flow through the entire chain without intermediate collection.
This makes it possible to compose independent hooks that each handle one concern (validation, sanitization, logging, etc.) into a clean pipeline.
Warning
For pre-tool hooks, if any hook in the chain returns "deny", execution stops immediately and subsequent hooks do not run. Design your hooks with this in mind. Place critical validation hooks at lower priority numbers so they run first.