Build with AI: Create Agents with the OpenAI Agents SDK

Add Agent Guardrails.py

"""
Build with AI: Create Agents with the OpenAI Agents SDK
All examples use Python and the OpenAI client.

Prereqs:
  pip install -r requirements.txt
  export API_KEY = os.environ[...] or set the api_key to the client
"""

import os
import asyncio
import json

from openai import OpenAI
from dotenv import load_dotenv, find_dotenv
from agents import (
    Agent,
    Runner,
    ModelSettings,
    WebSearchTool,
    GuardrailFunctionOutput,
    InputGuardrail,
    InputGuardrailTripwireTriggered,
    Runner,
)
from pydantic import BaseModel, ValidationError

# read local .env file
_ = load_dotenv(find_dotenv())

# retrieve OpenAI API key
client = OpenAI(
    api_key="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"
)


# # ---------------------------------------------------------------------------
# # Add Agent Guardrails
# # ---------------------------------------------------------------------------
class TravelOutput(BaseModel):
    destination: str
    duration: str
    summary: str
    cost: str
    tips: str


# ---- Guardrail Agent ----
class BudgetCheckOutput(BaseModel):
    is_valid: bool
    reasoning: str


budget_guardrail_agent = Agent(
    name="Budget Guardrail",
    instructions=(
        "Decide if the user's travel request includes an unrealistic budget. "
        "If the request says things like '10 days in Caribbean for $200' or obviously too low "
        "for the destination and duration, set is_valid to False and explain why in reasoning. "
        "Otherwise set is_valid to True."
    ),
    output_type=BudgetCheckOutput,
)


async def budget_guardrail(ctx, agent, input_data):
    result = await Runner.run(budget_guardrail_agent, input_data, context=ctx.context)
    final_output = result.final_output_as(BudgetCheckOutput)
    print("Budget gaurdrail reasoning: ", final_output.reasoning)
    return GuardrailFunctionOutput(
        output_info=final_output, tripwire_triggered=not final_output.is_valid
    )


# ---- Planner Agent (builds day-by-day itinerary) ----
planner_agent = Agent(
    name="Planner Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user asks to plan or outline an itinerary, schedule, or daily plan.",
    instructions=(
        "You specialize in building day-by-day travel itineraries and sequencing activities. "
        'Always return JSON with this structure: {"destination":"string","duration":"string","summary":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "medium"}, extra_body={"text": {"verbosity": "low"}}
    ),
    tools=[WebSearchTool()],
)

# ---- Budget Agent (estimates costs under constraints) ----
budget_agent = Agent(
    name="Budget Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user mentions budget, price, cost, dollars, under $X, or asks 'how much'.",
    instructions=(
        "You estimate costs for lodging, food, transport, and activities at a high level; flag budget violations. "
        'Always return JSON with this structure: {"cost":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "medium"}, extra_body={"text": {"verbosity": "low"}}
    ),
)

# ---- Local Guide Agent (adds local tips & dining) ----
local_guide_agent = Agent(
    name="Local Guide Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user asks for food, restaurants, neighborhoods, local tips, or 'what's good nearby'.",
    instructions=(
        "You provide restaurants, neighborhoods, cultural tips, and current local highlights. "
        'Always return JSON with this structure: {"tips":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "medium"}, extra_body={"text": {"verbosity": "low"}}
    ),
    tools=[WebSearchTool()],
)

