Output Guardrails - Agency Swarm

Output guardrails validate agent responses before they reach users or other agents. When a guardrail trips, the agent receives feedback and retries.

Function Signature

Each output guardrail receives three parameters:

from agency_swarm import Agent, GuardrailFunctionOutput, RunContextWrapper, output_guardrail
from pydantic import BaseModel

@output_guardrail
async def my_output_guardrail(
    context: RunContextWrapper,
    agent: Agent,
    response_text: str | BaseModel,
) -> GuardrailFunctionOutput:
    """Validate agent output."""
    return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)

Parameters:

context: Run context wrapper with access to shared state.
agent: The Agent instance generating the response.
response_text: The agent response as a string, or a structured model when output_type is set.

Return:

GuardrailFunctionOutput with:
- tripwire_triggered (bool): True if validation failed.
- output_info (str): Feedback message sent to the agent when tripwire_triggered=True.

Basic Output Guardrail

from agency_swarm import Agent, GuardrailFunctionOutput, RunContextWrapper, output_guardrail

@output_guardrail
async def response_content_guardrail(
    context: RunContextWrapper, agent: Agent, response_text: str
) -> GuardrailFunctionOutput:
    tripwire_triggered = "bad word" in response_text.lower()
    output_info = "Please avoid using inappropriate language." if tripwire_triggered else ""
    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],
)

Practical Example: Preventing Sensitive Information Leaks

from agency_swarm import Agent, GuardrailFunctionOutput, RunContextWrapper, output_guardrail

@output_guardrail(name="ForbidSensitiveEmail")
async def forbid_sensitive_email(
    context: RunContextWrapper, agent: Agent, response_text: str
) -> GuardrailFunctionOutput:
    if "@" in response_text:
        return GuardrailFunctionOutput(
            output_info="Do not share email addresses. Offer to connect via the support portal instead.",
            tripwire_triggered=True,
        )
    return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)

support_agent = Agent(
    name="SupportPilot",
    instructions="You handle customer support. Official email: support@example.com.",
    model="gpt-5.4-mini",
    output_guardrails=[forbid_sensitive_email],
    validation_attempts=1,
)

See the full example at examples/guardrails_output.py.

Example: Simple Format Enforcement

import json
from agency_swarm import GuardrailFunctionOutput, RunContextWrapper, output_guardrail

@output_guardrail(name="RequireJSONFormat")
async def require_json_format(
    context: RunContextWrapper, agent: Agent, response_text: str
) -> GuardrailFunctionOutput:
    try:
        json.loads(response_text)
        return GuardrailFunctionOutput(output_info="", tripwire_triggered=False)
    except json.JSONDecodeError:
        return GuardrailFunctionOutput(
            output_info="Response must be valid JSON. Wrap your response in curly braces.",
            tripwire_triggered=True,
        )

Output Guardrail Retry Flow

When an output guardrail trips, the agent gets multiple chances to fix its response. The validation_attempts parameter controls this behavior.

How Retry Works

Agent generates response

The agent produces its initial response.

Output guardrail checks response

Each output guardrail validates the response.

If validation fails

The agent receives a system message containing the guardrail output_info.

Agent retries

The agent generates a new response, informed by that message.

Repeat until success or limit reached

This cycle continues up to validation_attempts times.

If all attempts fail

OutputGuardrailTripwireTriggered is raised.

Configure `validation_attempts`

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

Setting	Behavior
`validation_attempts=0`	Fail-fast (no retry, immediate exception)
`validation_attempts=1`	Default (one retry after initial failure)
`validation_attempts=2+`	Multiple retries for more complex validations

Each retry sends the guardrail output_info message to the agent as a system message, giving the agent context to adjust its response.

Handling Validation Failures

from agency_swarm import OutputGuardrailTripwireTriggered

try:
    response = await agency.get_response("Hello!")
except OutputGuardrailTripwireTriggered as exc:
    print(f"Validation failed: {exc.guardrail_result.output_info}")

Message History

Output guardrail failures are stored as system messages with message_origin="output_guardrail_error". For most use cases, role, content, and message_origin are enough. Extra metadata is mainly for debugging and run tracing.

Origin	Meaning
`output_guardrail_error`	Output guardrail failure (system message)

Example history entry (with debug metadata)

{
  "role": "system",
  "content": "You are not allowed to include your email address in your response. Ask agent to redirect user to the contact page: https://www.example.com/contact",
  "message_origin": "output_guardrail_error",
  "agent": "DatabaseAgent",
  "callerAgent": "CustomerSupportAgent",
  "agent_run_id": "agent_run_id",
  "parent_run_id": "call_id",
  "timestamp": 1758103770629217,
  "type": "message"
}

Agent-to-Agent Validation

Use guardrails to control how agents communicate with each other. When adding communication flows between agents, the recipient agent’s guardrails define the message format.

Example: Task/Response contract between agents

from agency_swarm import Agency, Agent, GuardrailFunctionOutput, RunContextWrapper, input_guardrail, output_guardrail

@input_guardrail(name="RequireTaskPrefix")
async def require_task_prefix(
    context: RunContextWrapper, agent: Agent, agent_input: str | list[str]
) -> GuardrailFunctionOutput:
    text = agent_input if isinstance(agent_input, str) else " ".join(agent_input)
    blocked = not text.startswith("Task:")
    return GuardrailFunctionOutput(
        output_info="ERROR: Requests to this agent must begin with 'Task:'" if blocked else "",
        tripwire_triggered=blocked,
    )

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

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],
    raise_input_guardrail_error=True,
)

agency = Agency(ceo, communication_flows=[(ceo, worker)])

In this example:

If the CEO sends a message that does not start with Task:, the worker input guardrail triggers.
The CEO receives an error and adjusts its message.
The worker output guardrail enforces Response: in returned messages.

Agent-to-agent messages are always single strings, so input guardrails for inter-agent communication receive a string (not a list).

​Function Signature

​Basic Output Guardrail

​Practical Example: Preventing Sensitive Information Leaks

​Example: Simple Format Enforcement

​Output Guardrail Retry Flow

​How Retry Works

​Configure validation_attempts

​Handling Validation Failures

​Message History

​Agent-to-Agent Validation

Function Signature

Basic Output Guardrail

Practical Example: Preventing Sensitive Information Leaks

Example: Simple Format Enforcement

Output Guardrail Retry Flow

How Retry Works

Configure `validation_attempts`

Handling Validation Failures

Message History

Agent-to-Agent Validation