You can use tool hooks to perform validation, logging, or any other logic before or after a tool is called.
A tool hook is a function that takes a function name, function call, and arguments. Optionally, you can access the Agent or Team object as well. Inside the tool hook, you have to call the function call and return the result.
It is important to use exact parameter names when defining a tool hook. agent, team, function_name, function_call, and arguments are available parameters.
def logger_hook(
function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
"""Log the duration of the function call"""
start_time = time.time()
# Call the function
result = function_call(**arguments)
end_time = time.time()
duration = end_time - start_time
logger.info(f"Function {function_name} took {duration:.2f} seconds to execute")
# Return the result
return result
def confirmation_hook(
function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
"""Confirm the function call"""
if function_name != "get_top_hackernews_stories":
raise ValueError("This tool is not allowed to be called")
return function_call(**arguments)
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[DuckDuckGoTools()],
tool_hooks=[logger_hook],
)
Agent or Team object in the tool hook.
def grab_customer_profile_hook(
session_state: dict, function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
cust_id = arguments.get("customer")
if cust_id not in session_state["customer_profiles"]:
raise ValueError(f"Customer profile for {cust_id} not found")
customer_profile = session_state["customer_profiles"][cust_id]
# Replace the customer with the customer_profile for the function call
arguments["customer"] = json.dumps(customer_profile)
# Call the function with the updated arguments
result = function_call(**arguments)
return result
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[DuckDuckGoTools()],
tool_hooks=[logger_hook, confirmation_hook], # The logger_hook will run on the outer layer, and the confirmation_hook will run on the inner layer
)
@tool(tool_hooks=[logger_hook, confirmation_hook])
def get_top_hackernews_stories(num_stories: int) -> Iterator[str]:
"""Fetch top stories from Hacker News.
Args:
num_stories (int): Number of stories to retrieve
"""
# Fetch top story IDs
response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
story_ids = response.json()
# Yield story details
final_stories = []
for story_id in story_ids[:num_stories]:
story_response = httpx.get(
f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
)
story = story_response.json()
if "text" in story:
story.pop("text", None)
final_stories.append(story)
return json.dumps(final_stories)
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[get_top_hackernews_stories],
)
Pre and Post Hooks
Pre and post hooks let’s you modify what happens before and after a tool is called. It is an alternative to tool hooks.
Set the pre_hook in the @tool decorator to run a function before the tool call.
Set the post_hook in the @tool decorator to run a function after the tool call.
Here’s a demo example of using a pre_hook, post_hook along with Agent Context.
import json
from typing import Iterator
import httpx
from agno.agent import Agent
from agno.tools import FunctionCall, tool
def pre_hook(fc: FunctionCall):
print(f"Pre-hook: {fc.function.name}")
print(f"Arguments: {fc.arguments}")
print(f"Result: {fc.result}")
def post_hook(fc: FunctionCall):
print(f"Post-hook: {fc.function.name}")
print(f"Arguments: {fc.arguments}")
print(f"Result: {fc.result}")
@tool(pre_hook=pre_hook, post_hook=post_hook)
def get_top_hackernews_stories(agent: Agent) -> Iterator[str]:
num_stories = agent.context.get("num_stories", 5) if agent.context else 5
# Fetch top story IDs
response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
story_ids = response.json()
# Yield story details
for story_id in story_ids[:num_stories]:
story_response = httpx.get(
f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
)
story = story_response.json()
if "text" in story:
story.pop("text", None)
yield json.dumps(story)
agent = Agent(
dependencies={
"num_stories": 2,
},
tools=[get_top_hackernews_stories],
markdown=True,
)
agent.print_response("What are the top hackernews stories?", stream=True)
- Add hooks to tools for user confirmation
- Handle user input during tool execution
- Gracefully cancel operations based on user choice
Create the example
"""🤝 Human-in-the-Loop: Adding User Confirmation to Tool Calls
This example shows how to implement human-in-the-loop functionality in your Agno tools.
It shows how to:
- Add tool hooks to tools for user confirmation
- Handle user input during tool execution
- Gracefully cancel operations based on user choice
Some practical applications:
- Confirming sensitive operations before execution
- Reviewing API calls before they're made
- Validating data transformations
- Approving automated actions in critical systems
Run `pip install openai httpx rich agno` to install dependencies.
"""
import json
from typing import Any, Callable, Dict, Iterator
import httpx
from agno.agent import Agent
from agno.exceptions import StopAgentRun
from agno.models.openai import OpenAIChat
from agno.tools import FunctionCall, tool
from rich.console import Console
from rich.pretty import pprint
from rich.prompt import Prompt
# This is the console instance used by the print_response method
# We can use this to stop and restart the live display and ask for user confirmation
console = Console()
def confirmation_hook(
function_name: str, function_call: Callable, arguments: Dict[str, Any]
):
# Get the live display instance from the console
live = console._live
# Stop the live display temporarily so we can ask for user confirmation
live.stop() # type: ignore
# Ask for confirmation
console.print(f"\nAbout to run [bold blue]{fc.function.name}[/]")
message = (
Prompt.ask("Do you want to continue?", choices=["y", "n"], default="y")
.strip()
.lower()
)
# Restart the live display
live.start() # type: ignore
# If the user does not want to continue, raise a StopExecution exception
if message != "y":
raise StopAgentRun(
"Tool call cancelled by user",
agent_message="Stopping execution as permission was not granted.",
)
# Call the function
result = function_call(**arguments)
# Optionally transform the result
return result
@tool(tool_hooks=[confirmation_hook])
def get_top_hackernews_stories(num_stories: int) -> Iterator[str]:
"""Fetch top stories from Hacker News.
Args:
num_stories (int): Number of stories to retrieve
Returns:
str: JSON string containing story details
"""
# Fetch top story IDs
response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
story_ids = response.json()
# Yield story details
final_stories = []
for story_id in story_ids[:num_stories]:
story_response = httpx.get(
f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
)
story = story_response.json()
if "text" in story:
story.pop("text", None)
final_stories.append(story)
return json.dumps(final_stories)
# Initialize the agent with a tech-savvy personality and clear instructions
agent = Agent(
model=OpenAIChat(id="gpt-5-mini"),
tools=[get_top_hackernews_stories],
markdown=True,
)
agent.print_response(
"Fetch the top 2 hackernews stories?", stream=True, console=console
)
Run the example
Install librariesExport your keyexport OPENAI_API_KEY=xxx