# ---- Core orchestrator: Travel Agent ----
travel_agent = Agent(
    name="Travel Agent",
    model="gpt-5.1",
    instructions=(
        "You are a friendly and knowledgeable travel planner that helps users plan trips, suggest destinations, and create detailed summaries of their journeys.\n"
        "Your primary role is to orchestrate other specialized agents (used as tools) to complete the user's request.\n"
        "\n"
        "When planning an itinerary, call the **Planner Agent** to create daily schedules, organize destinations, and recommend attractions or activities. Do not create itineraries yourself.\n"
        "When estimating costs, call the **Budget Agent** to calculate the total trip cost including flights, hotels, and activities. Do not calculate or estimate prices on your own.\n"
        "When recommending local experiences, restaurants, neighborhoods, or cultural highlights, call the **Local Guide Agent** to provide these insights. Do not generate local recommendations without this agent.\n"
        "\n"
        "Use these agents one at a time in a logical order based on the request — start with the Planner Agent, then the Budget Agent, and finally the Local Guide Agent.\n"
        "After receiving results from these agents, combine their outputs into a single structured summary.\n"
        "\n"
        "Return JSON output using this exact structure:\n"
        '{"destination": "string", "duration": "string", "summary": "string", "cost": "string", "tips": "string"}.\n'
    ),
    output_type=TravelOutput,
    model_settings=ModelSettings(
        reasoning={"effort": "medium"},  # minimal | low | medium | high
        extra_body={"text": {"verbosity": "low"}},  # low | medium | high
    ),
    tools=[
        WebSearchTool(),
        planner_agent.as_tool(
            tool_name="planner_agent",
            tool_description="plan or outline an itinerary, schedule, or daily plan",
        ),
        budget_agent.as_tool(
            tool_name="budget_agent", tool_description="calculates the cost of a trip"
        ),
        local_guide_agent.as_tool(
            tool_name="local_guide_agent",
            tool_description="provide restaurants, neighborhoods, cultural tips, and current local highlights",
        ),
    ],
    input_guardrails=[
        InputGuardrail(guardrail_function=budget_guardrail),
    ],
)


# --- Pretty print helper ----------------------------------------------------
def print_fields(data):
    if isinstance(data, str):
        try:
            data = TravelOutput(**json.loads(data))
        except (json.JSONDecodeError, ValidationError) as e:
            print("Raw output:", data)
            return
    print(f"Destination: {data.destination}")
    print(f"Duration: {data.duration}")
    print(f"Summary: {data.summary}")
    print(f"Cost: {data.cost}")
    print(f"Tips: {data.tips}")


async def main():
    try:
        result = await Runner.run(
            travel_agent,
            """I'm considering a trip to Jamaica sometime in late September or early October. 
                                                I'm pretty flexible with the exact dates, maybe around a week-long trip.     
                                                I'd like to get an idea of flight ticket prices and some well-located hotels.     
                                                I'm also a big foodie, so any recommendations for great local restaurants would be fantastic! 
                                                Do not ask follow-up questions.
                                                My budget is limited to $100.""",
        )
        print_fields(result.final_output)
    except InputGuardrailTripwireTriggered as e:
        print("\nGuardrail blocked this budget: ", e)
    except Exception as e:
        print("Error: ", e)


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

Create the Core Travel Agent.py

"""
Build with AI: Create Agents with the OpenAI Agents SDK
All examples use Python and the OpenAI client.

Prereqs:
  pip install -r requirements.txt
  export API_KEY = os.environ[...] or set the api_key to the client
"""

import os
import asyncio
import json

from openai import OpenAI
from dotenv import load_dotenv, find_dotenv
from agents import Agent, Runner, ModelSettings
from pydantic import BaseModel, ValidationError

# read local .env file
_ = load_dotenv(find_dotenv())

# retrieve OpenAI API key
client = OpenAI(
    api_key=""
)


# # ---------------------------------------------------------------------------
# # Create the Core Travel Agent
# # ---------------------------------------------------------------------------
class TravelOutput(BaseModel):
    destination: str
    duration: str
    summary: str


