Validating the outputs of agents and tools is crucial for building reliable and secure AI agents. Validators help ensure data integrity and handle LLM hallucinations.

Validators

There are 4 different validators in Agency Swarm:
  • Tool validators:
    • field_validator: Validate individual fields independently.
    • model_validator: Perform complex checks involving multiple fields.
  • Agent validators:
    • guardrails (for v1.x and higher) or response_validator (for v0.x): Validate the response before sending it to the user or other agents.
  • Universal validators:
    • llm_validator (v0.x): Validate outputs against specified natural language rules.

Agent Response Validator

In v1.x, validation is handled through guardrails using the agents SDK. Guardrails provide more flexible validation for both input and output.

Output Guardrails

Output guardrails validate agent responses before they are sent to users or other agents. Each output guardrail function receives three parameters: context, agent, and response_text.In most cases, you’ll only need the response_text parameter, which contains the agent’s latest response as a simple text string if agent does not have output_type specified or it will follow a pydantic schema provided in the output_type parameter. This allows you to check the response against specific criteria.For more complex validation scenarios, you can utilize the context and agent parameters to account for run state and agent configuration during validation.Example:
from agency_swarm import (
    Agency,
    Agent,
    RunContextWrapper,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    OutputGuardrailTripwireTriggered,
    output_guardrail,
)
from pydantic import BaseModel

@output_guardrail
async def response_content_guardrail(
    context: RunContextWrapper, agent: Agent, response_text: str | Type[BaseModel]
) -> GuardrailFunctionOutput:
    tripwire_triggered = False
    output_info = ""

    if "bad word" in response_text.lower():
        tripwire_triggered = True
        output_info = "Please avoid using inappropriate language."

    return GuardrailFunctionOutput(
        output_info=output_info,
        tripwire_triggered=tripwire_triggered,
    )

agent = Agent(
    name="CustomerSupportAgent",
    instructions="You are a helpful customer support agent.",
    output_guardrails=[response_content_guardrail],
)

Validation Attempts

The validation_attempts parameter controls how many times an agent can retry when output validation fails. Default is 1 (one retry). Set validation_attempts=0 for immediate fail-fast behavior.
agent = Agent(
    name="CustomerSupportAgent",
    instructions="You are a helpful customer support agent.",
    output_guardrails=[response_content_guardrail],
    validation_attempts=2,
)
After the guardrail error is triggered, the agent will receive the error message provided in the output_info in the form of a system message and will adjust its output accordingly.

Output Guardrail Exceptions

After all validation attempts fail, the OutputGuardrailTripwireTriggered error will be raised to allow for additional processing, if needed.
try:
    response = await agency.get_response("Hello!")
except OutputGuardrailTripwireTriggered as e:
    print(f"Validation failed: {e.guardrail_result.output_info}")

Input Guardrails

Input guardrails validate incoming messages before they reach the agent. They can check both user input and inter-agent communication.Simplified Input ProcessingAgency Swarm simplifies guardrail implementation by automatically extracting text content from messages. While the Agents SDK originally passes the entire chat history to guardrails, Agency Swarm wraps your guardrail functions to provide only the relevant text content, eliminating the need for manual extraction logic.Parameters and Input TypesEach input guardrail receives three parameters: context, agent, and user_input. The user_input parameter format depends on the message structure:
  • Single message: A string containing the message content
  • Multiple consecutive messages: A list of strings, one for each message
For example, when a user sends multiple messages:
[
  {"role": "user", "content": "Hi"},
  {"role": "user", "content": "How are you?"}
]
Your guardrail will receive:
["Hi", "How are you?"]
This allows you to process each new input message individually or validate them as a group.
Note that file and image inputs inside of a user message will not be passed to the guardrail.
Example:
from agency_swarm import (
    Agency,
    Agent,
    RunContextWrapper,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    input_guardrail,
)

@input_guardrail
async def require_task_prefix(
    context: RunContextWrapper, agent: Agent, user_input: str | list[str]
) -> GuardrailFunctionOutput:
    """Require user requests to begin with 'Request:'"""

    # Handle single string input
    condition = not user_input.startswith("Request:")

    return GuardrailFunctionOutput(
        output_info="Prefix your request with 'Request:' describing what you need." if condition else "",
        tripwire_triggered=condition,
    )

agent = Agent(
    name="CustomerSupportAgent",
    instructions="You are a helpful customer support agent.",
    input_guardrails=[require_task_prefix],
)

Input Guardrail Return Format

Agent’s configuration includes an optional parameter called throw_input_guardrail_error, which defines how the guardrail error will be processed and presented to the caller:
agent = Agent(
    name="CustomerSupportAgent",
    instructions="You are a helpful customer support agent.",
    input_guardrails=[require_task_prefix],
    throw_input_guardrail_error=False,
)
If set to True, triggering a guardrail will raise a corresponding exception, as described in the next section. If set to False (its default value), the guardrail error will be returned to the user/caller agent as if it were a genuine response from the recipient. For example, using the agent above:
response = await agency.get_response("Hello!")
print(response.final_output)
This will not raise an exception; however, the response.final_output will contain the guardrail’s error message. In this case, it will be:
"Prefix your request with 'Request:' describing what you need."
For streaming responses, it will emit an extra event at the end of the stream in the following format:
RunItemStreamEvent(
    name='message_output_created', 
    item=MessageOutputItem(
        agent=<Agent name='CustomerSupportAgent' desc='' model='unknown'>, 
        raw_item=ResponseOutputMessage(
            id='msg_input_guardrail_guidance', 
            content=[
                ResponseOutputText(
                    annotations=[], 
                    text="Prefix your request with 'Request:' describing what you need.", 
                    type='output_text', 
                    logprobs=None
                )
            ],
            role='assistant', 
            status='completed', 
            type='message'
        ), 
        type='message_output_item'
    ), 
    type='run_item_stream_event'
)
This event will always have an id set to msg_input_guardrail_guidance, so you can further differentiate it, if needed.When to use: This feature’s primary focus is to simplify building user-facing applications and make the user experience more fluid. When using this feature, you won’t have to add any exception handlers for the input guardrails. Instead, users will receive the response as if it were coming from the agent, so they can adjust their input accordingly.
While having this feature turned on produces an agent’s response, that response is purely artificial and will not be registered in the thread history.
Instead, an appropriate system message will be added in order to log a guardrail error.

