Input Validation Pre-Hook

from agno.agent import Agent
from agno.exceptions import CheckTrigger, InputCheckError
from agno.models.openai import OpenAIResponses
from agno.run.team import TeamRunInput
from agno.team import Team
from agno.run import RunStatus
from pydantic import BaseModel


class TeamInputValidationResult(BaseModel):
    is_relevant: bool
    benefits_from_team: bool
    has_sufficient_detail: bool
    is_safe: bool
    concerns: list[str]
    recommendations: list[str]
    confidence_score: float


def comprehensive_team_input_validation(run_input: TeamRunInput, team: Team) -> None:
    """Validate input relevance, safety, and collaboration suitability for teams."""

    team_info = f"Team '{team.name}' with {len(team.members)} members: "
    team_info += ", ".join([member.name for member in team.members])

    validator_agent = Agent(
        name="Team Input Validator",
        model=OpenAIResponses(id="gpt-5.2"),
        instructions=[
            "You are a team input validation specialist. Analyze user requests for team execution:",
            "1. RELEVANCE: Ensure the request is appropriate for this specific team's capabilities",
            "2. TEAM BENEFIT: Verify the request genuinely benefits from multiple team members collaborating",
            "3. DETAIL: Check if there's enough information for effective team coordination",
            "4. SAFETY: Ensure the request is safe and appropriate for team execution",
            "",
            "Consider whether a single agent could handle this just as effectively.",
            "Teams work best for complex, multi-faceted problems requiring diverse expertise.",
            "Provide a confidence score (0.0-1.0) for your assessment.",
            "",
            "Be thorough but not overly restrictive - allow legitimate team requests through.",
        ],
        output_schema=TeamInputValidationResult,
    )

    try:
        validation_result = validator_agent.run(
            input=f"""
            {team_info}

            Validate this user request for team execution: '{run_input.input_content}'

            Don't be too restrictive!
            """
        )
    except Exception as exc:
        raise InputCheckError("Validator unavailable; validation is required.") from exc

    if validation_result.status != RunStatus.completed or not isinstance(
        validation_result.content, TeamInputValidationResult
    ):
        raise InputCheckError("Validator did not return a valid assessment.")
    result = validation_result.content

    if not 0.0 <= result.confidence_score <= 1.0:
        raise InputCheckError("Validator returned an invalid confidence score.")

    if not result.is_safe:
        raise InputCheckError(
            f"Input is unsafe for team execution. {result.recommendations[0] if result.recommendations else ''}",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )

    if not result.is_relevant:
        raise InputCheckError(
            f"Input is not suitable for this team's capabilities. {result.recommendations[0] if result.recommendations else ''}",
            check_trigger=CheckTrigger.OFF_TOPIC,
        )

    if not result.has_sufficient_detail:
        raise InputCheckError(
            "Input lacks sufficient detail. " + ", ".join(result.recommendations),
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )

    if not result.benefits_from_team:
        raise InputCheckError(
            f"This request would be better handled by a single agent rather than a team. Recommendation: {result.recommendations[0] if result.recommendations else 'Use a single specialized agent instead.'}",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )

    if result.confidence_score < 0.7:
        raise InputCheckError(
            f"Input validation confidence too low ({result.confidence_score:.2f}). Concerns: {', '.join(result.concerns)}",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )


frontend_agent = Agent(
    name="Frontend Developer",
    model=OpenAIResponses(id="gpt-5.2"),
    description="Expert in React, TypeScript, and modern frontend development",
)

backend_agent = Agent(
    name="Backend Developer",
    model=OpenAIResponses(id="gpt-5.2"),
    description="Specialist in Node.js, APIs, databases, and server architecture",
)

devops_agent = Agent(
    name="DevOps Engineer",
    model=OpenAIResponses(id="gpt-5.2"),
    description="Expert in deployment, CI/CD, cloud infrastructure, and monitoring",
)

dev_team = Team(
    name="Software Development Team",
    members=[frontend_agent, backend_agent, devops_agent],
    pre_hooks=[comprehensive_team_input_validation],
    description="A full-stack software development team providing comprehensive technical solutions.",
    instructions=[
        "Collaborate to provide complete software development guidance:",
        "Frontend Developer: Handle UI/UX, client-side architecture, and user experience",
        "Backend Developer: Cover server logic, APIs, databases, and system design",
        "DevOps Engineer: Address deployment, scaling, monitoring, and infrastructure",
        "",
        "Work together to deliver production-ready solutions.",
    ],
)


def main() -> None:
    requests = ['Plan a chat application with web and mobile interfaces, authentication, message history, and deployment monitoring.', 'How do I center a div in CSS?', 'What is the best pizza recipe?']
    for request in requests:
        response = dev_team.run(request)
        if response.status != RunStatus.completed:
            print("Request rejected or run failed; no answer displayed.")
            continue
        print(response.content)


if __name__ == "__main__":
    main()