travel_agent = Agent(
    name="Travel Agent",
    model="gpt-5.1",
    instructions=(
        "You are a friendly and knowledgeable travel planner that helps users plan trips, "
        "suggests destinations, and creates brief summaries of their journeys. "
        "Always return your response as valid JSON matching this structure: "
        '{"destination": "string", "duration": "string", "summary": "string"}'
    ),
    output_type=TravelOutput,
    model_settings=ModelSettings(
        reasoning={"effort": "medium"},  # minimal | low | medium | high
        extra_body={"text": {"verbosity": "low"}},  # low | medium | high
    ),
)


# --- Pretty print helper ----------------------------------------------------
def print_fields(data):
    if isinstance(data, str):
        try:
            data = TravelOutput(**json.loads(data))
        except (json.JSONDecodeError, ValidationError) as e:
            print("Raw output:", data)
            return
    print(f"Destination: {data.destination}")
    print(f"Duration: {data.duration}")
    print(f"Summary: {data.summary}")


async def main():
    try:
        result = await Runner.run(
            travel_agent,
            "Plan a 3-day trip to Jamaica under $1500. "
            "Find uncommon places that are off the beaten path and little known.",
        )
        print_fields(result.final_output)
    except Exception as e:
        print("Error", e)


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

Extend the Agent with Tools.py

"""
Build with AI: Create Agents with the OpenAI Agents SDK
All examples use Python and the OpenAI client.

Prereqs:
  pip install -r requirements.txt
  export API_KEY = os.environ[...] or set the api_key to the client
"""

import os
import asyncio
import json

from openai import OpenAI
from dotenv import load_dotenv, find_dotenv
from agents import Agent, Runner, ModelSettings, WebSearchTool
from pydantic import BaseModel, ValidationError

# read local .env file
_ = load_dotenv(find_dotenv())

# retrieve OpenAI API key
client = OpenAI(
    api_key=""
)


# # ---------------------------------------------------------------------------
# # Extend the Agent with Tools
# # ---------------------------------------------------------------------------
class TravelOutput(BaseModel):
    destination: str
    duration: str
    summary: str
    source: str


travel_agent = Agent(
    name="Travel Agent",
    model="gpt-5.1",
    instructions=(
        "You are a friendly and knowledgeable travel planner that helps users plan trips, "
        "suggest destinations, and create brief summaries of their journeys. "
        "Use tools when helpful (e.g., Web Search). "
        "Cite your sources (i.e., the URL from which your data was obtained). "
        "Always return your response as valid JSON matching this structure: "
        '{"destination": "string", "duration": "string", "summary": "string", "source": "string"}'
    ),
    output_type=TravelOutput,
    model_settings=ModelSettings(
        reasoning={"effort": "medium"},  # minimal | low | medium | high
        extra_body={"text": {"verbosity": "low"}},  # low | medium | high
    ),
    tools=[WebSearchTool()],
)


# --- Pretty print helper ----------------------------------------------------
def print_fields(data):
    if isinstance(data, str):
        try:
            data = TravelOutput(**json.loads(data))
        except (json.JSONDecodeError, ValidationError) as e:
            print("Raw output:", data)
            return
    print(f"Destination: {data.destination}")
    print(f"Duration: {data.duration}")
    print(f"Summary: {data.summary}")
    print(f"Source: {data.source}")


async def main():
    try:
        result = await Runner.run(
            travel_agent,
            "Plan a 3-day trip to Jamaica under $1500. "
            "Find uncommon places that are off the beaten path and little known. Use web search to get the latest information.",
        )
        print_fields(result.final_output)
    except Exception as e:
        print("Error", e)


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

Maintain Agent Context with Sessions.py

"""
Build with AI: Create Agents with the OpenAI Agents SDK
All examples use Python and the OpenAI client.

Prereqs:
  pip install -r requirements.txt
  export API_KEY = os.environ[...] or set the api_key to the client
"""

import os
import asyncio
import json

from openai import OpenAI
from dotenv import load_dotenv, find_dotenv
from agents import (
    Agent,
    Runner,
    ModelSettings,
    WebSearchTool,
    GuardrailFunctionOutput,
    InputGuardrail,
    InputGuardrailTripwireTriggered,
    Runner,
    SQLiteSession,
)
from pydantic import BaseModel, ValidationError

