Scaffolding OpenAI Agents

Build production AI agents using OpenAI Agents SDK with native async/await patterns.

Quick Start

# Project setup
mkdir my-agent && cd my-agent
python -m venv .venv && source .venv/bin/activate
pip install openai-agents

# Set API key
export OPENAI_API_KEY=sk-...

# main.py
import asyncio
from agents import Agent, Runner

agent = Agent(
    name="Python Tutor",
    instructions="You help students learn Python. Explain concepts clearly with examples."
)

async def main():
    result = await Runner.run(agent, "Explain list comprehensions")
    print(result.final_output)

asyncio.run(main())

Agent Configuration

Basic Agent

from agents import Agent

tutor = Agent(
    name="Python Tutor",
    instructions="""You are an expert Python tutor.
    Explain concepts clearly with examples.
    Ask clarifying questions when needed.
    Provide practice exercises after explanations.""",
    model="gpt-4o"
)

With Model Settings

from agents import Agent, ModelSettings

agent = Agent(
    name="Creative Writer",
    instructions="Write creative stories based on prompts.",
    model="gpt-4o",
    model_settings=ModelSettings(
        temperature=0.9,
        max_tokens=2000
    )
)

With Structured Output

from pydantic import BaseModel
from agents import Agent

class CodeReview(BaseModel):
    issues: list[str]
    suggestions: list[str]
    score: int

reviewer = Agent(
    name="Code Reviewer",
    instructions="Review Python code for issues and improvements.",
    output_type=CodeReview  # Forces structured JSON output
)

Runner Patterns

Async Run (Primary)

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(name="Helper", instructions="Be helpful")

    # Single query
    result = await Runner.run(agent, "What is Python?")
    print(result.final_output)

    # With conversation history
    messages = [
        {"role": "user", "content": "My name is Alex"},
        {"role": "assistant", "content": "Nice to meet you, Alex!"},
        {"role": "user", "content": "What's my name?"}
    ]
    result = await Runner.run(agent, messages)
    print(result.final_output)  # "Your name is Alex"

asyncio.run(main())

Sync Run (Simple Scripts)

from agents import Agent, Runner

agent = Agent(name="Helper", instructions="Be helpful")
result = Runner.run_sync(agent, "Hello!")
print(result.final_output)

Streaming Run

import asyncio
from agents import Agent, Runner

async def main():
    agent = Agent(name="Storyteller", instructions="Tell engaging stories")

    result = Runner.run_streamed(agent, "Tell me a short story")

    async for event in result.stream_events():
        if hasattr(event, 'delta'):
            print(event.delta, end='', flush=True)

    print()  # Newline at end

asyncio.run(main())

Conversation Continuation

async def chat_session():
    agent = Agent(name="Tutor", instructions="You are a Python tutor")

    # First turn
    result1 = await Runner.run(agent, "Explain decorators")
    print(f"Tutor: {result1.final_output}")

    # Continue conversation
    messages = result1.to_input_list() + [
        {"role": "user", "content": "Show me an example"}
    ]
    result2 = await Runner.run(agent, messages)
    print(f"Tutor: {result2.final_output}")

Function Tools

Basic Tool

from agents import Agent, function_tool

@function_tool
def get_current_time() -> str:
    """Get the current time."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S")

@function_tool
def calculate(expression: str) -> float:
    """Calculate a mathematical expression.

    Args:
        expression: A valid Python math expression like "2 + 2" or "10 * 5"
    """
    return eval(expression)  # Use safe_eval in production

agent = Agent(
    name="Assistant",
    instructions="Help with calculations and time queries.",
    tools=[get_current_time, calculate]
)

Async Tool

import httpx
from agents import Agent, function_tool

@function_tool
async def fetch_weather(city: str) -> str:
    """Fetch current weather for a city.

    Args:
        city: The city name to get weather for
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}?format=3"
        )
        return response.text

agent = Agent(
    name="Weather Bot",
    instructions="Provide weather information.",
    tools=[fetch_weather]
)

Tool with Pydantic Types

from pydantic import BaseModel
from agents import Agent, function_tool

class SearchQuery(BaseModel):
    query: str
    max_results: int = 10

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

@function_tool
async def search_docs(params: SearchQuery) -> list[SearchResult]:
    """Search documentation for a query."""
    # Implementation
    return [SearchResult(
        title="Python Tutorial",
        url="https://docs.python.org",
        snippet="Official Python documentation..."
    )]

agent = Agent(
    name="Doc Search",
    instructions="Search Python documentation.",
    tools=[search_docs]
)

Multi-Agent Patterns

Handoffs (Recommended for Routing)

from agents import Agent, Runner

# Specialist agents
concepts_agent = Agent(
    name="Concepts Tutor",
    handoff_description="Explains Python concepts and fundamentals",
    instructions="Explain Python concepts clearly with examples."
)

debug_agent = Agent(
    name="Debug Helper",
    handoff_description="Helps debug Python code errors",
    instructions="Help diagnose and fix Python errors."
)

exercise_agent = Agent(
    name="Exercise Generator",
    handoff_description="Creates practice problems and exercises",
    instructions="Generate practice problems with solutions."
)

# Triage agent with handoffs
triage_agent = Agent(
    name="Triage",
    instructions="""Route student questions to the right specialist:
    - Concepts questions → Concepts Tutor
    - Error/bug questions → Debug Helper
    - Practice requests → Exercise Generator

    Analyze the question and hand off to the appropriate agent.""",
    handoffs=[concepts_agent, debug_agent, exercise_agent]
)

