Guarding the desk's input and output

An input guardrail is a check that runs before the model and stops the run when its tripwire fires, so a message with a password never reaches the desk.

Last updated: 28 Sep, 2026 · openai-agents 0.22.3

The desk from The desk agent and its tools answers whatever it is asked. A guardrail, from Input guardrails and the tripwire, sits in front of it and refuses input you do not want the model to see.

The pieces, and where each came from

Piece	From lesson
The desk with its lookup tool	The desk agent and its tools
@input_guardrail returning GuardrailFunctionOutput	Input guardrails and the tripwire
InputGuardrailTripwireTriggered on a block	Input guardrails and the tripwire

The password check

A guardrail is a function with @input_guardrail. It returns a GuardrailFunctionOutput; setting tripwire_triggered=True stops the run.

python

@input_guardrail
async def no_passwords(ctx, agent, user_input) -> GuardrailFunctionOutput:
    text = user_input if isinstance(user_input, str) else str(user_input)
    tripped = "password" in text.lower()   # True blocks the run
    return GuardrailFunctionOutput(output_info={"blocked": tripped},
                                   tripwire_triggered=tripped)

The guarded desk

The check goes in the desk's input_guardrails list, so it runs before the model on each run.

python

desk = Agent(
    name="Shop desk",
    instructions="Help shoppers with their orders.",
    tools=[lookup_order],
    input_guardrails=[no_passwords],
    model=ShopModel(),
)

A safe run and a blocked run

A safe message answers as before; a message with a password raises InputGuardrailTripwireTriggered, which you catch.

python

print("SAFE:", Runner.run_sync(desk, "Where is my order A17?").final_output)
try:
    Runner.run_sync(desk, "My password is hunter2, refund me now")
except InputGuardrailTripwireTriggered:
    print("BLOCKED: the input guardrail tripped, the model never ran")

Project files used on this pageThis lesson builds on a project from earlier lessons. The code below imports this file. Click a file to see its code, or follow the link to the lesson that wrote it. To run the code yourself, keep it in the same folder.

View the code here

shop_model.py

"""A deterministic stand-in Model for the OpenAI Agents SDK course.

It implements the Model interface so an Agent runs with no API key. It reads the
last user message and the tool results out of `input`, and returns either a tool
call, a handoff, or a final message. Swap it for a real model at the end.
"""
from agents.models.interface import Model
from agents.items import ModelResponse
from agents.usage import Usage
from openai.types.responses import (
    ResponseOutputMessage, ResponseOutputText, ResponseFunctionToolCall,
    ResponseCompletedEvent, ResponseTextDeltaEvent, Response,
)


def _message(text):
    return ResponseOutputMessage(
        id="msg", role="assistant", type="message", status="completed",
        content=[ResponseOutputText(text=text, type="output_text", annotations=[])],
    )


def _tool_call(name, arguments, call_id="call_1"):
    return ResponseFunctionToolCall(
        id="fc", call_id=call_id, name=name, arguments=arguments, type="function_call",
    )


def last_user_text(input):
    if isinstance(input, str):
        return input
    for item in reversed(input):
        d = item if isinstance(item, dict) else item.__dict__
        if d.get("role") == "user":
            content = d.get("content")
            if isinstance(content, str):
                return content
            if isinstance(content, list):
                for part in content:
                    pd = part if isinstance(part, dict) else part.__dict__
                    if pd.get("text"):
                        return pd["text"]
    return ""


def tool_output(input):
    if isinstance(input, str):
        return None
    for item in reversed(input):
        d = item if isinstance(item, dict) else item.__dict__
        if d.get("type") == "function_call_output":
            return d.get("output")
    return None