# read local .env file
_ = load_dotenv(find_dotenv())

# retrieve OpenAI API key
client = OpenAI(
    api_key=""
)


# # ---------------------------------------------------------------------------
# # Maintain Agent Context with Sessions
# # ---------------------------------------------------------------------------
class TravelOutput(BaseModel):
    destination: str
    duration: str
    summary: str
    cost: str
    tips: str


# ---- Guardrail Agent ----
class BudgetCheckOutput(BaseModel):
    is_valid: bool
    reasoning: str


budget_guardrail_agent = Agent(
    name="Budget Guardrail",
    instructions=(
        "Decide if the user's travel request includes an unrealistic budget. "
        "If the request says things like '10 days in Caribbean for $200' or obviously too low "
        "for the destination and duration, set is_valid to False and explain why in reasoning. "
        "Otherwise set is_valid to True."
    ),
    output_type=BudgetCheckOutput,
)


async def budget_guardrail(ctx, agent, input_data):
    result = await Runner.run(budget_guardrail_agent, input_data, context=ctx.context)
    final_output = result.final_output_as(BudgetCheckOutput)
    return GuardrailFunctionOutput(
        output_info=final_output, tripwire_triggered=not final_output.is_valid
    )


# ---- Planner Agent (builds day-by-day itinerary) ----
planner_agent = Agent(
    name="Planner Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user asks to plan or outline an itinerary, schedule, or daily plan.",
    instructions=(
        "You specialize in building day-by-day travel itineraries and sequencing activities. "
        'Always return JSON with this structure: {"destination":"string","duration":"string","summary":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "low"}, extra_body={"text": {"verbosity": "low"}}
    ),
    tools=[WebSearchTool()],
)

# ---- Budget Agent (estimates costs under constraints) ----
budget_agent = Agent(
    name="Budget Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user mentions budget, price, cost, dollars, under $X, or asks 'how much'.",
    instructions=(
        "You estimate costs for lodging, food, transport, and activities at a high level; flag budget violations. "
        'Always return JSON with this structure: {"cost":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "low"}, extra_body={"text": {"verbosity": "low"}}
    ),
)

# ---- Local Guide Agent (adds local tips & dining) ----
local_guide_agent = Agent(
    name="Local Guide Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user asks for food, restaurants, neighborhoods, local tips, or 'what's good nearby'.",
    instructions=(
        "You provide restaurants, neighborhoods, cultural tips, and current local highlights. "
        'Always return JSON with this structure: {"tips":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "low"}, extra_body={"text": {"verbosity": "low"}}
    ),
    tools=[WebSearchTool()],
)

# ---- Core orchestrator: Travel Agent ----
travel_agent = Agent(
    name="Travel Agent",
    model="gpt-5.1",
    instructions=(
        "You are a friendly and knowledgeable travel planner that helps users plan trips, suggest destinations, and create detailed summaries of their journeys.\n"
        "Your primary role is to orchestrate other specialized agents (used as tools) to complete the user's request.\n"
        "\n"
        "When planning an itinerary, call the **Planner Agent** to create daily schedules, organize destinations, and recommend attractions or activities. Do not create itineraries yourself.\n"
        "When estimating costs, call the **Budget Agent** to calculate the total trip cost including flights, hotels, and activities. Do not calculate or estimate prices on your own.\n"
        "When recommending local experiences, restaurants, neighborhoods, or cultural highlights, call the **Local Guide Agent** to provide these insights. Do not generate local recommendations without this agent.\n"
        "\n"
        "Use these agents one at a time in a logical order based on the request — start with the Planner Agent, then the Budget Agent, and finally the Local Guide Agent.\n"
        "After receiving results from these agents, combine their outputs into a single structured summary.\n"
        "\n"
        "Return JSON output using this exact structure:\n"
        '{"destination": "string", "duration": "string", "summary": "string", "cost": "string", "tips": "string"}.\n'
    ),
    output_type=TravelOutput,
    model_settings=ModelSettings(
        reasoning={"effort": "low"},  # minimal | low | medium | high
        extra_body={"text": {"verbosity": "low"}},  # low | medium | high
    ),
    tools=[
        WebSearchTool(),
        planner_agent.as_tool(
            tool_name="planner_agent",
            tool_description="plan or outline an itinerary, schedule, or daily plan",
        ),
        budget_agent.as_tool(
            tool_name="budget_agent", tool_description="calculates the cost of a trip"
        ),
        local_guide_agent.as_tool(
            tool_name="local_guide_agent",
            tool_description="provide restaurants, neighborhoods, cultural tips, and current local highlights",
        ),
    ],
    input_guardrails=[
        InputGuardrail(guardrail_function=budget_guardrail),
    ],
)


