OpenAI Agents SDK Examples: 5 Real Builds

import asyncio
from agents import Agent, Runner, function_tool
import httpx

@function_tool
async def search_web(query: str) -> str:
    """Search the web for current information on a topic."""
    # Production: use Tavily, Serper, or Brave Search API
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.tavily.com/search",
            params={"query": query, "max_results": 3},
            headers={"Authorization": f"Bearer {TAVILY_API_KEY}"}
        )
        results = response.json()["results"]
        return "\n".join([f"- {r['title']}: {r['content'][:200]}" for r in results])

research_agent = Agent(
    name="Research Agent",
    model="gpt-4o",
    instructions="""You are a research assistant. For questions about current events
    or recent developments, always use the search_web tool to find accurate information.
    Summarize findings clearly and cite your sources.""",
    tools=[search_web]
)

async def main():
    result = await Runner.run(
        research_agent,
        input="What are the most significant AI agent framework releases in early 2026?"
    )
    print(result.final_output)

asyncio.run(main())

import asyncio
from agents import Agent, Runner, handoff

billing_agent = Agent(
    name="Billing Specialist",
    model="gpt-4o-mini",
    instructions="""You handle billing questions: invoices, charges, upgrades,
    cancellations, and refunds. Resolve issues thoroughly and confirm next steps."""
)

technical_agent = Agent(
    name="Technical Support Engineer",
    model="gpt-4o",
    instructions="""You resolve technical issues: API errors, integration problems,
    configuration questions, and bug reports. Ask clarifying questions when needed."""
)

sales_agent = Agent(
    name="Sales Representative",
    model="gpt-4o-mini",
    instructions="""You help prospects understand pricing, features, and fit.
    You can offer trials and schedule demos. Never offer unauthorized discounts."""
)

triage_agent = Agent(
    name="Customer Support Triage",
    model="gpt-4o-mini",
    instructions="""You are the first point of contact. Greet the customer warmly,
    understand their need, then hand off to the right specialist:
    - Billing questions → billing_specialist
    - Technical issues → technical_support
    - Pricing or features → sales
    Do not attempt to answer domain-specific questions yourself.""",
    handoffs=[
        handoff(billing_agent, tool_description="Transfer to billing for payment and account questions"),
        handoff(technical_agent, tool_description="Transfer to technical support for product issues"),
        handoff(sales_agent, tool_description="Transfer to sales for pricing and upgrade questions")
    ]
)

async def main():
    result = await Runner.run(
        triage_agent,
        input="Hi, I was charged twice for my subscription this month and my API integration stopped working."
    )
    print(result.final_output)

asyncio.run(main())

import asyncio
from agents import Agent, Runner
from pydantic import BaseModel
from typing import List, Literal

class ReviewIssue(BaseModel):
    line_range: str
    severity: Literal["critical", "high", "medium", "low", "info"]
    category: Literal["security", "performance", "style", "logic", "documentation"]
    description: str
    suggested_fix: str

class CodeReview(BaseModel):
    verdict: Literal["approved", "approved_with_suggestions", "changes_required", "rejected"]
    summary: str
    issues: List[ReviewIssue]
    positive_aspects: List[str]

code_reviewer = Agent(
    name="Code Reviewer",
    model="gpt-4o",
    instructions="""You are a senior software engineer conducting a thorough code review.
    Analyze the provided code for security vulnerabilities, performance issues, logic errors,
    and style problems. Be specific about line numbers and provide concrete fix suggestions.
    Output a structured review in the required format.""",
    output_type=CodeReview
)

async def review_code(code: str) -> CodeReview:
    result = await Runner.run(
        code_reviewer,
        input=f"Review this Python code:\n\n```python\n{code}\n```"
    )
    return result.final_output  # Already typed as CodeReview

async def main():
    sample_code = """
def get_user(user_id):
    query = f"SELECT * FROM users WHERE id = {user_id}"
    return db.execute(query)
    """
    review = await review_code(sample_code)
    print(f"Verdict: {review.verdict}")
    for issue in review.issues:
        print(f"[{issue.severity.upper()}] {issue.description}")

asyncio.run(main())

import asyncio
from agents import Agent, Runner, function_tool, input_guardrail, GuardrailFunctionOutput
from pydantic import BaseModel
import sqlite3

