Stage 08 — OpenAI Agents SDK

OpenAI Agents SDK  ·  Comprehensive Technical Training  ·  ⏱ 8–10 hours

Learning Objectives

By the end of this stage you will be able to:

• Use OpenAI Agents SDK to build production agents
• Understand async/await for concurrent LLM calls
• Implement structured outputs with Pydantic
• Build tool hierarchies (agents as tools)
• Monitor agent execution with tracing
• Implement guardrails and safety constraints

Section 1: Async Python Fundamentals

Agents often need to make multiple LLM calls. Async/await allows concurrent execution rather than blocking on each call.

Async Basics

import asyncio

async def fetch_data(url):
    """Simulated async API call"""
    print(f"Fetching {url}")
    await asyncio.sleep(1)  # Simulate network delay
    return f"Data from {url}"

async def main():
    # Concurrent calls
    results = await asyncio.gather(
        fetch_data("http://api1.com"),
        fetch_data("http://api2.com"),
        fetch_data("http://api3.com"),
    )
    return results

# Run
results = asyncio.run(main())
# Completes in ~1 second, not 3 seconds

Key benefit: 3 sequential API calls taking 1s each = 3s total. Same calls async = ~1s total.

Section 2: OpenAI Agents SDK

from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class SearchResult(BaseModel):
    title: str
    url: str
    description: str

# Define tools
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                },
                "required": ["query"],
            },
        },
    }
]

# Agent with tools
agent = client.agents.create(
    model="gpt-4o",
    name="Research Agent",
    instructions="You are a research agent. Use tools to search for information.",
    tools=tools,
)

# Execute a task
run = client.agents.execute(
    agent_id=agent.id,
    task="Find information about recent AI breakthroughs",
)

# Check results
print(run.output)

Section 3: Concurrent Agent Calls

async def run_multiple_agents():
    tasks = [
        client.agents.execute_async(agent1, "Task 1"),
        client.agents.execute_async(agent2, "Task 2"),
        client.agents.execute_async(agent3, "Task 3"),
    ]

    results = await asyncio.gather(*tasks)
    return results

# Run parallel agents
results = asyncio.run(run_multiple_agents())

Section 4: Structured Outputs

from pydantic import BaseModel

class EmailResponse(BaseModel):
    recipient: str
    subject: str
    body: str
    priority: str  # low, medium, high

# Force JSON structure
response = client.beta.messages.parse(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": "Draft an email requesting a meeting about Q2 planning",
        }
    ],
    response_format=EmailResponse,
)

# Now safe to access fields
email = response.parsed
print(f"To: {email.recipient}")
print(f"Subject: {email.subject}")
print(f"Priority: {email.priority}")

What's Next

Stage 9 explores multi-agent frameworks like CrewAI and LangGraph, where agents collaborate to solve complex problems through orchestrated workflows.