Building Production Agents with OpenAI Agents SDK
OpenAIProductionTutorialDeployment
Build production-ready AI agents with OpenAI's official SDK, from basic setup to deployment.
Building Production Agents with OpenAI Agents SDK
Overview
OpenAI Agents SDK is the official framework for building production-ready AI agents. This tutorial covers the complete lifecycle from basic agent creation to production deployment.
Prerequisites
- Python 3.10+
- OpenAI API key
- Basic understanding of async Python
Installation
# Install the SDK
pip install openai-agents
# With extras for all providers
pip install openai-agents[all]
# Set your API key
export OPENAI_API_KEY=sk-...
Your First Agent
Basic Agent
# src/basic_agent.py
from openai_agents import Agent
# Create a simple agent
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant.",
model="gpt-4o"
)
# Run a query
result = agent.run("What is the capital of France?")
print(result.output)
With Tools
# src/agent_with_tools.py
from openai_agents import Agent, tool
@tool
def get_weather(location: str) -> str:
"""Get current weather for a location."""
# Implement weather API call
return f"The weather in {location} is 72°F and sunny."
@tool
def calculate_tax(amount: float, rate: float = 0.08) -> float:
"""Calculate sales tax for an amount."""
return amount * (1 + rate)
# Create agent with tools
agent = Agent(
model="gpt-4o",
tools=[get_weather, calculate_tax]
)
# Agent will automatically use tools when needed
result = agent.run("What's the weather in New York and how much tax on $100?")
Agent Architecture
Component Overview
┌─────────────────────────────────────────────────────────────┐
│ Agent │
├─────────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Model │◀──▶│ Tools │◀──▶│ Memory │ │
│ │ Layer │ │ Layer │ │ Layer │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ LLM │ │ Function│ │ Context │ │
│ │ Calls │ │ Calling │ │ Store │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘
Model Configuration
# src/model_config.py
from openai_agents import Agent
from openai_agents.models import OpenAIModel
# Direct model string
agent = Agent(model="gpt-4o")
# With explicit model config
model = OpenAIModel(
name="gpt-4o",
api_key="sk-...",
base_url="https://api.openai.com/v1",
temperature=0.7,
max_tokens=4096
)
agent = Agent(model=model)
# Multiple model support
agent = Agent(
model="gpt-4o",
fallback_models=["gpt-4o-mini", "gpt-3.5-turbo"]
)
Advanced Patterns
Multi-Agent Systems
# src/multi_agent.py
from openai_agents import Agent, Handoff
# Define specialized agents
researcher = Agent(
name="Researcher",
instructions="You are a research assistant. Find and summarize information.",
model="gpt-4o"
)
writer = Agent(
name="Writer",
instructions="You are a technical writer. Create clear, well-structured content.",
model="gpt-4o"
)
reviewer = Agent(
name="Reviewer",
instructions="You are a quality reviewer. Check for accuracy and clarity.",
model="gpt-4o"
)
# Create handoffs between agents
research_to_write = Handoff(
from_agent=researcher,
to_agent=writer,
trigger="when research is complete"
)
write_to_review = Handoff(
from_agent=writer,
to_agent=reviewer,
trigger="when draft is ready"
)
# Orchestrate workflow
def run_workflow(query: str):
# Step 1: Research
research_result = researcher.run(query)
# Step 2: Write (with research context)
write_prompt = f"Based on this research, write a comprehensive article:\n\n{research_result.output}"
draft = writer.run(write_prompt)
# Step 3: Review
review_prompt = f"Review this draft for accuracy and clarity:\n\n{draft.output}"
review = reviewer.run(review_prompt)
return {
'research': research_result,
'draft': draft,
'review': review,
'final': review.output
}
State Management
# src/state.py
from openai_agents import Agent, State
from dataclasses import dataclass, field
from typing import Any
@dataclass
class AgentState:
"""Custom state for your agent."""
conversation_history: list[dict] = field(default_factory=list)
user_preferences: dict = field(default_factory=dict)
current_task: str = ""
completed_steps: list[str] = field(default_factory=list)
context: dict[str, Any] = field(default_factory=dict)
# Create agent with state
agent = Agent(
model="gpt-4o",
state_type=AgentState,
initial_state=AgentState(
user_preferences={'tone': 'professional', 'format': 'markdown'}
)
)
# Access and modify state
result = agent.run("What's my preferred tone?")
print(agent.state.user_preferences)
# Update state
agent.state.completed_steps.append("analyzed_preferences")
Streaming
# src/streaming.py
from openai_agents import Agent
agent = Agent(model="gpt-4o")
# Stream response
async def stream_response(query: str):
async for chunk in agent.run_stream(query):
print(chunk.output, end='', flush=True)
# Stream with events
async def stream_with_events(query: str):
async for event in agent.iter_events(query):
if event.type == 'text':
print(event.data, end='', flush=True)
elif event.type == 'tool_call':
print(f"\n🔧 Calling tool: {event.data.name}")
elif event.type == 'thought':
print(f"\n💭 {event.data}")
import asyncio
asyncio.run(stream_with_events("Explain quantum computing"))
Production Patterns
Error Handling
# src/error_handling.py
from openai_agents import Agent
from openai_agents.errors import AgentError, ToolError, ModelError
class ResilientAgent:
def __init__(self, agent: Agent, max_retries: int = 3):
self.agent = agent
self.max_retries = max_retries
async def run_with_retry(self, query: str):
"""Run agent with automatic retry."""
for attempt in range(self.max_retries):
try:
return await self.agent.run(query)
except ModelError as e:
# Rate limit or model error
if attempt < self.max_retries - 1:
await asyncio.sleep(2 ** attempt) # Exponential backoff
continue
raise
except ToolError as e:
# Tool execution failed
# Log and continue with alternative
print(f"Tool error: {e}, trying alternative...")
continue
except AgentError as e:
# Agent-level error
print(f"Agent error: {e}")
raise
raise MaxRetriesExceeded(f"Failed after {self.max_retries} attempts")
# Usage
agent = Agent(model="gpt-4o")
resilient = ResilientAgent(agent)
result = await resilient.run_with_retry("Complex query")
Observability
# src/observability.py
from openai_agents import Agent
import logging
from datetime import datetime
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ObservableAgent:
def __init__(self, agent: Agent):
self.agent = agent
self.metrics = {
'total_runs': 0,
'total_tokens': 0,
'tool_calls': 0,
'errors': 0
}
async def run(self, query: str, context: dict = None):
"""Run with observability."""
start_time = datetime.now()
self.metrics['total_runs'] += 1
try:
result = await self.agent.run(query)
# Log metrics
duration = (datetime.now() - start_time).total_seconds()
self.metrics['total_tokens'] += result.usage.total_tokens
logger.info(
f"Agent run completed in {duration:.2f}s, "
f"tokens: {result.usage.total_tokens}"
)
return result
except Exception as e:
self.metrics['errors'] += 1
logger.error(f"Agent run failed: {e}")
raise
def get_metrics(self) -> dict:
"""Get current metrics."""
return self.metrics.copy()
def reset_metrics(self):
"""Reset metrics."""
self.metrics = {k: 0 for k in self.metrics}
Batch Processing
# src/batch.py
from openai_agents import Agent
import asyncio
async def process_batch(queries: list[str], agent: Agent, batch_size: int = 5):
"""Process multiple queries in parallel."""
results = []
for i in range(0, len(queries), batch_size):
batch = queries[i:i + batch_size]
# Process batch in parallel
tasks = [agent.run(query) for query in batch]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
for result in batch_results:
if isinstance(result, Exception):
results.append({'error': str(result)})
else:
results.append({'output': result.output})
# Rate limit delay
await asyncio.sleep(0.1)
return results
# Usage
agent = Agent(model="gpt-4o")
queries = ["Query 1", "Query 2", "Query 3", ...]
results = await process_batch(queries, agent)
Deployment
API Server
# src/api_server.py
from fastapi import FastAPI
from openai_agents import Agent
from pydantic import BaseModel
app = FastAPI(title="AI Agent API")
# Initialize agent
agent = Agent(model="gpt-4o", tools=[get_weather])
class QueryRequest(BaseModel):
query: str
stream: bool = False
class QueryResponse(BaseModel):
output: str
usage: dict
tool_calls: list
@app.post("/query")
async def query(request: QueryRequest):
"""Execute agent query."""
result = await agent.run(request.query)
return QueryResponse(
output=result.output,
usage={
'prompt_tokens': result.usage.prompt_tokens,
'completion_tokens': result.usage.completion_tokens,
'total_tokens': result.usage.total_tokens
},
tool_calls=[tc.model_dump() for tc in result.tool_calls]
)
@app.post("/query/stream")
async def query_stream(request: QueryRequest):
"""Stream agent response."""
async def generate():
async for chunk in agent.run_stream(request.query):
yield chunk.output
return StreamingResponse(generate(), media_type="text/event-stream")
# Run with: uvicorn src.api_server:app --host 0.0.0.0 --port 8000
Docker Deployment
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application
COPY src/ ./src/
# Environment variables
ENV OPENAI_API_KEY=${OPENAI_API_KEY}
# Health check
HEALTHCHECK --interval=30s --timeout=10s \
CMD python -c "import requests; requests.get('http://localhost:8000/health')"
# Run server
CMD ["uvicorn", "src.api_server:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
agent-api:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
deploy:
resources:
limits:
memory: 2G
reservations:
memory: 512M
restart: unless-stopped
Best Practices
1. Prompt Engineering
# Good: Specific and structured
agent = Agent(
model="gpt-4o",
instructions="""
You are a technical documentation writer.
Guidelines:
- Use clear, concise language
- Include code examples when relevant
- Structure with headings and bullet points
- Cite sources when making claims
Output Format:
- Start with a brief summary
- Follow with detailed explanation
- End with key takeaways
"""
)
# Bad: Vague
agent = Agent(
model="gpt-4o",
instructions="Be helpful and write good documentation."
)
2. Tool Design
# Good: Clear, focused tools
@tool
def search_documents(query: str, max_results: int = 5) -> list[dict]:
"""Search the documentation database for relevant articles.
Args:
query: Search query string
max_results: Maximum number of results to return (default: 5)
Returns:
List of matching documents with title, url, and snippet
"""
...
# Bad: Vague, overloaded tools
@tool
def search(query: str) -> any:
"""Search for things."""
...
3. Context Management
# Good: Explicit context passing
class TaskContext(BaseModel):
project_name: str
user_role: str
deadline: datetime
constraints: list[str]
agent = Agent(
model="gpt-4o",
deps_type=TaskContext,
instructions=lambda ctx: f"""
You are helping {ctx.user_role} with the {ctx.project_name} project.
Deadline: {ctx.deadline}
Constraints: {', '.join(ctx.constraints)}
"""
)
# Bad: Implicit context
agent = Agent(
model="gpt-4o",
instructions="Help with the project."
)
Troubleshooting
Common Issues
Rate limits:
import asyncio
from openai_agents import Agent
agent = Agent(model="gpt-4o")
async def rate_limited_run(query: str, delay: float = 1.0):
await asyncio.sleep(delay)
return await agent.run(query)
Token limits:
# Use smaller model for simple tasks
agent = Agent(model="gpt-4o-mini")
# Or truncate context
def truncate_context(context: str, max_tokens: int = 4000) -> str:
# Implement truncation logic
...
Tool calling issues:
# Ensure tool has clear description
@tool
def calculate_total(items: list[dict]) -> float:
"""Calculate total price for a list of items.
Each item should have 'price' and 'quantity' fields.
"""
...
Resources
- OpenAI Agents SDK Docs: https://platform.openai.com/docs/agents
- GitHub: https://github.com/openai/agents
- API Reference: https://platform.openai.com/docs/api-reference
Last updated: May 2026