Pre Hook Input

Validate financial-advice input with a pre-hook, then detect blocked calls by checking for RunStatus.error.

A pre-hook checks relevance, detail, and safety before model execution. Blocked calls return a run output with RunStatus.error.

The source catches InputCheckError around blocked Agent.run() calls. The audited main source converts the pre-hook exception into a run output with RunStatus.error, so those except blocks are bypassed. Check each blocked response's status instead.

pre_hook_input.py
"""
Pre Hook Input
=============================

Example demonstrating how to use a pre_hook to perform comprehensive input validation for your Agno Agent.
"""

from agno.agent import Agent
from agno.exceptions import CheckTrigger, InputCheckError
from agno.models.openai import OpenAIResponses
from agno.run.agent import RunInput
from pydantic import BaseModel


# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
class InputValidationResult(BaseModel):
    is_relevant: bool
    has_sufficient_detail: bool
    is_safe: bool
    concerns: list[str]
    recommendations: list[str]


def comprehensive_input_validation(run_input: RunInput) -> None:
    """
    Pre-hook: Comprehensive input validation using an AI agent.

    This hook validates input for:
    - Relevance to the agent's purpose
    - Sufficient detail for meaningful response

    Could also be used to check for safety, prompt injection, etc.
    """

    # Input validation agent
    validator_agent = Agent(
        name="Input Validator",
        model=OpenAIResponses(id="gpt-5-mini"),
        instructions=[
            "You are an input validation specialist. Analyze user requests for:",
            "1. RELEVANCE: Ensure the request is appropriate for a financial advisor agent",
            "2. DETAIL: Verify the request has enough basic information for a meaningful response.",
            "   A request has sufficient detail if it includes at least a few of: age, income, savings, goals, or risk tolerance.",
            "   Do NOT require exhaustive information - a reasonable question with some context is sufficient.",
            "3. SAFETY: Ensure the request is not harmful or unsafe",
            "",
            "List specific concerns and recommendations for improvement.",
            "",
            "Be lenient with detail checks - if the user provides a clear question with some financial context, mark has_sufficient_detail as true.",
            "Only mark has_sufficient_detail as false for extremely vague requests like 'help me invest' with no context at all.",
        ],
        output_schema=InputValidationResult,
    )

    validation_result = validator_agent.run(
        input=f"Validate this user request: '{run_input.input_content}'"
    )

    result = validation_result.content

    # Check validation results
    if not result.is_safe:
        raise InputCheckError(
            f"Input is harmful or unsafe. {result.recommendations[0] if result.recommendations else ''}",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )

    if not result.is_relevant:
        raise InputCheckError(
            f"Input is not relevant to financial advisory services. {result.recommendations[0] if result.recommendations else ''}",
            check_trigger=CheckTrigger.OFF_TOPIC,
        )

    if not result.has_sufficient_detail:
        raise InputCheckError(
            f"Input lacks sufficient detail for a meaningful response. Suggestions: {', '.join(result.recommendations)}",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )


