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