pip install openai-agents

export OPENAI_API_KEY=sk-...

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(name="Assistant", instructions="Be helpful")
    result = await Runner.run(agent, "Hello!")
    print(result.final_output)

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

from agents import Agent, Runner
import asyncio

agent = Agent(
    name="Math Tutor",
    instructions="Help with math problems. Explain step-by-step.",
    model="gpt-4o"  # Optional: defaults to gpt-4o
)

async def main():
    result = await Runner.run(agent, "What is 15% of 200?")
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner, function_tool
import asyncio

@function_tool
def get_weather(city: str) -> str:
    """Get current weather for a city.

    Args:
        city: The city name to check weather for
    """
    # In production, call a real weather API
    return f"Weather in {city}: Sunny, 72°F"

@function_tool
def convert_temperature(fahrenheit: float) -> str:
    """Convert Fahrenheit to Celsius.

    Args:
        fahrenheit: Temperature in Fahrenheit
    """
    celsius = (fahrenheit - 32) * 5/9
    return f"{fahrenheit}°F = {celsius:.1f}°C"

agent = Agent(
    name="Weather Assistant",
    instructions="Help users with weather information and temperature conversions.",
    tools=[get_weather, convert_temperature]
)

async def main():
    result = await Runner.run(
        agent,
        "What's the weather in Seattle and convert that temperature to Celsius?"
    )
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner
from pydantic import BaseModel, Field
import asyncio

class CalendarEvent(BaseModel):
    name: str = Field(description="Event name")
    date: str = Field(description="Event date in YYYY-MM-DD format")
    participants: list[str] = Field(description="List of participant names")

agent = Agent(
    name="Calendar Extractor",
    instructions="Extract calendar events from text",
    output_type=CalendarEvent  # Forces structured output
)

async def main():
    result = await Runner.run(
        agent,
        "Schedule a meeting with John and Sarah on 2026-03-15 to discuss Q1 results"
    )

    # Access typed output
    event = result.final_output_as(CalendarEvent)
    print(f"Event: {event.name}")
    print(f"Date: {event.date}")
    print(f"Participants: {', '.join(event.participants)}")

asyncio.run(main())

from agents import Agent, Runner
import asyncio

# Specialized agents
spanish_translator = Agent(
    name="Spanish Translator",
    instructions="Translate to Spanish. Return only the translation.",
)

french_translator = Agent(
    name="French Translator",
    instructions="Translate to French. Return only the translation.",
)

# Manager agent using specialists as tools
manager = Agent(
    name="Translation Manager",
    instructions="Use translation tools to translate messages.",
    tools=[
        spanish_translator.as_tool(
            tool_name="translate_spanish",
            tool_description="Translate text to Spanish"
        ),
        french_translator.as_tool(
            tool_name="translate_french",
            tool_description="Translate text to French"
        ),
    ],
)

async def main():
    result = await Runner.run(
        manager,
        "Translate 'Good morning' to Spanish and French"
    )
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner
import asyncio

# Specialist agents
math_tutor = Agent(
    name="Math Tutor",
    handoff_description="Expert in mathematics and problem-solving",
    instructions="Explain math concepts clearly with examples.",
)

history_tutor = Agent(
    name="History Tutor",
    handoff_description="Expert in historical events and context",
    instructions="Explain historical events with context and significance.",
)

# Triage agent routes to specialists
triage = Agent(
    name="Triage Agent",
    instructions="Route student questions to the appropriate tutor.",
    handoffs=[math_tutor, history_tutor]  # Peer agents
)

async def main():
    result = await Runner.run(
        triage,
        "Who was the first president of the United States?"
    )
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner, InputGuardrail, GuardrailFunctionOutput
from agents.exceptions import InputGuardrailTripwireTriggered
from pydantic import BaseModel
import asyncio

class HomeworkCheck(BaseModel):
    is_homework: bool
    reasoning: str

guardrail_agent = Agent(
    name="Homework Checker",
    instructions="Determine if the user is asking about homework.",
    output_type=HomeworkCheck,
)

async def homework_guardrail(ctx, agent, input_data):
    result = await Runner.run(guardrail_agent, input_data, context=ctx.context)
    check = result.final_output_as(HomeworkCheck)

    return GuardrailFunctionOutput(
        output_info=check,
        tripwire_triggered=not check.is_homework  # Block non-homework
    )

tutor_agent = Agent(
    name="Homework Tutor",
    instructions="Help with homework. Don't do it for them.",
    input_guardrails=[InputGuardrail(guardrail_function=homework_guardrail)]
)

async def main():
    try:
        result = await Runner.run(tutor_agent, "What's the capital of France?")
        print(result.final_output)
    except InputGuardrailTripwireTriggered as e:
        print("⛔ Input blocked by guardrail:", e)

asyncio.run(main())

from agents import Agent, Runner, SQLiteSession
import asyncio

agent = Agent(
    name="Assistant",
    instructions="Be helpful and concise.",
)

async def main():
    # Session persists conversation history
    session = SQLiteSession("user_123", "conversations.db")

    # Turn 1
    result1 = await Runner.run(
        agent,
        "What city is the Golden Gate Bridge in?",
        session=session
    )
    print(f"Turn 1: {result1.final_output}")

    # Turn 2 - Agent remembers previous context
    result2 = await Runner.run(
        agent,
        "What state is that in?",  # "that" refers to San Francisco
        session=session
    )
    print(f"Turn 2: {result2.final_output}")

asyncio.run(main())

from agents import Agent, Runner, RunContextWrapper, function_tool
from dataclasses import dataclass
import asyncio

@dataclass
class UserInfo:
    name: str
    user_id: int

@function_tool
async def get_user_data(ctx: RunContextWrapper[UserInfo]) -> str:
    """Fetch user-specific data."""
    return f"User {ctx.context.name} (ID: {ctx.context.user_id}) has 5 orders"

agent = Agent[UserInfo](
    name="Customer Service",
    tools=[get_user_data],
)

async def main():
    user = UserInfo(name="Alice", user_id=12345)

    result = await Runner.run(
        agent,
        "How many orders do I have?",
        context=user  # Shared across all tool calls
    )
    print(result.final_output)

asyncio.run(main())

from agents.exceptions import (
    MaxTurnsExceeded,
    InputGuardrailTripwireTriggered,
    OutputGuardrailTripwireTriggered
)

async def run_agent_safely(agent, user_input):
    try:
        result = await Runner.run(
            agent,
            user_input,
            max_turns=10  # Prevent infinite loops
        )
        return result.final_output
    except MaxTurnsExceeded:
        return "⚠️ Agent exceeded maximum turns. Task too complex."
    except InputGuardrailTripwireTriggered as e:
        return f"⛔ Input blocked: {e.guardrail_result.output.output_info}"
    except OutputGuardrailTripwireTriggered as e:
        return f"⛔ Output blocked: {e.guardrail_result.output.output_info}"

from agents.tracing.traces import trace

# Add metadata to traces
with trace(
    "Customer Support",
    group_id="chat_456",
    metadata={"customer": "user_789"}
):
    result = await Runner.run(agent, query)

export OPENAI_AGENTS_DISABLE_TRACING=1

from agents.run import RunConfig

result = await Runner.run(
    agent,
    input_text,
    run_config=RunConfig(tracing_disabled=True)
)