How to Implement Pre and Post Hooks in Agno Agents for Custom Logic
Agno agents support pre-hooks and post-hooks that execute automatically around every run, enabling custom logic for input validation, logging, telemetry, and response modification without altering core agent code.
The Agno framework provides a robust hook system that allows developers to implement pre and post hooks in Agno agents for custom logic execution. These hooks integrate seamlessly into the agent's execution pipeline, running automatically before and after the LLM processing cycle. By leveraging the hook architecture defined in the agno-agi/agno repository, you can inject observability, validation, or transformation logic while maintaining clean separation from the core agent implementation.
Understanding the Hook Architecture
Agno's Agent object supports two distinct hook pipelines that wrap every run() and arun() invocation.
Pre-Hooks vs Post-Hooks
Pre-hooks execute immediately after the session loads but before any tool calls or LLM processing occurs. They receive the RunInput object and can modify it before the agent processes the request.
Post-hooks run after the LLM assembles its response but before the final RunOutput returns to the caller. They receive the completed RunOutput for logging, validation, or side effects.
Hook Signatures and Context
Both hook types receive rich context objects. Pre-hooks use the signature:
def hook(
run_input: RunInput,
agent: Agent,
session: AgentSession,
run_context: RunContext,
user_id: Optional[str] = None,
debug_mode: Optional[bool] = None,
**extra
) -> None
Post-hooks substitute run_output: RunOutput for run_input. Async variants use async def and are executed via aexecute_pre_hooks and aexecute_post_hooks in libs/agno/agno/agent/_hooks.py.
Implementing Pre-Hooks in Agno Agents
Basic Synchronous Pre-Hook
Create a function that logs incoming prompts before the agent processes them:
# my_hooks.py
def log_run_start(run_input, agent, session, run_context, **_):
"""Log the incoming prompt for observability."""
print(f"[Hook] Agent {agent.name!r} is about to run:")
print(f" Prompt: {run_input.messages[-1].content!r}")
Attach the hook via the pre_hooks parameter defined in libs/agno/agno/agent/agent.py:
from agno.agent.agent import Agent
from my_hooks import log_run_start
agent = Agent(
name="DemoAgent",
model=..., # your model configuration
pre_hooks=[log_run_start],
)
response = agent.run("Explain the difference between AI and ML.")
print(response.output)
Modifying Input with Pre-Hooks
Pre-hooks receive run_input by reference, allowing mutation before the LLM sees the data:
def inject_system_prompt(run_input, **_):
"""Prepend a system-level instruction to every run."""
system_msg = {"role": "system", "content": "You are a friendly assistant."}
run_input.messages.insert(0, system_msg)
The modified run_input propagates downstream because the hook operates on the same object instance passed to the LLM.
Implementing Post-Hooks in Agno Agents
Asynchronous Post-Hooks with Background Tasks
Post-hooks support async execution and can run as FastAPI background tasks when _run_hooks_in_background is enabled:
# async_hooks.py
import aiohttp
async def send_telemetry(run_output, agent, session, run_context, **_):
"""Fire-and-forget telemetry to an external endpoint."""
async with aiohttp.ClientSession() as client:
await client.post(
"https://telemetry.example.com/ingest",
json={"run_id": run_output.run_id, "tokens": run_output.token_usage},
)
Configure the agent to schedule hooks as background tasks:
from agno.agent.agent import Agent
from async_hooks import send_telemetry
agent = Agent(
name="TelemetryAgent",
model=...,
post_hooks=[send_telemetry],
_run_hooks_in_background=True,
)
output = await agent.arun("Summarize the latest news.")
print(output.output)
The implementation in libs/agno/agno/agent/_hooks.py handles background task scheduling via background_tasks.add_task(), allowing the agent to return responses immediately while hooks execute non-blocking side effects.
Combining Multiple Hooks and Execution Order
You can attach multiple hooks to a single agent. They execute sequentially in the order defined:
agent = Agent(
name="MultiHookAgent",
model=...,
pre_hooks=[log_run_start, inject_system_prompt],
post_hooks=[send_telemetry, lambda ro, **_: print("Run finished!")],
_run_hooks_in_background=False,
)
The enumerate(hooks) pattern in execute_pre_hooks and execute_post_hooks ensures deterministic sequencing. The filter_hook_args utility in libs/agno/agno/utils/hooks.py strips unused arguments, allowing hooks to define minimal signatures regardless of the full parameter set available.
Key Implementation Files
Understanding the source structure helps when debugging hook behavior:
libs/agno/agno/agent/agent.py– Defines theAgentdataclass includingpre_hooks,post_hooks, and_run_hooks_in_backgroundfields.libs/agno/agno/agent/_hooks.py– Containsexecute_pre_hooks,aexecute_pre_hooks,execute_post_hooks, andaexecute_post_hookswhich orchestrate hook execution, streaming, and background task scheduling.libs/agno/agno/agent/_run.py– Entry point that invokes hook executors from the main run loop and integrates hook iterators into the output event stream.libs/agno/agno/utils/hooks.py– Providesfilter_hook_argsfor argument filtering andcopy_args_for_backgroundfor background task preparation.
Summary
- Pre-hooks execute before LLM processing via
execute_pre_hooksinlibs/agno/agno/agent/_hooks.py, receivingRunInputfor validation or modification. - Post-hooks execute after response assembly via
execute_post_hooks, receivingRunOutputfor telemetry or logging. - Async support is available through
aexecute_pre_hooksandaexecute_post_hooks, with optional FastAPI background task execution when_run_hooks_in_background=True. - Argument filtering via
filter_hook_argsallows hooks to accept only the parameters they need. - Multiple hooks execute sequentially in list order, enabling composable middleware patterns.
Frequently Asked Questions
Can pre-hooks modify the input before it reaches the LLM?
Yes. Pre-hooks receive the run_input object by reference. Mutations such as inserting system messages or modifying the message list persist downstream because the hook operates on the same object instance passed to the LLM.
What happens if a hook raises an exception?
Non-validation exceptions are caught and logged via log_error or log_exception, allowing the run to continue. However, InputCheckError and OutputCheckError exceptions are re-raised immediately to abort the run, enabling strict validation workflows.
How do I run hooks asynchronously without blocking the agent response?
Set _run_hooks_in_background=True on the Agent instance. This schedules hooks as FastAPI background tasks using background_tasks.add_task(), allowing the agent to return the response immediately while hooks execute non-blocking side effects like external logging or analytics.
Can I use the same hook function for both pre and post hooks?
Technically yes, but you must design the function signature to accept both run_input and run_output parameters, or use **kwargs to ignore unused arguments. The filter_hook_args utility in libs/agno/agno/utils/hooks.py strips unused arguments, so a flexible signature can work for both hook types.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →