tomevault-io 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 messagingscaffolding-fastapi-dapr- FastAPI backend integrationstreaming-llm-responses- Response streaming patternsbuilding-chat-interfaces- Frontend chat UI
Converted and distributed by TomeVault — claim your Tome and manage your conversions.