Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions examples/exceptions/agents_exception.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
from __future__ import annotations
import asyncio
from agents import Agent, Runner, function_tool
from agents.exceptions import AgentsException

"""
This example demonstrates the use of the OpenAI Agents SDK with tools and comprehensive error handling.

The agent, 'TriageAgent', is configured to handle two tasks:
- Fetching weather information for a specified city using the `get_weather` tool.
- Adding two numbers using the `sum_numbers` tool.

The agent is instructed to use only one tool per execution cycle and can switch to another tool in subsequent cycles. The example sets a `max_turns=1` limit to intentionally restrict the agent to a single turn, which may trigger a `MaxTurnsExceeded` error if the agent attempts multiple tool calls.

Error handling is implemented with `AgentsException`, which is the base class for all SDK-related exceptions, including:
- `MaxTurnsExceeded`: Raised when the run exceeds the `max_turns` specified in the run methods.
- `ModelBehaviorError`: Raised when the model produces invalid outputs, e.g., malformed JSON or using non-existent tools.
- `UserError`: Raised when the SDK user makes an error in code implementation.
- `InputGuardrailTripwireTriggered`: Raised when an input guardrail is violated (e.g., invalid or off-topic input).
- `OutputGuardrailTripwireTriggered`: Raised when an output guardrail is violated (e.g., invalid tool output).

Although this example does not include explicit guardrails, the structure supports adding input/output guardrails to validate user inputs or tool outputs. The `AgentsException` catch block ensures all SDK-related errors are handled gracefully.
"""

@function_tool
def get_weather(city: str) -> str:
"""Returns weather info for the specified city."""
return f"The weather in {city} is sunny"

@function_tool
def sum_numbers(a: int, b: int) -> int:
"""Adds two numbers."""
return a + b

agent = Agent(
name="TriageAgent",
instructions="Get weather or sum numbers. You can use one tool at a time, switching to another tool in subsequent turns.",
tools=[sum_numbers, get_weather],
)

async def main():
try:
result = await Runner.run(
agent, "tell me karachi weather and sum 2+2 ans", max_turns=1
)
print(result.final_output)
except AgentsException as e:
print(f"Caught AgentsException: {e}")

if __name__ == "__main__":
asyncio.run(main())
44 changes: 44 additions & 0 deletions examples/exceptions/input_guardrail_tripwire_triggered.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
from __future__ import annotations
import asyncio
from pydantic import BaseModel
from agents import Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered, Runner, input_guardrail
"""
This example demonstrates an OpenAI Agents SDK agent with an input guardrail to block math homework queries.

The 'CustomerSupportAgent' processes user queries provided as direct string inputs in an interactive loop. A guardrail, implemented via 'GuardrailAgent' and a Pydantic model (`MathHomeworkOutput`), checks if the input is a math homework question. If detected, the guardrail raises `InputGuardrailTripwireTriggered`, triggering a refusal message. Otherwise, the agent responds to the query. The loop continues to prompt for new inputs, handling each independently.
"""
class MathHomeworkOutput(BaseModel):
is_math_homework: bool

guardrail_agent = Agent(
name="GuardrailAgent",
instructions="Check if the input is a math homework question.",
output_type=MathHomeworkOutput,
)

@input_guardrail
async def math_guardrail(context: Runner, agent: Agent, input: str) -> GuardrailFunctionOutput:
result = await Runner.run(guardrail_agent, input)
output = result.final_output_as(MathHomeworkOutput)
return GuardrailFunctionOutput(
output_info=output,
tripwire_triggered=output.is_math_homework,
)

async def main():
agent = Agent(
name="CustomerSupportAgent",
instructions="Answer user queries.",
input_guardrails=[math_guardrail],
)

while True:
user_input = input("Enter a message: ")
try:
result = await Runner.run(agent, user_input)
print(result.final_output)
except InputGuardrailTripwireTriggered:
print("Sorry, I can't help with math homework.")

if __name__ == "__main__":
asyncio.run(main())
37 changes: 37 additions & 0 deletions examples/exceptions/max_turns_exceeded.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
from __future__ import annotations
import asyncio
from agents import Agent, Runner, function_tool
from agents.exceptions import MaxTurnsExceeded
"""
This example demonstrates an OpenAI Agents SDK agent that triggers a MaxTurnsExceeded error.

The 'TriageAgent' handles user queries using tools for fetching weather (`get_weather`) or adding numbers (`sum_numbers`). The agent is instructed to use one tool per execution cycle. With `max_turns=1`, attempting to process multiple tasks (e.g., weather and sum) in one input causes a `MaxTurnsExceeded` error. The interactive loop processes user queries as direct string inputs, catching and displaying the `MaxTurnsExceeded` error message.
"""
@function_tool
def get_weather(city: str) -> str:
"""Returns weather info for the specified city."""
return f"The weather in {city} is sunny"