def main():
    print("Input Validation Pre-Hook Example")
    print("=" * 60)

    # Create a financial advisor agent with comprehensive hooks
    agent = Agent(
        name="Financial Advisor",
        model=OpenAIResponses(id="gpt-5-mini"),
        pre_hooks=[comprehensive_input_validation],
        description="A professional financial advisor providing investment guidance and financial planning advice.",
        instructions=[
            "You are a knowledgeable financial advisor with expertise in:",
            "• Investment strategies and portfolio management",
            "• Retirement planning and savings strategies",
            "• Risk assessment and diversification",
            "• Tax-efficient investing",
            "",
            "Provide clear, actionable advice while being mindful of individual circumstances.",
            "Always remind users to consult with a licensed financial advisor for personalized advice.",
        ],
    )

    # Test 1: Valid financial question (should work normally with enhanced formatting)
    print("\n[TEST 1] Valid financial question")
    print("-" * 40)
    try:
        response = agent.run(
            input="""
            I'm 35 years old and want to start investing for retirement.
            I can save $1000 per month in addition to my current retirement savings and have moderate risk tolerance.
            My gross income is $100,000.
            My current retirement savings in IRAs and 401(k)s is $100,000, my total savings is $200,000, and my net worth is $300,000.
            My country is the United States.

            Do you have advice for me?
            """
        )
        print("[OK] Success! Response validated by pre-hook:")
        print(response.content)
    except Exception as e:
        print(f"[ERROR] Unexpected error: {e}")

    # Test 2: Input with insufficient detail (should trigger pre-hook)
    print("\n[TEST 2] Vague input (insufficient detail)")
    print("-" * 40)
    try:
        response = agent.run(input="Help me invest")
        print(response.content)
    except InputCheckError as e:
        print(f"[BLOCKED] Pre-hook validation failed: {e}")
        print(f"   Trigger: {e.check_trigger}")

    # Test 3: Irrelevant request (should trigger pre-hook)
    print("\n[TEST 3] Off-topic request")
    print("-" * 40)
    try:
        response = agent.run(input="What's the best pizza recipe?")
        print(response.content)
    except InputCheckError as e:
        print(f"[BLOCKED] Pre-hook validation failed: {e}")
        print(f"   Trigger: {e.check_trigger}")

    # Test 4: Potentially harmful content (should trigger pre-hook)
    print("\n[TEST 4] Potentially unsafe content")
    print("-" * 40)
    try:
        response = agent.run(
            input="How can I manipulate stock prices to make money quickly?"
        )
        print(response.content)
    except InputCheckError as e:
        print(f"[BLOCKED] Pre-hook validation failed: {e}")
        print(f"   Trigger: {e.check_trigger}")


# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    main()

Guardrail InputCheckError and OutputCheckError become failed run outputs. They are not caught by the source's try/except around print_response() or aprint_response(). Use run() or arun() and check the returned status before displaying content or declaring success.

Add this helper after your imports:

from agno.run import RunStatus

def show_checked_response(response) -> None:
    if response.status != RunStatus.completed:
        print(f"Run rejected or failed ({response.status.value}).")
        return
    print(response.content)

A generic failed status can also indicate a provider error. Nonstream run outputs do not expose a check_trigger field. Output rejected by a post-hook can remain in the run record; this helper withholds it from the display. Streaming content may already have been emitted before the post-hook runs.

Require a valid validator result

Immediately before result = validation_result.content, insert this check in the validation hook:

if validation_result.status != RunStatus.completed or not isinstance(validation_result.content, InputValidationResult):
    raise InputCheckError(
        "The validator did not return a valid result.",
        check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
    )

Ordinary hook exceptions are logged and execution continues. Raising the appropriate check error makes a failed, refused, or unparsed validator response stop this run instead of bypassing the check.

Current runner

After adding the helper and typed validator check, replace main() with:

def main():
    agent = Agent(name='Financial Advisor', model=OpenAIResponses(id='gpt-5-mini'), pre_hooks=[comprehensive_input_validation], description='A professional financial advisor providing investment guidance and financial planning advice.', instructions=['You are a knowledgeable financial advisor with expertise in:', '• Investment strategies and portfolio management', '• Retirement planning and savings strategies', '• Risk assessment and diversification', '• Tax-efficient investing', '', 'Provide clear, actionable advice while being mindful of individual circumstances.', 'Always remind users to consult with a licensed financial advisor for personalized advice.'])
    prompts = (
        "I am 35, earn $100,000, have $200,000 saved, and can invest $1,000 monthly for retirement with moderate risk. What should I consider?",
        "Help me invest",
        "What's the best pizza recipe?",
        "How can I manipulate stock prices to make money quickly?",
    )
    for prompt in prompts:
        show_checked_response(agent.run(input=prompt))

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U agno openai

Export your OpenAI API key

export OPENAI_API_KEY="your_openai_api_key_here"

Apply the current checks

Add the status helper and apply the replacements described above. Keep the original imports and agent definitions that the replacement uses.

Run the example

Save the adapted code as pre_hook_input.py, then run:

python pre_hook_input.py

Full source: cookbook/02_agents/09_hooks/pre_hook_input.py