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.
Scanned 6/10/2026
Install via CLI
Skills Directoryopenskills install majiayu000/claude-skill-registry---
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
```bash
# 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-...
```
```python
# 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
```python
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
```python
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
```python
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)
```python
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)
```python
from agents import Agent, Runner
agent = Agent(name="Helper", instructions="Be helpful")
result = Runner.run_sync(agent, "Hello!")
print(result.final_output)
```
### Streaming Run
```python
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
```python
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
```python
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
```python
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
```python
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)
```python
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)
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
No comments yet. Be the first to comment!