@function_tool
def sum_numbers(a: int, b: int) -> int:
"""Adds two numbers."""
return a + b

async def main():
agent = Agent(
name="TriageAgent",
instructions="Get weather or sum numbers. Use one tool at a time, switching to another tool in subsequent turns.",
tools=[sum_numbers, get_weather],
)

while True:
user_input = input("Enter a message: ")
try:
result = await Runner.run(agent, user_input, max_turns=1)
print(result.final_output)
except MaxTurnsExceeded as e:
print(f"Error: {e}")
break

if __name__ == "__main__":
asyncio.run(main())
35 changes: 35 additions & 0 deletions examples/exceptions/model_behavior_error.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@

from __future__ import annotations
import asyncio
from typing import Literal
from pydantic import BaseModel
from agents import Agent, Runner
from agents.exceptions import ModelBehaviorError

"""
This example demonstrates an OpenAI Agents SDK agent that triggers a ModelBehaviorError due to invalid model output.

The 'MiniErrorBot' agent uses a Pydantic model (`Output`) requiring a `value` field with the literal 'EXPECTED_VALUE'. The instructions tell the model to return 'Hello', causing a `ModelBehaviorError` when the output fails validation. The interactive loop processes user queries as direct string inputs, catching and displaying the `ModelBehaviorError` message.
"""

class Output(BaseModel):
value: Literal["EXPECTED_VALUE"]

async def main():
agent = Agent(
name="MiniErrorBot",
instructions="Just say: Hello",
output_type=Output,
)

while True:
user_input = input("Enter a message: ")
try:
result = await Runner.run(agent, user_input)
print(result.final_output)
except ModelBehaviorError as e:
print(f"ModelBehaviorError: {e}")
break

if __name__ == "__main__":
asyncio.run(main())
45 changes: 45 additions & 0 deletions examples/exceptions/output_guardrail_tripwire_triggered.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
from __future__ import annotations
import asyncio
from pydantic import BaseModel
from agents import Agent, GuardrailFunctionOutput, OutputGuardrailTripwireTriggered, Runner, output_guardrail
"""
This example demonstrates an OpenAI Agents SDK agent with an output guardrail to block math homework responses.

The 'Assistant' agent processes user queries provided as direct string inputs in an interactive loop. An output guardrail, using a Pydantic model (`MathHomeworkOutput`) and a guardrail agent, checks if the response is a math homework answer. If detected, the guardrail raises `OutputGuardrailTripwireTriggered`, and a refusal message is printed. The loop continues to prompt for new inputs, handling each independently.
"""

class MathHomeworkOutput(BaseModel):
is_math_homework: bool

guardrail_agent = Agent(
name="GuardrailAgent",
instructions="Check if the output is a math homework answer.",
output_type=MathHomeworkOutput,
)

@output_guardrail
async def math_guardrail(context: Runner, agent: Agent, output: str) -> GuardrailFunctionOutput:
result = await Runner.run(guardrail_agent, output)
output_data = result.final_output_as(MathHomeworkOutput)
return GuardrailFunctionOutput(
output_info=output_data,
tripwire_triggered=output_data.is_math_homework,
)

async def main():
agent = Agent(
name="Assistant",
instructions="Answer user queries.",
output_guardrails=[math_guardrail],
)

while True:
user_input = input("Enter a message: ")
try:
result = await Runner.run(agent, user_input)
print(result.final_output)
except OutputGuardrailTripwireTriggered:
print("Sorry, I can't provide math homework answers.")

if __name__ == "__main__":
asyncio.run(main())
35 changes: 35 additions & 0 deletions examples/exceptions/user_error.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
from __future__ import annotations
import asyncio
from agents import Agent, Runner, function_tool
from agents.exceptions import UserError

"""
This example demonstrates an OpenAI Agents SDK agent that triggers a UserError due to incorrect SDK usage.

The 'Assistant' agent is configured with an invalid tool (`invalid_tool`) that declares a `None` return type but returns a string, causing a `UserError` when the agent is executed. The error indicates improper tool configuration by the user. The interactive loop processes user queries as direct string inputs, catching and displaying the `UserError` message.
"""

@function_tool
def invalid_tool() -> None:
return "I return a string" # Type mismatch triggers UserError


async def main():
agent = Agent(
name="Assistant",
instructions="Use the invalid_tool to process queries.",
tools=[invalid_tool],
)

while True:
user_input = input("Enter a message: ")
try:
result = await Runner.run(agent, user_input)
print(result.final_output)
except UserError as e:
print(f"UserError: {e}")
break


if __name__ == "__main__":
asyncio.run(main())
Loading