class ShopModel(Model):
    async def get_response(self, system_instructions, input, model_settings, tools,
                           output_schema, handoffs, tracing, *, previous_response_id=None,
                           conversation_id=None, prompt=None):
        result = tool_output(input)
        if result is not None:
            # A handoff transfer looks like {"assistant": "..."}; the specialist answers for real.
            if result.strip().startswith('{"assistant"'):
                if "refund" in (system_instructions or "").lower():
                    return ModelResponse(
                        output=[_message(
                            "Your refund is approved and will be processed in 5 to 7 days.")],
                        usage=Usage(), response_id=None)
                return ModelResponse(output=[_message("Handled by the specialist.")],
                                     usage=Usage(), response_id=None)
            return ModelResponse(output=[_message(result)], usage=Usage(), response_id=None)
        text = last_user_text(input).lower()
        if handoffs and "refund" in text:
            return ModelResponse(output=[_tool_call(handoffs[0].tool_name, "{}")],
                                 usage=Usage(), response_id=None)
        if tools and "order" in text:
            return ModelResponse(output=[_tool_call("lookup_order", '{"order_id": "A17"}')],
                                 usage=Usage(), response_id=None)
        return ModelResponse(output=[_message("How can I help with your order?")],
                             usage=Usage(), response_id=None)

    async def stream_response(self, system_instructions, input, model_settings, tools,
                              output_schema, handoffs, tracing, *, previous_response_id=None,
                              conversation_id=None, prompt=None):
        text = "How can I help with your order?"
        for i, word in enumerate(text.split()):
            yield ResponseTextDeltaEvent(
                type="response.output_text.delta", delta=word + " ",
                content_index=0, item_id="msg", output_index=0,
                sequence_number=i, logprobs=[],
            )
        response = Response(
            id="r", created_at=0.0, model="shop-standin", object="response",
            output=[_message(text)], parallel_tool_calls=False,
            tool_choice="auto", tools=[],
        )
        yield ResponseCompletedEvent(type="response.completed", response=response,
                                     sequence_number=99)

The desk refusing a password

One safe run and one blocked run, side by side.

Example

from agents import (Agent, Runner, function_tool, set_tracing_disabled,
    input_guardrail, GuardrailFunctionOutput, InputGuardrailTripwireTriggered,
    RunContextWrapper)
from shop_model import ShopModel
set_tracing_disabled(True)

@function_tool
def lookup_order(order_id: str) -> str:
    "Look up an order by its id."
    return f"Order {order_id}: shipped on 3 March, arriving 7 March."

@input_guardrail
async def no_passwords(ctx: RunContextWrapper, agent: Agent, user_input) -> GuardrailFunctionOutput:
    text = user_input if isinstance(user_input, str) else str(user_input)
    tripped = "password" in text.lower()
    return GuardrailFunctionOutput(output_info={"blocked": tripped}, tripwire_triggered=tripped)

desk = Agent(
    name="Shop desk",
    instructions="Help shoppers with their orders.",
    tools=[lookup_order],
    input_guardrails=[no_passwords],
    model=ShopModel(),
)

print("SAFE:", Runner.run_sync(desk, "Where is my order A17?").final_output)
try:
    Runner.run_sync(desk, "My password is hunter2, refund me now")
except InputGuardrailTripwireTriggered:
    print("BLOCKED: the input guardrail tripped, the model never ran")

Output

SAFE: Order A17: shipped on 3 March, arriving 7 March.
BLOCKED: the input guardrail tripped, the model never ran

What the guardrail did

The safe message passed the check, so the run reached the model and the tool answered normally.
The password message tripped the wire, so the SDK raised InputGuardrailTripwireTriggered before the model saw a single word.
Catching the exception is how your code turns a trip into a polite refusal instead of a crash.

Input guardrail vs output guardrail

Guardrail	When it runs, and on what
Input guardrail	Before the model, on the run's first agent, checking the incoming message
Output guardrail	After the model, on the final answer, checking what the agent produced

When to guard the desk

Refuse secrets or unsafe requests before the model reads them, so they never enter a prompt or a log.
Check the final answer with an output guardrail when a reply must never leak a price, a policy, or a rule.

Watch out. An input guardrail runs only on the first agent of a run, per the guardrails docs. After a handoff the specialist is not re-checked, so put the guard on the agent the run starts with.

Previous: A refund specialist by handoff
Next: The desk that remembers with a session
Reference: Guardrails

Try it yourself

Change the banned word from password to your own and watch a new message get blocked.
Remove the try/except and read the traceback the tripwire raises.
Return tripwire_triggered=False always and confirm the password message now reaches the model.

PreviousA refund specialist by handoff

You understood something today that you didn't yesterday.

Guarding the desk's input and output

The pieces, and where each came from

The password check

The guarded desk

A safe run and a blocked run

The desk refusing a password

What the guardrail did

Input guardrail vs output guardrail

When to guard the desk

Related