Input Guardrail Exceptions

Unlike output guardrails, if an input guardrail exception is raised (throw_input_guardrail_error should be set to True for that to happen), the error will be immediately returned back to the user/caller agent.
For user messages, the error processing is similar to output guardrails.
try:
    response = await agency.get_response("Hello!")
except InputGuardrailTripwireTriggered as e:
    print(f"Validation failed: {e.guardrail_result.output_info}")
Each triggered guardrail will also leave a corresponding system message containing error message inside of a chat history. Regardless of the throw_input_guardrail_error parameter value.
The validation_attempts parameter currently does not apply to input guardrails.

Agent-to-agent message validation

You can use guardrails to control how agents send messages and responses to one another. When adding communication flows between agents, the guardrails of the recipient agent will define the format of input and output messages, for example:
@input_guardrail(name="RequireTaskPrefix")
async def require_task_prefix(
    context: RunContextWrapper, agent: Agent, agent_input: str | list[str]
) -> GuardrailFunctionOutput:
    condition = not agent_input.startswith("Task:")
    return GuardrailFunctionOutput(
        output_info="ERROR: Requests to this agent must begin with 'Task:'" if condition else "",
        tripwire_triggered=condition,
    )


@output_guardrail(name="RequireResponsePrefix")
async def require_response_prefix(
    context: RunContextWrapper, agent: Agent, response_text: str
) -> GuardrailFunctionOutput:
    condition = not response_text.startswith("Response:")
    return GuardrailFunctionOutput(
        output_info="ERROR: Your response must start with 'Response:'" if condition else "",
        tripwire_triggered=condition,
    )

ceo = Agent(
    name="CEO",
    instructions="You are the CEO agent.",
)

worker = Agent(
    name="Worker",
    instructions="You are the worker agent.",
    input_guardrails=[require_task_prefix],
    output_guardrails=[require_response_prefix],
    throw_input_guardrail_error=True,
)

agency = Agency(
    ceo,
    communication_flows=[(ceo, worker)],
)
In this example, if the CEO agent sends a message to the worker that does not start with the word “Task:”, the input guardrail will be triggered and the CEO will be notified with an error message saying "ERROR: Requests to this agent must begin with 'Task:'". Then it will either adjust the input and try again or notify the user, as per its instructions.
Agent-to-agent messages are always single strings, so if you are using a guardrail exclusively to check inter-agent communication, input guardrail will always receive a string.
Similarly, the response of the worker to the CEO agent will be checked against the output guardrail and within the specified number of validation_attempts it will need to generate a correct response. In this case, the requirement is that the worker’s response should start with the word “Response:”, otherwise it will receive the error message "ERROR: Your response must start with 'Response:'"It is recommended to set throw_input_guardrail_error=True for the agency’s internal agents. While False value is also supported, setting it to True will add an extra error prefix to the recipient’s response to indicate the issue and avoid potential confusion.
Due to the nature of Handoffs, using SendMessageHandoff for agent-to-agent communication will bypass input guardrails set between agents.

Tool Validators

When defining tools, you can use Pydantic validators to prevent invalid data from being passed to the tool by the calling agent. There are 2 types of validators used specifically in tools: field-level and model-level validators. Here is the comparison table to help you understand the difference between them:
TypePurposeUsage
Field ValidatorsValidate individual fields independently.Use the @field_validator decorator on methods, specifying the field(s) to validate.
Model ValidatorsValidate the entire model, allowing checks involving multiple fields.Use the @model_validator decorator on methods.
This example ensures that the username field does not contain spaces using a field validator:
from pydantic import field_validator
from agency_swarm import BaseTool

class User(BaseTool):
    username: str

    @field_validator('username')
    @classmethod
    def validate_username(cls, value):
        if ' ' in value:
            raise ValueError('Username must not contain spaces.')
        return value

LLM Validator (v0.x)

The llm_validator validates outputs against specified natural language rules. (v0.x) Example:
from agency_swarm.tools.send_message import SendMessage
from agency_swarm.util.validators import llm_validator
from pydantic import model_validator

class SendMessageLLMValidation(SendMessage):
    @model_validator(mode='after')
    def validate_recipient(self):
        if self.recipient == "CustomerSupportAgent":
            llm_validator(
                statement="The message is related to customer support."
            )(self.message)
        return self
In this example, the llm_validator will throw an error if the message is not related to customer support. The caller agent will then have to fix the recipient or the message and send it again.
Since llm_validator uses LLMs for validation, it may incur additional costs and latency due to extra API calls. Use it for fields that require complex validation beyond simple checks.
By combining all the validators described above, you can create validation logic to ensure your agents and tools perform reliably.