OpenAI Agents SDK

from agents import Agent, Runner

agent = Agent(
    name="Support Assistant",
    instructions="You are a helpful customer support agent. Be concise.",
)

result = Runner.run_sync(agent, "How do I reset my password?")
print(result.final_output)

import asyncio

async def main():
    result = await Runner.run(agent, "How do I reset my password?")
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # In production: call a weather API
    return f"The weather in {city} is 18°C and cloudy."

@function_tool
def search_knowledge_base(query: str) -> str:
    """Search internal documentation for relevant articles."""
    # In production: call your vector store
    return f"Found 3 articles matching '{query}'..."

agent = Agent(
    name="Research Assistant",
    instructions="Help users find information. Use the knowledge base before answering.",
    tools=[get_weather, search_knowledge_base],
)

from agents import Agent, Runner

billing_agent = Agent(
    name="Billing Agent",
    instructions="Handle billing questions: invoices, refunds, payment methods.",
)

technical_agent = Agent(
    name="Technical Agent",
    instructions="Handle technical issues: bugs, integrations, API errors.",
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="""Classify the customer's issue and hand off to the right agent.
- Billing questions → billing_agent
- Technical questions → technical_agent
- Everything else: handle yourself.""",
    handoffs=[billing_agent, technical_agent],
)

result = Runner.run_sync(triage_agent, "My payment keeps failing with error code 402")
print(result.final_output)
# Triage agent routes to billing_agent, which handles the payment issue

from agents import Agent, Runner, GuardrailFunctionOutput, input_guardrail, output_guardrail
from pydantic import BaseModel

class ContentCheck(BaseModel):
    is_safe: bool
    reason: str

@input_guardrail
async def check_safe_input(ctx, agent, input: str) -> GuardrailFunctionOutput:
    """Block requests containing PII or harmful content."""
    # Use a cheap model for the check
    check = await fast_safety_check(input)
    return GuardrailFunctionOutput(
        output_info=ContentCheck(is_safe=check.safe, reason=check.reason),
        tripwire_triggered=not check.safe,  # True = block the request
    )

@output_guardrail
async def check_safe_output(ctx, agent, output: str) -> GuardrailFunctionOutput:
    """Ensure output doesn't leak sensitive data."""
    contains_pii = detect_pii(output)
    return GuardrailFunctionOutput(
        output_info=ContentCheck(is_safe=not contains_pii, reason="PII detected" if contains_pii else "OK"),
        tripwire_triggered=contains_pii,
    )

agent = Agent(
    name="Secure Agent",
    instructions="Answer user questions helpfully.",
    input_guardrails=[check_safe_input],
    output_guardrails=[check_safe_output],
)

from pydantic import BaseModel
from agents import Agent, Runner

class TicketClassification(BaseModel):
    category: str        # billing | technical | refund | general
    priority: str        # low | medium | high
    sentiment: str       # positive | neutral | negative
    summary: str

classifier = Agent(
    name="Classifier",
    instructions="Classify support tickets accurately.",
    output_type=TicketClassification,
)

result = Runner.run_sync(classifier, "I've been charged twice this month and I'm furious!")
ticket = result.final_output  # type: TicketClassification
print(ticket.category)   # billing
print(ticket.sentiment)  # negative

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

@dataclass
class UserContext:
    user_id: str
    plan: str           # free | pro | enterprise
    account_age_days: int

@function_tool
def get_user_plan(ctx: RunContextWrapper[UserContext]) -> str:
    """Get the user's current subscription plan."""
    return f"Plan: {ctx.context.plan}, Account age: {ctx.context.account_age_days} days"

agent = Agent(
    name="Account Agent",
    instructions="Help users understand their account.",
    tools=[get_user_plan],
)

user_ctx = UserContext(user_id="u_123", plan="pro", account_age_days=180)
result = Runner.run_sync(agent, "What features do I have access to?", context=user_ctx)

from agents import Agent, Runner
from agents.tracing import set_tracing_export_api_key

# Or export to a custom endpoint
set_tracing_export_api_key("your-key")

agent = Agent(name="My Agent", instructions="...")
result = Runner.run_sync(agent, "Hello")
# Visit platform.openai.com to inspect the trace

from agents import Agent, Runner

agent = Agent(name="Streaming Agent", instructions="Be helpful.")

async def stream_response(query: str):
    async with Runner.run_streamed(agent, query) as stream:
        async for event in stream.stream_events():
            if event.type == "raw_response_event":
                # Print each text chunk as it arrives
                print(event.data.delta, end="", flush=True)

import litellm
from agents import Agent, Runner, OpenAIChatCompletionsModel
from openai import AsyncOpenAI

# LiteLLM proxies Claude to OpenAI-compatible endpoints
litellm_client = AsyncOpenAI(
    api_key="sk-1234",  # LiteLLM proxy key
    base_url="http://localhost:4000",
)

claude_model = OpenAIChatCompletionsModel(
    model="claude-sonnet-4-6",
    openai_client=litellm_client,
)

agent = Agent(
    name="Claude Agent",
    instructions="You are a helpful assistant.",
    model=claude_model,
)

from agents import Agent, Runner, function_tool

@function_tool
def search_web(query: str) -> str:
    """Search the web and return relevant excerpts."""
    # call Perplexity, Tavily, or similar
    return web_search(query)

researcher = Agent(
    name="Researcher",
    instructions="""Search for information on the given topic.
Return a structured research summary with sources.""",
    tools=[search_web],
)

writer = Agent(
    name="Writer",
    instructions="""Take the research provided and write a clear, concise blog post.
Target 500 words. Use the research findings directly.""",
)

# Sequential pipeline
async def research_and_write(topic: str) -> str:
    research_result = await Runner.run(researcher, f"Research: {topic}")
    research_summary = research_result.final_output

    write_result = await Runner.run(
        writer,
        f"Write a blog post based on this research:\n\n{research_summary}",
    )
    return write_result.final_output