async def main():
    # Question gets routed automatically
    result = await Runner.run(
        triage_agent,
        "I'm getting a KeyError in my dictionary code"
    )
    print(result.final_output)  # Handled by debug_agent

Agents as Tools (Orchestration)

from agents import Agent, Runner

# Create specialist agents
researcher = Agent(
    name="Researcher",
    instructions="Research topics thoroughly."
)

writer = Agent(
    name="Writer",
    instructions="Write clear, engaging content."
)

# Manager uses agents as tools
manager = Agent(
    name="Content Manager",
    instructions="""Coordinate research and writing:
    1. Use researcher tool to gather information
    2. Use writer tool to create content""",
    tools=[
        researcher.as_tool(
            tool_name="research",
            tool_description="Research a topic"
        ),
        writer.as_tool(
            tool_name="write",
            tool_description="Write content about a topic"
        )
    ]
)

async def main():
    result = await Runner.run(
        manager,
        "Create a blog post about async Python"
    )
    print(result.final_output)

Guardrails

Input Validation

from agents import Agent, input_guardrail, GuardrailFunctionOutput

@input_guardrail
async def check_homework_topic(context, agent, input_text: str) -> GuardrailFunctionOutput:
    """Ensure questions are homework-related."""
    keywords = ["python", "code", "programming", "function", "class", "error"]

    if not any(kw in input_text.lower() for kw in keywords):
        return GuardrailFunctionOutput(
            output_info="Not a programming question",
            tripwire_triggered=True
        )

    return GuardrailFunctionOutput(
        output_info="Valid programming question",
        tripwire_triggered=False
    )

tutor = Agent(
    name="Python Tutor",
    instructions="Help with Python homework.",
    input_guardrails=[check_homework_topic]
)

Output Validation

from agents import Agent, output_guardrail, GuardrailFunctionOutput

@output_guardrail
async def check_no_solutions(context, agent, output: str) -> GuardrailFunctionOutput:
    """Ensure we don't give complete homework solutions."""
    solution_indicators = ["here's the complete", "full solution", "copy this code"]

    if any(ind in output.lower() for ind in solution_indicators):
        return GuardrailFunctionOutput(
            output_info="Contains complete solution",
            tripwire_triggered=True
        )

    return GuardrailFunctionOutput(
        output_info="Output is appropriate",
        tripwire_triggered=False
    )

tutor = Agent(
    name="Python Tutor",
    instructions="Guide students without giving full solutions.",
    output_guardrails=[check_no_solutions]
)

Context Injection

Shared State Across Agents

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

@dataclass
class TutoringContext:
    student_id: str
    session_id: str
    topics_covered: list[str]
    difficulty_level: str = "beginner"

@function_tool
def log_topic(wrapper: RunContextWrapper[TutoringContext], topic: str) -> str:
    """Log a topic as covered in this session."""
    wrapper.context.topics_covered.append(topic)
    return f"Logged: {topic}"

tutor = Agent(
    name="Python Tutor",
    instructions="Teach Python, tracking topics covered.",
    tools=[log_topic]
)

async def main():
    ctx = TutoringContext(
        student_id="student-123",
        session_id="session-456",
        topics_covered=[]
    )

    result = await Runner.run(
        tutor,
        "Teach me about loops",
        context=ctx
    )

    print(f"Topics covered: {ctx.topics_covered}")

Project Structure

learnflow-agents/
├── agents/
│   ├── __init__.py
│   ├── triage.py          # Routing agent
│   ├── concepts.py        # Concepts specialist
│   ├── debug.py           # Debug specialist
│   └── exercise.py        # Exercise generator
├── tools/
│   ├── __init__.py
│   ├── code_runner.py     # Execute Python safely
│   └── search.py          # Search documentation
├── guardrails/
│   ├── __init__.py
│   ├── input.py           # Input validation
│   └── output.py          # Output validation
├── main.py                # FastAPI integration
└── pyproject.toml

FastAPI Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agents import Agent, Runner

app = FastAPI()

# Initialize agents
triage = Agent(
    name="Triage",
    instructions="Route questions to specialists",
    handoffs=[concepts_agent, debug_agent]
)

class Question(BaseModel):
    text: str
    session_id: str

class Answer(BaseModel):
    response: str
    agent_used: str

@app.post("/ask", response_model=Answer)
async def ask_question(question: Question):
    try:
        result = await Runner.run(triage, question.text)
        return Answer(
            response=result.final_output,
            agent_used=result.last_agent.name
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/ask/stream")
async def ask_stream(question: Question):
    from fastapi.responses import StreamingResponse

    async def generate():
        result = Runner.run_streamed(triage, question.text)
        async for event in result.stream_events():
            if hasattr(event, 'delta'):
                yield event.delta

    return StreamingResponse(generate(), media_type="text/plain")

Tracing & Debugging

View Traces

Traces available at: https://platform.openai.com/traces

Custom Tracing

from agents import Runner, RunConfig

config = RunConfig(
    workflow_name="tutoring-session",
    trace_id="custom-trace-123"
)

result = await Runner.run(agent, "Hello", run_config=config)

Verification

Run:

python scripts/verify.py

Related Skills

• configuring-dapr-pubsub

  - Agent-to-agent messaging

• scaffolding-fastapi-dapr

  - FastAPI backend integration

• streaming-llm-responses

  - Response streaming patterns

• building-chat-interfaces

  - Frontend chat UI