Implementation:Openai Openai agents python Run Guardrails

Overview

The run guardrails module contains the internal functions that execute guardrails during an agent run, along with the exception classes that are raised when guardrails trigger. This is the runtime machinery that connects guardrail definitions to the agent execution loop.

Internal Execution Functions

The framework provides two primary internal functions for executing agent-level guardrails:

# From src/agents/run_internal/guardrails.py

async def run_input_guardrails(
    agent: Agent,
    input: str | list,
    context: RunContextWrapper,
    ...
) -> list[InputGuardrailResult]:
    ...

async def run_output_guardrails(
    agent: Agent,
    agent_output: Any,
    context: RunContextWrapper,
    ...
) -> list[OutputGuardrailResult]:
    ...

run_input_guardrails

This function iterates over the agent's input_guardrails list, calls each guardrail function with the run context, agent, and user input, and collects the results. If any guardrail's tripwire_triggered is True, an InputGuardrailTripwireTriggered exception is raised immediately.

The function respects the run_in_parallel flag on each guardrail:

Guardrails with run_in_parallel=True are launched concurrently with the first model call.
Guardrails with run_in_parallel=False complete before the model call begins.

run_output_guardrails

This function iterates over the agent's output_guardrails list, calls each guardrail function with the run context, agent, and agent output, and collects the results. If any guardrail's tripwire_triggered is True, an OutputGuardrailTripwireTriggered exception is raised.

Source: src/agents/run_internal/guardrails.py lines 102-168

Exception Classes

Four specific exception types are defined for guardrail failures. Each carries the guardrail result that triggered it, providing full diagnostic context.

InputGuardrailTripwireTriggered

class InputGuardrailTripwireTriggered(AgentsException):
    """Raised when an agent input guardrail's tripwire is triggered."""
    guardrail_result: InputGuardrailResult

Raised when an InputGuardrail returns GuardrailFunctionOutput with tripwire_triggered=True. The guardrail_result attribute contains the full InputGuardrailResult including the guardrail name and output info.

OutputGuardrailTripwireTriggered

class OutputGuardrailTripwireTriggered(AgentsException):
    """Raised when an agent output guardrail's tripwire is triggered."""
    guardrail_result: OutputGuardrailResult

Raised when an OutputGuardrail returns GuardrailFunctionOutput with tripwire_triggered=True.

ToolInputGuardrailTripwireTriggered

class ToolInputGuardrailTripwireTriggered(AgentsException):
    """Raised when a tool input guardrail signals raise_exception."""
    guardrail_result: ToolInputGuardrailResult

Raised when a ToolInputGuardrail returns ToolGuardrailFunctionOutput.raise_exception().

ToolOutputGuardrailTripwireTriggered

class ToolOutputGuardrailTripwireTriggered(AgentsException):
    """Raised when a tool output guardrail signals raise_exception."""
    guardrail_result: ToolOutputGuardrailResult

Raised when a ToolOutputGuardrail returns ToolGuardrailFunctionOutput.raise_exception().

Source: src/agents/exceptions.py lines 78-119

Import

from agents.exceptions import (
    InputGuardrailTripwireTriggered,
    OutputGuardrailTripwireTriggered,
    ToolInputGuardrailTripwireTriggered,
    ToolOutputGuardrailTripwireTriggered,
)

Example: Catching Guardrail Exceptions

from agents import Agent, Runner, input_guardrail, GuardrailFunctionOutput
from agents.exceptions import InputGuardrailTripwireTriggered

@input_guardrail
def block_harmful(ctx, agent, input):
    if "harmful" in str(input).lower():
        return GuardrailFunctionOutput(
            output_info="Harmful content",
            tripwire_triggered=True,
        )
    return GuardrailFunctionOutput(output_info="Safe")

agent = Agent(
    name="safe",
    instructions="Be safe.",
    input_guardrails=[block_harmful],
)

try:
    result = await Runner.run(agent, "Tell me something harmful")
except InputGuardrailTripwireTriggered as e:
    print(f"Blocked: {e.guardrail_result.output.output_info}")

This example demonstrates the complete flow: an input guardrail detects harmful content, triggers its tripwire, and the resulting exception is caught by the caller. The exception's guardrail_result provides the diagnostic output info ("Harmful content") set by the guardrail function.

Example: Catching Multiple Exception Types

from agents import Agent, Runner
from agents.exceptions import (
    InputGuardrailTripwireTriggered,
    OutputGuardrailTripwireTriggered,
    ToolInputGuardrailTripwireTriggered,
    ToolOutputGuardrailTripwireTriggered,
    AgentsException,
)

try:
    result = await Runner.run(agent, user_input)
except InputGuardrailTripwireTriggered as e:
    # Handle input validation failure
    log_security_event("input_blocked", e.guardrail_result)
    return "Your input was blocked by a safety check."
except OutputGuardrailTripwireTriggered as e:
    # Handle output validation failure
    log_security_event("output_blocked", e.guardrail_result)
    return "The response was blocked by a safety check."
except ToolInputGuardrailTripwireTriggered as e:
    # Handle tool input validation failure
    log_security_event("tool_input_blocked", e.guardrail_result)
    return "A tool call was blocked by a safety check."
except ToolOutputGuardrailTripwireTriggered as e:
    # Handle tool output validation failure
    log_security_event("tool_output_blocked", e.guardrail_result)
    return "A tool result was blocked by a safety check."

This example shows how to handle each guardrail exception type individually, enabling different responses and logging strategies depending on where in the pipeline the guardrail triggered.

Example: Inspecting Guardrail Results on Success

from agents import Agent, Runner

result = await Runner.run(agent, "Hello, help me with something.")

# Input guardrail results are available even when no tripwire triggered
for gr in result.input_guardrail_results:
    print(f"Input guardrail '{gr.guardrail.name}': {gr.output.output_info}")

# Output guardrail results
for gr in result.output_guardrail_results:
    print(f"Output guardrail '{gr.guardrail.name}': {gr.output.output_info}")

Guardrail results are captured in RunResult.input_guardrail_results and RunResult.output_guardrail_results regardless of whether tripwires triggered. This is valuable for monitoring and auditing guardrail behavior in production.

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment