Overview
This guide demonstrates how to leverage the OpenAI Agents SDK with FastAPI to create scalable, production-ready AI agent systems. The OpenAI Agents SDK provides a robust framework for building structured AI agents, while FastAPI offers high-performance API exposure with automatic documentation and validation.
🤖 OpenAI Agents SDK: The Foundation
Core Components
The OpenAI Agents SDK provides several key abstractions:
Agent: The core AI entity with specific instructions and capabilities
Runner: Execution engine for running agents
AgentOutputSchema: Structured output validation using Pydantic
Model Integration: Seamless integration with OpenAI's latest models
🏗️ Architecture Overview
The system follows a clean separation of concerns:
- Backend: Python-based AI agents orchestrated through FastAPI
- Frontend: Modern web application with responsive design
- Integration: RESTful APIs connecting the two layers
🤖 Building AI Agents: The Core Pattern
1. Agent Definition Structure
Every agent in the system follows a consistent pattern using the agents framework:
from pydantic import BaseModel
from agents import Agent, AgentOutputSchema
class AgentOutput(BaseModel):
"""Structured output schema for the agent"""
result_data: str
success: bool
message: str
AGENT_INSTRUCTIONS = """
You are a specialized AI agent that performs specific tasks.
Your instructions define the agent's behavior and expertise.
"""
agent = Agent(
name="SpecializedAgent",
instructions=AGENT_INSTRUCTIONS,
model="gpt-4o",
output_type=AgentOutputSchema(AgentOutput, strict_json_schema=False),
)
2. Real-World Example: Database Schema Agent
Here's how the project implements a database schema generation agent:
from database_research.create_db_schema import DatabaseSchema
SCHEMA_GENERATION_INSTRUCTIONS = """
You are a senior database architect specialized in creating database schemas
from functional requirements. Analyze requirements and generate normalized
database schemas with proper relationships and constraints.
"""
create_db_schema_agent = Agent(
name="CreateDBSchemaAgent",
instructions=SCHEMA_GENERATION_INSTRUCTIONS,
model="gpt-4o",
output_type=AgentOutputSchema(DatabaseSchema, strict_json_schema=False),
)
3. Agent Execution Pattern
Agents are executed using the Runner pattern:
from agents import Runner
async def generate_database_schema(requirements: list[str]) -> DatabaseSchema:
input_text = f"Functional Requirements:\n" + "\n".join(requirements)
result = await Runner.run(
create_db_schema_agent,
input_text,
)
return result.final_output_as(DatabaseSchema)
🚀 Exposing Agents via FastAPI
1. FastAPI Server Setup
The project uses FastAPI to create a production-ready API server:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title="AI Agents API", version="1.0.0")
class AgentRequest(BaseModel):
input_data: str
parameters: dict = {}
@app.post("/agent/process")
async def process_with_agent(request: AgentRequest):
try:
# Execute agent logic
result = await run_agent_workflow(request.input_data)
return {"success": True, "data": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
2. Multi-Agent Orchestration
The system implements complex workflows that chain multiple agents:
class PRDResearchManager:
async def run(self, prd_text: str, db_instructions: str = ""):
# Step 1: Extract requirements
requirements = await self.extract_functional_requirements(prd_text)
# Step 2: Generate database schema
if requirements.success:
schema = await self.generate_database_schema(
requirements.requirements, db_instructions
)
# Step 3: Generate API contracts
contracts = await self.generate_api_contracts(schema, prd_text)
return {
"requirements": requirements,
"schema": schema,
"contracts": contracts
}
3. Streaming Responses
For long-running agent operations, the system supports streaming responses:
from fastapi.responses import StreamingResponse
@app.post("/research/stream")
async def research_stream_endpoint(request: ResearchRequest):
async def generate_updates():
manager = ResearchManager()
async for update in manager.run(request.query):
yield f"data: {json.dumps({'update': str(update)})}\n\n"
return StreamingResponse(
generate_updates(),
media_type="text/plain",
headers={"Cache-Control": "no-cache"}
)
🔧 Configuration Management
The system implements secure configuration management:
class Config:
def get_openai_api_key(self) -> str:
# Try config file first
api_key = self._config.get("openai_api_key")
# Fallback to environment variable
if not api_key:
api_key = os.getenv("OPENAI_API_KEY")
return api_key
def set_environment_variables(self):
api_key = self.get_openai_api_key()
os.environ["OPENAI_API_KEY"] = api_key
🎨 Frontend Integration
Modern Web Interface
The frontend provides an intuitive interface for interacting with AI agents:
async function callAgentAPI(endpoint, payload) {
const response = await fetch(`/api/${endpoint}`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload)
});
return await response.json();
}
// Example: Generate database schema
async function generateDatabaseSchema() {
const result = await callAgentAPI('database', {
prd_text: prdContent,
db_instructions: instructions
});
displayResults(result);
}
Proxy Architecture
The UI server acts as a proxy to the AI agents backend:
// UI Server (Node.js/Express)
app.post('/api/database', async (req, res) => {
const response = await fetch('http://localhost:8000/database', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(req.body)
});
const data = await response.json();
res.json(data);
});
🛠️ Key Agent Types in the System
1. Research Agent
- Purpose: Web search and information gathering
- Input: Search queries and research topics
- Output: Comprehensive research reports
2. Database Schema Agent
- Purpose: Generate database schemas from requirements
- Input: Functional requirements and constraints
- Output: Complete database schema with tables, relationships
3. Contract Generation Agent
- Purpose: Create OpenAPI specifications from database schemas
- Input: Database schema and business requirements
- Output: Complete OpenAPI 3.0.3 specification
4. Sequence Diagram Agent
- Purpose: Generate PlantUML sequence diagrams
- Input: Architecture requirements and constraints
- Output: PlantUML diagram code
📊 Benefits of This Architecture
Scalability
- Each agent is independently deployable
- FastAPI provides async support for concurrent requests
- Streaming responses prevent timeouts on long operations
Maintainability
- Clear separation between agent logic and API exposure
- Consistent patterns across all agents
- Comprehensive error handling and logging
Flexibility
- Framework-agnostic OpenAPI specifications
- Multiple output formats (JSON, streaming, files)
- Easy to add new agents following established patterns
🚀 Getting Started
- Setup Backend:
cd server
pip install -r requirements.txt
cp config.template.json config.json
# Add your OpenAI API key to config.json
python server.py
- Setup Frontend:
cd UI
npm install
npm start
-
Test the System:
- Navigate to
http://localhost:3000 - Try generating a database schema from requirements
- Explore the generated API contracts
- Navigate to
💡 Best Practices Learned
- Agent Design: Keep agents focused on single responsibilities
- Error Handling: Always provide meaningful error responses
- Configuration: Use secure configuration management
- API Design: Follow REST conventions and provide clear documentation
- Frontend Integration: Use proxy patterns for clean separation
This architecture demonstrates how to build production-ready AI agent systems that can scale from prototype to enterprise deployment. The combination of structured agents, robust APIs, and modern web interfaces creates a powerful platform for AI-driven development workflows.
Top comments (0)