update all

Author: muhammadhamidrazasidtechno

examples/exceptions/agents_exception.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+from agents import Agent, RunContextWrapper, Runner, function_tool
+from agents.exceptions import AgentsException
+import asyncio
+
+
+"""
+This example demonstrates the use of the OpenAI Agents SDK with tools and comprehensive error handling.
+
+The agent, 'Triage Agent', 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="Triage Agent",
+    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:
+        user_input = input("Enter a message: ")
+
+        result = await Runner.run(agent, user_input, max_turns=1)
+        print(result.final_output)
+    except AgentsException as e:
+        print(f"Caught AgentsException: {e}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/exceptions/input_guardrail_tripwire_triggered.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+import asyncio
+from pydantic import BaseModel
+from agents import (
+    Agent,
+    GuardrailFunctionOutput,
+    InputGuardrailTripwireTriggered,
+    Runner,
+    input_guardrail,
+)
+from agents.exceptions import AgentsException
+
+"""
+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 ("Sorry, I can't help with math homework."). 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, 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],
+    )
+
+    user_input = "What is 2 + 2"
+    try:
+        result = await Runner.run(agent, user_input)
+        print(result.final_output)
+    except InputGuardrailTripwireTriggered:
+        print("InputGuardrailTripwireTriggered, I can't help with math homework.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/exceptions/max_turns_exceeded.py

@@ -0,0 +1,42 @@
+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 instructions direct the agent to process both tasks in a single turn, but with `max_turns=1`, this 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="Process both get_weather and sum_numbers in a single turn when asked for both.",
+        tools=[sum_numbers, get_weather],
+        model=model,
+    )
+
+    user_input = "What is US Weather and sum 2 + 2."
+    try:
+        result = await Runner.run(agent, user_input, max_turns=1)
+        print(result.final_output)
+    except MaxTurnsExceeded as e:
+        print(f"Caught MaxTurnsExceeded: {e}")
+    
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/exceptions/model_behavior_error.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+import asyncio
+from pydantic import BaseModel
+from typing import Literal
+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,
+    )
+
+    user_input = "hello"
+    try:
+        result = await Runner.run(agent, user_input)
+        print(result.final_output)
+    except ModelBehaviorError as e:
+        print(f"ModelBehaviorError: {e}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/exceptions/output_guardrail_tripwire_triggered.py

@@ -0,0 +1,60 @@
+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, 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],
+    )
+
+    user_input = "What is 2 + 2"
+
+    try:
+        result = await Runner.run(agent, user_input)
+        print(result.final_output)
+    except OutputGuardrailTripwireTriggered:
+        print(
+            "OutputGuardrailTripwireTriggered, I can't provide math homework answers."
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/exceptions/user_error.py

@@ -0,0 +1,36 @@
+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_use_behavior` (empty string) and an invalid tool (`invalid_tool`) that declares a `None` return type but returns a string. Either issue raises a `UserError` when the agent is executed, indicating improper SDK 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],
+        tool_use_behavior="invalid_tool",
+        model=model
+    )
+    user_input = "Do Something."
+    try:
+        result = await Runner.run(agent, user_input)
+        print(result.final_output)
+    except UserError as e:
+        print(f"UserError: {e}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())