# --- Pretty print helper ----------------------------------------------------
def print_fields(data):
    if isinstance(data, str):
        try:
            data = TravelOutput(**json.loads(data))
        except (json.JSONDecodeError, ValidationError) as e:
            print("Raw output:", data)
            return
    print(f"Destination: {data.destination}")
    print(f"Duration: {data.duration}")
    print(f"Summary: {data.summary}")
    print(f"Cost: {data.cost}")
    print(f"Tips: {data.tips}")


async def main():
    try:
        # In-memory database (lost when process ends)
        # Create a session instance with a session ID
        session = SQLiteSession("travel_agent_123")

        # First turn
        result = await Runner.run(
            travel_agent,
            """I'm considering a trip to Jamaica sometime in late September or early October.
               My budget is about $3,000. Do not ask follow-up questions.""",
            session=session,
        )
        print_fields(result.final_output)
        print("==================NEXT RUN=======================")

        # Second turn - agent automatically remembers previous context
        result = await Runner.run(
            travel_agent,
            "I actually think I'd like to go to the Bahamas instead. My budget is still the same.",
            session=session,
        )
        print_fields(result.final_output)
    except InputGuardrailTripwireTriggered as e:
        print("\nGuardrail blocked this budget: ", e)
    except Exception as e:
        print("Error: ", e)


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

Orchestrate Multiple Agents.py

"""
Build with AI: Create Agents with the OpenAI Agents SDK
All examples use Python and the OpenAI client.

Prereqs:
  pip install -r requirements.txt
  export API_KEY = os.environ[...] or set the api_key to the client
"""

import os
import asyncio
import json

from openai import OpenAI
from dotenv import load_dotenv, find_dotenv
from agents import Agent, Runner, ModelSettings, WebSearchTool
from pydantic import BaseModel, ValidationError

# read local .env file
_ = load_dotenv(find_dotenv())

# retrieve OpenAI API key
client = OpenAI(
    api_key=""
)


# # ---------------------------------------------------------------------------
# # Orchestrate Multiple Agents
# # ---------------------------------------------------------------------------
class TravelOutput(BaseModel):
    destination: str
    duration: str
    summary: str
    cost: str
    tips: str


# ---- Planner Agent (builds day-by-day itinerary) ----
planner_agent = Agent(
    name="Planner Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user asks to plan or outline an itinerary, schedule, or daily plan.",
    instructions=(
        "You specialize in building day-by-day travel itineraries and sequencing activities. "
        'Always return JSON with this structure: {"destination":"string","duration":"string","summary":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "medium"}, extra_body={"text": {"verbosity": "low"}}
    ),
    tools=[WebSearchTool()],
)

# ---- Budget Agent (estimates costs under constraints) ----
budget_agent = Agent(
    name="Budget Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user mentions budget, price, cost, dollars, under $X, or asks 'how much'.",
    instructions=(
        "You estimate costs for lodging, food, transport, and activities at a high level; flag budget violations. "
        'Always return JSON with this structure: {"cost":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "medium"}, extra_body={"text": {"verbosity": "low"}}
    ),
    tools=[WebSearchTool()],
)

