Implementation:Openai Openai agents python InputGuardrail OutputGuardrail
Overview
The InputGuardrail and OutputGuardrail classes, along with their decorator shortcuts and the GuardrailFunctionOutput result type, provide the API for defining agent-level guardrails. These guardrails validate the user's input before the LLM processes it and the agent's final output after the LLM produces a response.
Class: InputGuardrail
@dataclass
class InputGuardrail(Generic[TContext]):
guardrail_function: Callable[
[RunContextWrapper[TContext], Agent[Any], str | list[TResponseInputItem]],
MaybeAwaitable[GuardrailFunctionOutput],
]
name: str | None = None
run_in_parallel: bool = True
The InputGuardrail dataclass wraps a function that validates the user's input. The function signature receives three arguments:
ctx-- ARunContextWrapperproviding access to the run context, including any custom context data.agent-- TheAgentinstance that the input is being validated for.input-- The user's input, either as astror as a list ofTResponseInputItemobjects.
The run_in_parallel field (default True) controls whether this guardrail executes concurrently with the first model call or blocks the model until validation completes.
Class: OutputGuardrail
@dataclass
class OutputGuardrail(Generic[TContext]):
guardrail_function: Callable[
[RunContextWrapper[TContext], Agent[Any], Any],
MaybeAwaitable[GuardrailFunctionOutput],
]
name: str | None = None
The OutputGuardrail dataclass wraps a function that validates the agent's final output. The function signature receives three arguments:
ctx-- ARunContextWrapperproviding access to the run context.agent-- TheAgentinstance that produced the output.output-- The agent's final output (type depends on the agent'soutput_typeconfiguration).
Output guardrails do not have a run_in_parallel flag because they inherently execute after the model has already produced its response.
Class: GuardrailFunctionOutput
@dataclass
class GuardrailFunctionOutput:
output_info: Any
tripwire_triggered: bool = False
Both input and output guardrails return a GuardrailFunctionOutput to signal their result:
output_info-- Arbitrary data describing the guardrail's assessment. This is captured in the guardrail result and available for logging and debugging. Typically a short string explanation.tripwire_triggered-- A boolean indicating whether the guardrail should halt execution. WhenTrue, the framework raises the corresponding exception (InputGuardrailTripwireTriggeredorOutputGuardrailTripwireTriggered). Defaults toFalse.
Decorators
def input_guardrail(func=None, *, name=None, run_in_parallel=True) -> InputGuardrail: ...
def output_guardrail(func=None, *, name=None) -> OutputGuardrail: ...
The decorators provide a concise syntax for creating guardrail instances from functions:
# Without arguments
@input_guardrail
def my_input_check(ctx, agent, input):
...
# With arguments
@input_guardrail(name="topic_filter", run_in_parallel=False)
def my_input_check(ctx, agent, input):
...
# Output guardrail
@output_guardrail
def my_output_check(ctx, agent, output):
...
# Output guardrail with name
@output_guardrail(name="length_check")
def my_output_check(ctx, agent, output):
...
Source: src/agents/guardrail.py lines 72-343
Import
from agents import (
InputGuardrail,
OutputGuardrail,
GuardrailFunctionOutput,
input_guardrail,
output_guardrail,
)
Example: Input and Output Guardrails on an Agent
from agents import Agent, input_guardrail, output_guardrail, GuardrailFunctionOutput
@input_guardrail
def check_topic(ctx, agent, input):
"""Reject off-topic inputs."""
if isinstance(input, str) and "unrelated" in input.lower():
return GuardrailFunctionOutput(
output_info="Off topic",
tripwire_triggered=True,
)
return GuardrailFunctionOutput(output_info="OK")
@output_guardrail
def check_output_length(ctx, agent, output):
"""Ensure output is not too long."""
if isinstance(output, str) and len(output) > 10000:
return GuardrailFunctionOutput(
output_info="Too long",
tripwire_triggered=True,
)
return GuardrailFunctionOutput(output_info="OK")
agent = Agent(
name="assistant",
instructions="Be helpful.",
input_guardrails=[check_topic],
output_guardrails=[check_output_length],
)
Example: Sequential Input Guardrail
from agents import input_guardrail, GuardrailFunctionOutput
@input_guardrail(run_in_parallel=False)
def expensive_safety_check(ctx, agent, input):
"""Block model call until safety check completes."""
# This guardrail runs before the model call starts
risk_score = compute_risk_score(str(input))
if risk_score > 0.9:
return GuardrailFunctionOutput(
output_info=f"High risk score: {risk_score}",
tripwire_triggered=True,
)
return GuardrailFunctionOutput(output_info=f"Risk score: {risk_score}")
Setting run_in_parallel=False ensures the model call does not start until this guardrail completes. This is useful when the guardrail is expensive and you want to avoid wasting model compute on inputs that will be rejected.
Example: Using Class Constructors
from agents import Agent, InputGuardrail, OutputGuardrail, GuardrailFunctionOutput
def my_input_check(ctx, agent, input):
if "forbidden" in str(input).lower():
return GuardrailFunctionOutput(
output_info="Forbidden content",
tripwire_triggered=True,
)
return GuardrailFunctionOutput(output_info="OK")
def my_output_check(ctx, agent, output):
if len(str(output)) > 5000:
return GuardrailFunctionOutput(
output_info="Output too long",
tripwire_triggered=True,
)
return GuardrailFunctionOutput(output_info="OK")
agent = Agent(
name="assistant",
instructions="Be helpful.",
input_guardrails=[
InputGuardrail(guardrail_function=my_input_check, name="forbidden_check"),
],
output_guardrails=[
OutputGuardrail(guardrail_function=my_output_check, name="length_check"),
],
)
See Also
- Agent Level Guardrail Definition -- The principle behind agent-level guardrails
- Guardrail Attachment Pattern -- How to wire guardrails to tools and agents
- Run Guardrails -- The runtime pipeline that executes guardrails