scaffolding-openai-agents

name

scaffolding-openai-agents

description

Builds AI agents using OpenAI Agents SDK with async/await patterns and multi-agent orchestration. Use when creating tutoring agents, building agent handoffs, implementing tool-calling agents, or orchestrating multiple specialists. Covers Agent class, Runner patterns, function tools, guardrails, and streaming responses. NOT when using raw OpenAI API without SDK or other agent frameworks like LangChain.

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
uv venv && source .venv/bin/activate
uv add 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")

Battle-Tested Patterns (from production implementation)

OPENAI_API_KEY + pydantic-settings Pitfall

pydantic-settings reads .env into a Settings object but does NOT export to os.environ. The OpenAI Agents SDK reads OPENAI_API_KEY directly from os.environ. You must explicitly bridge this:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    openai_api_key: str = ""

    def __init__(self, **kwargs):
        import os
        super().__init__(**kwargs)
        # OpenAI Agents SDK reads from os.environ, not from Settings
        if self.openai_api_key and not os.environ.get("OPENAI_API_KEY"):
            os.environ["OPENAI_API_KEY"] = self.openai_api_key

Without this, you get: OpenAIError: The api_key client option must be set either by passing api_key to the client or by setting the OPENAI_API_KEY environment variable

Streaming Event Types (Runner.run_streamed)

The streaming API emits two key event types — use isinstance checks, not hasattr:

from agents.stream_events import RawResponsesStreamEvent, RunItemStreamEvent
from agents.items import ToolCallOutputItem

result = Runner.run_streamed(agent, messages, context=ctx)

async for event in result.stream_events():
    if isinstance(event, RawResponsesStreamEvent):
        # Text token delta — accumulate for display
        delta = event.data.delta  # str fragment
        yield delta
    elif isinstance(event, RunItemStreamEvent):
        if isinstance(event.item, ToolCallOutputItem):
            # Tool finished — extract result from event.item.output
            tool_output = event.item.output  # JSON string from tool

FastAPI SSE Streaming Endpoint

Use sse-starlette for Server-Sent Events with proper event types:

from sse_starlette.sse import EventSourceResponse
from agents import Runner
import json

@router.post("/chat")
async def chat(request: ChatRequest, user = Depends(get_current_user)):
    async def event_generator():
        ctx = ChatContext(user_id=str(user.id), thread_id=str(thread.id))
        result = Runner.run_streamed(agent, messages, context=ctx)

        yield {"event": "stream_start", "data": json.dumps({"thread_id": str(thread.id)})}

        async for event in result.stream_events():
            if isinstance(event, RawResponsesStreamEvent) and event.data.delta:
                yield {"event": "text_token", "data": json.dumps({"token": event.data.delta})}
            elif isinstance(event, RunItemStreamEvent):
                if isinstance(event.item, ToolCallOutputItem):
                    yield {"event": "task_action", "data": json.dumps({"result": event.item.output})}

        yield {"event": "stream_end", "data": json.dumps({"final": result.final_output})}

    return EventSourceResponse(event_generator())

Frontend: Use fetch + ReadableStream, not EventSource — EventSource is GET-only and can't send Authorization headers or POST bodies.

Testable Tool Pattern (_impl separation)

Separate business logic from the @function_tool decorator for unit testing:

from agents import function_tool, RunContextWrapper

# Separated for monkeypatching in tests
def _get_session_factory():
    return get_session_factory()

# Testable implementation (no decorator)
async def _add_task_impl(ctx: RunContextWrapper[ChatContext], title: str) -> str:
    session_factory = _get_session_factory()
    async with session_factory() as session:
        task = Task(title=title, user_id=ctx.context.user_id)
        session.add(task)
        await session.commit()
        ctx.context.modified_tasks.append(("created", str(task.id)))
        return json.dumps({"id": str(task.id), "title": task.title})

# Thin decorator wrapper
@function_tool
async def add_task(ctx: RunContextWrapper[ChatContext], title: str) -> str:
    """Create a new task for the user.

    Args:
        title: The task title (1-500 characters)
    """
    return await _add_task_impl(ctx, title)

In tests: monkeypatch _get_session_factory to return a SQLite test session factory. Test the _impl functions directly.

RunContextWrapper Construction for Tests

from agents import RunContextWrapper

# Create context for test
ctx = RunContextWrapper(context=ChatContext(
    user_id="test-user-id",
    thread_id="test-thread-id",
))

# Call _impl directly
result = await _add_task_impl(ctx, "Buy groceries")

Standalone DB Sessions for Tools

Tools run outside FastAPI's dependency injection — they can't use Depends(get_session). Create standalone sessions:

# In tools.py — tools manage their own sessions
def _get_tool_session_factory():
    """Separated for test monkeypatching."""
    return get_session_factory()

async def _list_tasks_impl(ctx, completed_filter=None):
    session_factory = _get_tool_session_factory()
    async with session_factory() as session:
        query = select(Task).where(Task.user_id == ctx.context.user_id)
        # ...

Agent Error Response Format

Design tool error responses for agent consumption (not just humans):

return json.dumps({
    "error": "NOT_FOUND",
    "message": f"No task found with ID '{task_id}'",
    "suggestion": "Use list_tasks to see available tasks"
})

The suggestion field guides the agent toward recovery without re-prompting the user.

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
• tool-design - Designing tools agents use effectively