DB_PATH = "analytics.db"

@function_tool
def get_database_schema() -> str:
    """Get the schema of all tables in the analytics database."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute("SELECT name, sql FROM sqlite_master WHERE type='table'")
    return "\n".join([f"{row[0]}:\n{row[1]}" for row in cursor.fetchall()])

@function_tool
def run_sql_query(sql: str) -> str:
    """Execute a read-only SQL SELECT query and return results as CSV."""
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries are permitted."
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(sql)
    cols = [desc[0] for desc in cursor.description]
    rows = cursor.fetchmany(100)  # Limit results
    return ",".join(cols) + "\n" + "\n".join([",".join(str(v) for v in row) for row in rows])

class AnalysisResult(BaseModel):
    sql_query: str
    interpretation: str
    key_finding: str
    chart_recommendation: str

data_analyst = Agent(
    name="Data Analyst",
    model="gpt-4o",
    instructions="""You are a data analyst. When asked a question:
    1. Use get_database_schema to understand available tables
    2. Write an appropriate SQL SELECT query
    3. Execute with run_sql_query
    4. Interpret the results in plain language
    5. Recommend the best chart type to visualize the answer
    Always explain what the numbers mean in business terms.""",
    tools=[get_database_schema, run_sql_query],
    output_type=AnalysisResult
)

async def main():
    result = await Runner.run(
        data_analyst,
        input="Which product category had the highest revenue growth between Q3 and Q4 last year?"
    )
    analysis = result.final_output
    print(f"SQL: {analysis.sql_query}")
    print(f"Finding: {analysis.key_finding}")
    print(f"Chart: {analysis.chart_recommendation}")

asyncio.run(main())

import asyncio
from agents import Agent, Runner, handoff, input_guardrail, GuardrailFunctionOutput
from pydantic import BaseModel

class DocumentValidation(BaseModel):
    is_valid_document: bool
    rejection_reason: str

@input_guardrail
async def document_validator(ctx, agent, input: str) -> GuardrailFunctionOutput:
    """Reject inputs that don't appear to contain document content."""
    is_too_short = len(input.strip()) < 50
    is_question_not_doc = input.strip().endswith("?") and len(input) < 200

    if is_too_short or is_question_not_doc:
        return GuardrailFunctionOutput(
            output_info=DocumentValidation(
                is_valid_document=False,
                rejection_reason="Input does not appear to be a document. Please paste document content."
            ),
            tripwire_triggered=True
        )
    return GuardrailFunctionOutput(
        output_info=DocumentValidation(is_valid_document=True, rejection_reason=""),
        tripwire_triggered=False
    )

contract_agent = Agent(
    name="Contract Analyst",
    model="gpt-4o",
    instructions="Extract parties, obligations, key dates, and risk clauses from contracts."
)

invoice_agent = Agent(
    name="Invoice Processor",
    model="gpt-4o-mini",
    instructions="Extract vendor, amounts, line items, due date, and payment terms from invoices."
)

resume_agent = Agent(
    name="Resume Parser",
    model="gpt-4o-mini",
    instructions="Extract candidate name, contact info, skills, experience, and education from resumes."
)

document_classifier = Agent(
    name="Document Intake Classifier",
    model="gpt-4o-mini",
    instructions="""Classify the document type and route to the appropriate specialist:
    - Legal contracts or agreements → contract_analyst
    - Invoices or purchase orders → invoice_processor
    - Resumes or CVs → resume_parser
    Never process the document yourself — always route to a specialist.""",
    input_guardrails=[document_validator],
    handoffs=[
        handoff(contract_agent),
        handoff(invoice_agent),
        handoff(resume_agent)
    ]
)

async def main():
    # Test with an invoice
    invoice_text = """
    INVOICE #2026-0234
    From: Acme Supplies Inc. | To: Beta Corp
    Date: 2026-02-15 | Due: 2026-03-15
    Line Items:
    - Cloud Storage (500GB): $299.00
    - Support Plan (Annual): $1,200.00
    Total Due: $1,499.00
    """
    result = await Runner.run(document_classifier, input=invoice_text)
    print(result.final_output)

asyncio.run(main())