# ---- Local Guide Agent (adds local tips & dining) ----
local_guide_agent = Agent(
    name="Local Guide Agent",
    model="gpt-5.1",
    handoff_description="Use me when the user asks for food, restaurants, neighborhoods, local tips, or 'what's good nearby'.",
    instructions=(
        "You provide restaurants, neighborhoods, cultural tips, and current local highlights. "
        'Always return JSON with this structure: {"tips":"string"}.'
    ),
    model_settings=ModelSettings(
        reasoning={"effort": "medium"}, extra_body={"text": {"verbosity": "low"}}
    ),
    tools=[WebSearchTool()],
)

# ---- Core orchestrator: Travel Agent ----
travel_agent = Agent(
    name="Travel Agent",
    model="gpt-5.1",
    instructions=(
        "You are a friendly and knowledgeable travel planner that helps users plan trips, suggest destinations, and create detailed summaries of their journeys.\n"
        "Your primary role is to orchestrate other specialized agents (used as tools) to complete the user's request.\n"
        "\n"
        "When planning an itinerary, call the **Planner Agent** to create daily schedules, organize destinations, and recommend attractions or activities. Do not create itineraries yourself.\n"
        "When estimating costs, call the **Budget Agent** to calculate the total trip cost including flights, hotels, and activities. Do not calculate or estimate prices on your own.\n"
        "When recommending local experiences, restaurants, neighborhoods, or cultural highlights, call the **Local Guide Agent** to provide these insights. Do not generate local recommendations without this agent.\n"
        "\n"
        "Use these agents one at a time in a logical order based on the request — start with the Planner Agent, then the Budget Agent, and finally the Local Guide Agent.\n"
        "After receiving results from these agents, combine their outputs into a single structured summary.\n"
        "\n"
        "Return JSON output using this exact structure:\n"
        '{"destination": "string", "duration": "string", "summary": "string", "cost": "string", "tips": "string"}.\n'
    ),
    output_type=TravelOutput,
    model_settings=ModelSettings(
        reasoning={"effort": "medium"},  # minimal | low | medium | high
        extra_body={"text": {"verbosity": "low"}},  # low | medium | high
    ),
    tools=[
        WebSearchTool(),
        planner_agent.as_tool(
            tool_name="planner_agent",
            tool_description="plan or outline an itinerary, schedule, or daily plan",
        ),
        budget_agent.as_tool(
            tool_name="budget_agent", tool_description="calculates the cost of a trip"
        ),
        local_guide_agent.as_tool(
            tool_name="local_guide_agent",
            tool_description="provide restaurants, neighborhoods, cultural tips, and current local highlights",
        ),
    ],
)


# --- Pretty print helper ----------------------------------------------------
def print_fields(data):
    if isinstance(data, str):
        try:
            data = TravelOutput(**json.loads(data))
        except (json.JSONDecodeError, ValidationError) as e:
            print("Raw output:", data)
            return
    print(f"Destination: {data.destination}")
    print(f"Duration: {data.duration}")
    print(f"Summary: {data.summary}")
    print(f"Cost: {data.cost}")
    print(f"Tips: {data.tips}")


async def main():
    try:
        result = await Runner.run(
            travel_agent,
            """I'm considering a trip to Jamaica sometime in late September or early October. 
                                                I'm pretty flexible with the exact dates, maybe around a week-long trip.     
                                                I'd like to get an idea of flight ticket prices and some well-located hotels.     
                                                I'm also a big foodie, so any recommendations for great local restaurants would be fantastic! 
                                                Do not ask follow-up questions.""",
        )
        print_fields(result.final_output)
    except Exception as e:
        print("Error", e)


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