---
title: "Output Guardrails"
description: "Validate agent responses before they reach users."
icon: "arrow-right-from-bracket"
---
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:
```python
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
```python
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
```python
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`](https://github.com/VRSEN/agency-swarm/blob/main/examples/guardrails_output.py).
## Example: Simple Format Enforcement
```python
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
The agent produces its initial response.
Each output guardrail validates the response.
The agent receives a **system message** containing the guardrail `output_info`.
The agent generates a new response, informed by that message.
This cycle continues up to `validation_attempts` times.
`OutputGuardrailTripwireTriggered` is raised.
## Configure `validation_attempts`
```python
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
```python
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) |
```json
{
"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.
```python
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).