| name | fastapi-development |
| description | Modern Python API development with FastAPI covering async patterns, Pydantic validation, dependency injection, and production deployment |
FastAPI Development
A comprehensive skill for building modern, high-performance Python APIs with FastAPI. Master async/await patterns, Pydantic data validation, dependency injection, authentication, database integration, and production-ready deployment strategies.
When to Use This Skill
Use this skill when:
- Building RESTful APIs with Python for web, mobile, or microservices
- Developing high-performance, asynchronous backend services
- Creating APIs with automatic interactive documentation (OpenAPI/Swagger)
- Implementing OAuth2, JWT authentication, or other security patterns
- Integrating with SQL or NoSQL databases in Python applications
- Building APIs that require strong data validation and type safety
- Developing microservices with automatic request/response validation
- Creating APIs with WebSocket support for real-time features
- Migrating from Flask, Django REST Framework, or other Python frameworks
- Building production-ready APIs with proper error handling and testing
- create FastAPI endpoint with SSE streaming for agent communication
- implement Unix socket IPC for osxphotos sandbox integration
- add circuit breaker for sidecar management and crash recovery
Core Concepts
FastAPI Philosophy
FastAPI is built on three foundational principles:
- Fast to Code: Reduce development time with automatic validation and documentation
- Fast to Run: High performance comparable to NodeJS and Go (via Starlette and Pydantic)
- Fewer Bugs: Automatic validation reduces human errors by about 40%
- Standards-Based: Built on OpenAPI and JSON Schema standards
- Editor Support: Full autocomplete, type checking, and inline documentation
Key FastAPI Features
- Type Hints: Python 3.6+ type hints for validation and documentation
- Async Support: Native async/await for high-performance I/O operations
- Pydantic Models: Automatic request/response validation and serialization
- Dependency Injection: Elegant system for sharing logic across endpoints
- OpenAPI Docs: Automatic interactive API documentation
- Security: Built-in support for OAuth2, JWT, API keys, and more
- Testing: Easy to test with TestClient and async test support
Core Architecture Components
- FastAPI App: The main application instance
- Path Operations: Endpoint definitions with HTTP methods
- Pydantic Models: Data validation and serialization schemas
- Dependencies: Reusable logic for authentication, database, etc.
- Routers: Organize endpoints into modules
- Middleware: Process requests/responses globally
- Background Tasks: Execute code after returning responses
Getting Started
Installation
pip install fastapi
pip install "fastapi[all]"
pip install fastapi uvicorn[standard]
pip install python-multipart
pip install python-jose[cryptography]
pip install passlib[bcrypt]
pip install sqlalchemy
pip install databases
Minimal FastAPI Application
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
Pydantic Models for Data Validation
Basic Model Definition
from pydantic import BaseModel, Field, EmailStr, HttpUrl
from typing import Optional, List
from datetime import datetime
class User(BaseModel):
id: int
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
full_name: Optional[str] = None
is_active: bool = True
created_at: datetime = Field(default_factory=datetime.utcnow)
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
password: str = Field(..., min_length=8)
full_name: Optional[str] = None
class UserResponse(BaseModel):
id: int
username: str
email: EmailStr
full_name: Optional[str]
is_active: bool
class Config:
orm_mode = True
TRAE_Extractor-app: Agent Configuration Models
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Literal, Dict, Any
class AgentToolConfig(BaseModel):
"""Configuration for a single agent tool"""
name: str = Field(..., description="Tool name")
type: Literal["mcp", "native", "api", "think", "memory", "filesystem"] = Field(
...,
description="Tool type"
)
config: Dict[str, Any] = Field(
default_factory=dict,
description="Tool-specific configuration"
)
command: Optional[str] = Field(None, description="Command for MCP tools")
args: Optional[List[str]] = Field(None, description="Command arguments")
env: Optional[Dict[str, str]] = Field(None, description="Environment variables")
class AgentConfig(BaseModel):
"""Configuration for a single agent in Cagent team"""
name: str = Field(..., min_length=, max_length=)
model: = Field(..., description=)
description: = Field(..., min_length=, description=)
instruction: = Field(..., min_length=, description=)
toolsets: [AgentToolConfig] = Field(
default_factory=,
description=
)
rag: [[]] = Field(
,
description=
)
sub_agents: [[]] = Field(
,
description=
)
add_prompt_files: [[]] = Field(
,
description=
)
():
version: = Field(..., pattern=)
models: [, ] = Field(
...,
description=
)
agents: [, AgentConfig] = Field(
...,
description=
)
rag: [[, ]] = Field(
,
description=
)
metadata: [[, ]] = Field(
,
description=
)
config = CagentTeamConfig(
version=,
models={
: {
: ,
: ,
:
}
},
agents={
: AgentConfig(
name=,
model=,
description=,
instruction=,
toolsets=[
AgentToolConfig(
=,
command=,
args=[, ],
env={: }
)
],
rag=[, ]
)
}
)
TRAE_Extractor-app: Brand Knowledge Schemas
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Dict
from enum import Enum
class ToneStyle(str, Enum):
"""Brand tone styles"""
INFORMAL_CONSCIOUS = "informal_conscious"
PASSIONATE = "passionate"
EDUCATIONAL = "educational"
DIRECT = "direct"
class BrandGuideline(BaseModel):
"""Single brand guideline entry"""
category: str = Field(..., description="Guideline category")
content: str = Field(..., min_length=10, description="Guideline content")
priority: Literal["high", "medium", "low"] = Field("medium")
examples: Optional[List[str]] = Field(None, description="Example messages")
class BrandTone(BaseModel):
"""Brand tone of voice configuration"""
style: ToneStyle = Field(..., description="Primary tone style")
characteristics: List[str] = Field(
...,
min_items=1,
description="Tone characteristics"
)
do_examples: List[] = Field(
...,
min_items=,
description=
)
dont_examples: [] = Field(
...,
min_items=,
description=
)
():
brand_name: = Field(..., min_length=)
mission: = Field(..., min_length=)
core_values: [] = Field(..., min_items=)
tone_guidelines: BrandTone
visual_identity: [[, ]] = Field()
communication_principles: [] = Field(..., min_items=)
key_messages: [] = Field(..., min_items=)
audience_segments: [[, [, ]]] = Field()
hashtags: [, []] = Field(
default_factory=,
description=
)
slow_food_knowledge = BrandKnowledge(
brand_name=,
mission=,
core_values=[
,
,
,
],
tone_guidelines=BrandTone(
style=ToneStyle.INFORMAL_CONSCIOUS,
characteristics=[, ],
do_examples=[
],
dont_examples=[
,
]
),
communication_principles=[
,
,
,
,
],
key_messages=[
,
],
hashtags={
: [, ],
: [, ],
: [, ]
}
)
TRAE_Extractor-app: Media Metadata Validation
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Tuple
from datetime import datetime
from enum import Enum
class MediaType(str, Enum):
"""Media file types"""
IMAGE = "image"
VIDEO = "video"
class EXIFData(BaseModel):
"""EXIF metadata from photos"""
camera_make: Optional[str] = Field(None, max_length=50)
camera_model: Optional[str] = Field(None, max_length=50)
date_taken: Optional[datetime] = Field(None)
gps_coordinates: Optional[Tuple[float, float]] = Field(None)
iso: Optional[int] = Field(None, ge=100, le=6400)
aperture: Optional[float] = Field(None, ge=1.0, le=32.0)
shutter_speed: Optional[str] = Field(None, max_length=20)
focal_length: Optional[] = Field(, ge=, le=)
():
filename: = Field(..., pattern=)
file_size_bytes: = Field(..., gt=)
media_type: MediaType = Field(..., description=)
exif_data: [EXIFData] = Field()
tags: [] = Field(
default_factory=,
max_length=,
description=
)
album: [] = Field(, max_length=)
date_imported: datetime = Field(default_factory=datetime.utcnow)
cloudinary_public_id: [] = Field()
cloudinary_url: [] = Field(, regex=)
():
valid_extensions = [, , , , ]
(v.lower().endswith(ext) ext valid_extensions):
ValueError()
v
():
album: = Field(..., description=)
export_path: = Field(..., description=)
media_count: = Field(..., ge=, description=)
media_files: [MediaMetadata] = Field(..., description=)
extraction_time: = Field(..., gt=, description=)
success: = Field(..., description=)
extraction_result = MediaExtractionResult(
album=,
export_path=,
media_count=,
media_files=[
MediaMetadata(
filename=,
file_size_bytes=,
media_type=MediaType.IMAGE,
exif_data=EXIFData(
camera_make=,
camera_model=,
date_taken=datetime(, , , , , ),
iso=,
aperture=
),
tags=[, , ],
album=
)
],
extraction_time=,
success=
)
TRAE_Extractor-app: Post Scheduling Data Structures
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum
class SocialPlatform(str, Enum):
"""Supported social media platforms"""
INSTAGRAM = "instagram"
FACEBOOK = "facebook"
LINKEDIN = "linkedin"
TWITTER = "twitter"
TIKTOK = "tiktok"
class PostStatus(str, Enum):
"""Post scheduling status"""
PENDING = "pending"
SCHEDULED = "scheduled"
PUBLISHED = "published"
FAILED = "failed"
CANCELLED = "cancelled"
class ScheduledPost(BaseModel):
"""Scheduled social media post"""
post_id: Optional[str] = Field(None, description="Post ID from platform")
media_id: str = Field(..., description="Cloudinary public ID or media file ID")
caption: str = Field(..., min_length=10, max_length=2200, description="Post caption")
platforms: List[SocialPlatform] = Field(..., min_items=1, description="Target platforms")
scheduled_at: datetime = Field(..., description=)
hashtags: [] = Field(
default_factory=,
max_length=,
description=
)
status: PostStatus = Field(default=PostStatus.PENDING)
created_at: datetime = Field(default_factory=datetime.utcnow)
published_at: [datetime] = Field()
error_message: [] = Field(, description=)
():
v <= datetime.utcnow():
ValueError()
v
():
platform: SocialPlatform = Field(..., description=)
caption: = Field(..., min_length=, max_length=)
hashtags: [] = Field(default_factory=, max_length=)
character_limit: [] = Field(, description=)
():
media_id: = Field(..., description=)
variations: [CaptionVariation] = Field(..., min_items=)
suggested_hashtags: [] = Field(default_factory=, max_length=)
tone_analysis: [[, ]] = Field(, description=)
generation_time: = Field(..., gt=)
scheduled_post = ScheduledPost(
media_id=,
caption=,
platforms=[SocialPlatform.INSTAGRAM, SocialPlatform.FACEBOOK],
scheduled_at=datetime(, , , , , ),
hashtags=[, , , ]
)
Nested Models
class Image(BaseModel):
url: HttpUrl
name: str
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float = Field(..., gt=0)
tax: Optional[float] = None
tags: List[str] = []
images: Optional[List[Image]] = None
Model Validation and Examples
from pydantic import BaseModel, Field, validator
class Product(BaseModel):
name: str = Field(..., example="MacBook Pro")
price: float = Field(..., gt=0, example=1999.99)
discount: Optional[float] = Field(None, ge=0, le=100, example=10.0)
@validator('discount')
def discount_check(cls, v, values):
if v and 'price' in values:
discounted = values['price'] * (1 - v/100)
if discounted < 0:
raise ValueError('Discounted price cannot be negative')
return v
class Config:
schema_extra = {
"example": {
"name": "MacBook Pro 16",
"price": 2499.99,
"discount": 15.0
}
}
Path Operations and Routing
HTTP Methods and Path Parameters
from fastapi import FastAPI, Path, Query, Body
from typing import Optional
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(
item_id: int = Path(..., title="The ID of the item", ge=1),
q: Optional[str] = Query(None, max_length=50)
):
return {"item_id": item_id, "q": q}
@app.post("/items/")
async def create_item(item: Item):
return {"item": item, "message": "Item created"}
@app.put("/items/{item_id}")
async def update_item(
item_id: int,
item: Item = Body(...),
):
return {"item_id": item_id, "item": item}
@app.delete("/items/{item_id}")
():
{: }
():
{: item_id, : item}
Query Parameters with Validation
from fastapi import Query
from typing import List, Optional
@app.get("/search/")
async def search_items(
q: str = Query(..., min_length=3, max_length=50),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
sort_by: Optional[str] = Query(None, regex="^(name|price|date)$"),
tags: List[str] = Query([], description="Filter by tags")
):
return {
"q": q,
"skip": skip,
"limit": limit,
"sort_by": sort_by,
"tags": tags
}
Response Models
from typing import List
@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
db_user = {
"id": 1,
"username": user.username,
"email": user.email,
"full_name": user.full_name,
"is_active": True
}
return db_user
@app.get("/users/", response_model=List[UserResponse])
async def list_users(skip: int = 0, limit: int = 100):
users = [...]
return users
class UserInDB(User):
hashed_password: str
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
user = get_user_from_db(user_id)
return user
Async/Await Patterns
When to Use async vs def
@app.get("/async-example")
async def async_endpoint():
result = await async_database_query()
external_data = await async_http_call()
return {"result": result, "external": external_data}
@app.get("/sync-example")
def sync_endpoint():
result = synchronous_database_query()
return {"result": result}
Async Database Operations
import asyncio
from databases import Database
DATABASE_URL = "postgresql://user:password@localhost/dbname"
database = Database(DATABASE_URL)
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
@app.get("/users/{user_id}")
async def get_user(user_id: int):
query = "SELECT * FROM users WHERE id = :user_id"
user = await database.fetch_one(query, {"user_id": user_id})
return user
@app.post("/users/")
async def create_user(user: UserCreate):
query = """
INSERT INTO users (username, email, hashed_password)
VALUES (:username, :email, :password)
RETURNING *
"""
hashed_password = hash_password(user.password)
new_user = await database.fetch_one(
query,
{
"username": user.username,
"email": user.email,
"password": hashed_password
}
)
return new_user
Concurrent Operations
import asyncio
import httpx
async def fetch_user(user_id: int):
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.example.com/users/{user_id}")
return response.json()
@app.get("/users/batch")
async def get_multiple_users(user_ids: List[int] = Query(...)):
users = await asyncio.gather(*[fetch_user(uid) for uid in user_ids])
return {"users": users}
Dependency Injection
Basic Dependencies
from fastapi import Depends
from typing import Optional
async def common_parameters(
q: Optional[str] = None,
skip: int = 0,
limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return commons
Class-Based Dependencies
class Pagination:
def __init__(
self,
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=100)
):
self.skip = skip
self.limit = limit
@app.get("/items/")
async def list_items(pagination: Pagination = Depends()):
return {
"skip": pagination.skip,
"limit": pagination.limit,
"items": []
}
Database Session Dependency
from sqlalchemy.orm import Session
from typing import Generator
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
db: Session = Depends(get_db)
):
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
Sub-Dependencies
from fastapi import Header, HTTPException
async def verify_token(x_token: str = Header(...)):
if x_token != "secret-token":
raise HTTPException(status_code=400, detail="Invalid token")
return x_token
async def verify_key(x_key: str = Header(...)):
if x_key != "secret-key":
raise HTTPException(status_code=400, detail="Invalid key")
return x_key
async def verify_credentials(
token: str = Depends(verify_token),
key: str = Depends(verify_key)
):
return {"token": token, "key": key}
@app.get("/protected/")
async def protected_route(credentials: dict = Depends(verify_credentials)):
return {"message": "Access granted", "credentials": credentials}
Global Dependencies
async def log_requests():
print("Request received")
app = FastAPI(dependencies=[Depends(log_requests)])
Authentication and Security
OAuth2 Password Bearer with JWT
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
:
expire = datetime.utcnow() + timedelta(minutes=)
to_encode.update({: expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
encoded_jwt
():
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=,
headers={: },
)
:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: = payload.get()
username :
credentials_exception
token_data = TokenData(username=username)
JWTError:
credentials_exception
user = get_user_from_db(username=token_data.username)
user :
credentials_exception
user
():
current_user.is_active:
HTTPException(status_code=, detail=)
current_user
():
user = authenticate_user(form_data.username, form_data.password)
user:
HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=,
headers={: },
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={: user.username}, expires_delta=access_token_expires
)
{: access_token, : }
():
current_user
API Key Authentication
from fastapi import Security
from fastapi.security import APIKeyHeader
API_KEY = "your-api-key"
api_key_header = APIKeyHeader(name="X-API-Key")
async def verify_api_key(api_key: str = Security(api_key_header)):
if api_key != API_KEY:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid API Key"
)
return api_key
@app.get("/secure-data")
async def get_secure_data(api_key: str = Depends(verify_api_key)):
return {"data": "This is secure data"}
OAuth2 with Scopes
from fastapi.security import OAuth2PasswordBearer, SecurityScopes
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="token",
scopes={
"items:read": "Read items",
"items:write": "Create and update items",
"users:read": "Read user information"
}
)
async def get_current_user_with_scopes(
security_scopes: SecurityScopes,
token: str = Depends(oauth2_scheme)
):
if security_scopes.scopes:
authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
else:
authenticate_value = "Bearer"
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": authenticate_value},
)
return user
@app.get("/items/", dependencies=[Security(get_current_user_with_scopes, scopes=["items:read"])])
async def read_items():
return [{"item": "Item 1"}, {"item": "Item 2"}]
@app.post("/items/", dependencies=[Security(get_current_user_with_scopes, scopes=["items:write"])])
():
{: item}
Database Integration
SQLAlchemy Setup
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class UserModel(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
Base.metadata.create_all(bind=engine)
CRUD Operations
from sqlalchemy.orm import Session
@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = UserModel(
username=user.username,
email=user.email,
hashed_password=get_password_hash(user.password)
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
async def read_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(UserModel).filter(UserModel.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.put("/users/{user_id}", response_model=UserResponse)
async def update_user(
user_id: int,
user_update: UserCreate,
db: Session = Depends(get_db)
):
user = db.query(UserModel).filter(UserModel.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail=)
user.username = user_update.username
user.email = user_update.email
user_update.password:
user.hashed_password = get_password_hash(user_update.password)
db.commit()
db.refresh(user)
user
():
user = db.query(UserModel).(UserModel. == user_id).first()
user:
HTTPException(status_code=, detail=)
db.delete(user)
db.commit()
{: }
Background Tasks
from fastapi import BackgroundTasks
def send_email(email: str, message: str):
print(f"Sending email to {email}: {message}")
def process_file(filename: str):
print(f"Processing file: {filename}")
@app.post("/send-notification/")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(send_email, email, "Welcome!")
return {"message": "Notification scheduled"}
@app.post("/upload/")
async def upload_file(
file: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(process_file, file)
return {"message": "File uploaded, processing in background"}
SSE Streaming for Agent Communication
Server-Sent Events (SSE) enable real-time streaming of agent updates from the Python sidecar to the Electron renderer. This pattern is essential for providing live feedback during long-running agent operations.
Basic SSE Endpoint
from fastapi import APIRouter
from sse_starlette.sse import EventSourceResponse
import asyncio
from typing import AsyncGenerator
agent_router = APIRouter(prefix="/agent", tags=["agent"])
event_queues: dict[str, asyncio.Queue] = {}
class StreamEvent(BaseModel):
"""Event streamed via SSE"""
event_type: str
data: dict
timestamp: float
async def agent_event_generator(request_id: str) -> AsyncGenerator:
"""Generate events from agent event queue for SSE streaming"""
if request_id not in event_queues:
event_queues[request_id] = asyncio.Queue()
queue = event_queues[request_id]
try:
while True:
try:
event = await asyncio.wait_for(queue.get(), timeout=30.0)
yield {
"event": event.event_type,
"data": event.json(),
}
if event.event_type (, ):
asyncio.TimeoutError:
{
: ,
: ,
}
asyncio.CancelledError:
:
request_id event_queues:
event_queues[request_id]
():
EventSourceResponse(
agent_event_generator(request_id),
media_type=,
)
Sending Events to Stream
async def send_agent_event(request_id: str, event_type: str, data: dict):
"""Send event to agent stream"""
if request_id in event_queues:
event = StreamEvent(
event_type=event_type,
data=data,
timestamp=time.time()
)
await event_queues[request_id].put(event)
async def execute_agent_with_streaming(agent_id: str, request_id: str, input_data: dict):
await send_agent_event(request_id, "thinking", {"agent": agent_id, "status": "starting"})
result = await run_agent(agent_id, input_data)
await send_agent_event(request_id, "result", {"agent": agent_id, "result": result})
Client-Side SSE Consumption
const eventSource = new EventSource('http://localhost:8000/agent/stream/req-123');
eventSource.addEventListener('thinking', (e) => {
const data = JSON.parse(e.data);
console.log('Agent thinking:', data);
});
eventSource.addEventListener('result', (e) => {
const data = JSON.parse(e.data);
console.log('Agent result:', data);
eventSource.close();
});
eventSource.onerror = () => {
console.error('SSE connection error');
eventSource.close();
};
Unix Socket IPC with osxphotos Sandbox
Unix socket IPC enables secure communication between the Python sidecar and the osxphotos sandbox process. This pattern ensures osxphotos runs in isolation with no network access and strict path whitelisting.
Security Considerations
- NO network access: osxphotos process cannot make HTTP requests
- Read-only on Photos Library: osxphotos can only read from Apple Photos
- Write whitelist: Export paths must be pre-approved directories
- Sandbox isolation: Process runs in restricted environment
Unix Socket Client
import socket
import json
import os
from typing import Optional, Dict, Any
class OsxphotosSandboxClient:
"""Client for communicating with osxphotos sandbox via Unix socket"""
def __init__(self, socket_path: str):
self.socket_path = socket_path
self.timeout = 30
async def extract_photos(
self,
album: str,
export_path: str,
options: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Extract photos from album via osxphotos sandbox.
Args:
album: Photos album name
export_path: Destination directory (must be in whitelist)
options: Additional extraction options
Returns:
Dict with extraction results
Raises:
ValueError: If export path is not in whitelist
ConnectionError: If socket connection fails
"""
if not self._is_path_allowed(export_path):
raise ValueError(
f"Export path not in whitelist: {export_path}. "
f"Allowed paths: "
)
request = {
: ,
: {
: album,
: export_path,
: options {}
}
}
:
socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s:
s.settimeout(.timeout)
s.connect(.socket_path)
s.sendall(json.dumps(request).encode())
response_data =
:
chunk = s.recv()
chunk:
response_data += chunk
response = json.loads(response_data.decode())
response.get() == :
RuntimeError(response.get(, ))
response
socket.timeout:
ConnectionError()
FileNotFoundError:
ConnectionError()
Exception e:
ConnectionError()
() -> :
expanded_path = os.path.expanduser(path)
allowed_paths = ._get_allowed_paths()
(expanded_path.startswith(p) p allowed_paths)
() -> []:
[
os.path.expanduser(),
os.path.expanduser(),
os.path.expanduser(),
]
Unix Socket Server (osxphotos Side)
import socket
import json
import os
from pathlib import Path
class OsxphotosSandboxServer:
"""Unix socket server for osxphotos sandbox"""
def __init__(self, socket_path: str):
self.socket_path = socket_path
self.running = False
def start(self):
"""Start the Unix socket server"""
if os.path.exists(self.socket_path):
os.unlink(self.socket_path)
server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server_socket.bind(self.socket_path)
server_socket.listen(5)
self.running = True
print(f"osxphotos sandbox server listening on {self.socket_path}")
while self.running:
try:
conn, _ = server_socket.accept()
self._handle_request(conn)
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error handling connection: {e}")
server_socket.close()
def ():
:
request_data =
:
chunk = conn.recv()
chunk:
request_data += chunk
request = json.loads(request_data.decode())
request[] == :
response = ._extract_photos(request[])
:
response = {
: ,
:
}
conn.sendall(json.dumps(response).encode())
Exception e:
error_response = {
: ,
: (e)
}
conn.sendall(json.dumps(error_response).encode())
:
conn.close()
() -> :
osxphotos
album = params[]
export_path = params[]
options = params.get(, {})
._is_path_allowed(export_path):
{
: ,
:
}
photos = osxphotos.export(
album,
export_path,
**options
)
{
: ,
: photos,
: (photos)
}
() -> :
expanded_path = os.path.expanduser(path)
allowed_paths = [
os.path.expanduser(),
os.path.expanduser(),
os.path.expanduser(),
]
(expanded_path.startswith(p) p allowed_paths)
():
.running =
os.path.exists(.socket_path):
os.unlink(.socket_path)
Integration with FastAPI
from fastapi import FastAPI, HTTPException
app = FastAPI()
osxphotos_client = OsxphotosSandboxClient("/tmp/osxphotos.sock")
@app.post("/photos/extract")
async def extract_photos(album: str, export_path: str):
"""Extract photos from album via osxphotos sandbox"""
try:
result = await osxphotos_client.extract_photos(album, export_path)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except ConnectionError as e:
raise HTTPException(status_code=503, detail="osxphotos sandbox not available")
Circuit Breaker for Sidecar Lifecycle
Circuit breaker pattern prevents cascading failures when the Python sidecar crashes repeatedly. It tracks failures and temporarily stops restart attempts after a threshold.
Circuit Breaker Implementation
from datetime import datetime, timedelta
from typing import List, Optional
import logging
logger = logging.getLogger(__name__)
class CircuitBreaker:
"""Circuit breaker for sidecar lifecycle management"""
def __init__(
self,
max_failures: int = 3,
window_minutes: int = 5,
cooldown_minutes: int = 10
):
"""
Initialize circuit breaker.
Args:
max_failures: Maximum failures before opening circuit
window_minutes: Time window to count failures
cooldown_minutes: Time to wait before retrying after opening
"""
self.max_failures = max_failures
self.window_minutes = window_minutes
self.cooldown_minutes = cooldown_minutes
self.failures: List[datetime] = []
self.state = "closed"
self.last_failure_time: Optional[datetime] = None
def record_failure(self):
"""Record a failure and update circuit state"""
now = datetime.now()
self.failures.append(now)
self.last_failure_time = now
self.failures = [
f f .failures
now - f < timedelta(minutes=.window_minutes)
]
(.failures) >= .max_failures:
.state =
logger.warning(
)
():
.failures.clear()
.state =
.last_failure_time =
logger.info()
() -> :
.state == :
.state == :
.last_failure_time:
time_since_failure = datetime.now() - .last_failure_time
time_since_failure >= timedelta(minutes=.cooldown_minutes):
.state =
logger.info()
.state == :
() -> :
{
: .state,
: (.failures),
: .max_failures,
: .last_failure_time.isoformat() .last_failure_time ,
: .can_execute()
}
Integration with Sidecar Manager
from typing import Optional
class SidecarManager:
"""Manages Python sidecar lifecycle with circuit breaker"""
def __init__(self):
self.circuit_breaker = CircuitBreaker(
max_failures=3,
window_minutes=5,
cooldown_minutes=10
)
self.process: Optional[subprocess.Popen] = None
async def start(self) -> bool:
"""Start sidecar if circuit breaker allows"""
if not self.circuit_breaker.can_execute():
logger.warning(
f"Cannot start sidecar: Circuit breaker is {self.circuit_breaker.state}"
)
return False
try:
self.process = subprocess.Popen(
["python", "main.py"],
cwd="python/",
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
await self._wait_for_health()
self.circuit_breaker.record_success()
return
Exception e:
logger.error()
.circuit_breaker.record_failure()
():
httpx
start_time = time.time()
time.time() - start_time < timeout:
:
response = httpx.get()
response.status_code == :
Exception:
asyncio.sleep()
TimeoutError()
():
.process:
.process.terminate()
.process.wait()
.process =
() -> :
{
: .process ,
: .circuit_breaker.get_status()
}
Usage Example
sidecar_manager = SidecarManager()
if await sidecar_manager.start():
print("Sidecar started successfully")
else:
print("Sidecar cannot start (circuit breaker open)")
print(f"Status: {sidecar_manager.get_status()}")
status = sidecar_manager.get_status()
print(f"Circuit breaker state: {status['circuit_breaker']['state']}")
print(f"Can execute: {status['circuit_breaker']['can_execute']}")
Error Handling
Custom Exception Handlers
from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
class CustomException(Exception):
def __init__(self, name: str):
self.name = name
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name} did something wrong."},
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={"detail": exc.errors()},
)
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id == "error":
raise CustomException(name="Item")
return {"item_id": item_id}
Testing
Basic Tests with TestClient
from fastapi.testclient import TestClient
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item", "price": 10.5}
)
assert response.status_code == 200
assert response.json()["name"] == "Test Item"
def test_authentication():
response = client.post(
"/token",
data={"username": "testuser", "password": "testpass"}
)
assert response.status_code == 200
assert "access_token" in response.json()
Async Tests
import pytest
from httpx import AsyncClient
@pytest.mark.anyio
async def test_read_items():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/items/")
assert response.status_code == 200
assert isinstance(response.json(), list)
@pytest.mark.anyio
async def test_create_user():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.post(
"/users/",
json={
"username": "newuser",
"email": "new@example.com",
"password": "securepass123"
}
)
assert response.status_code == 200
assert response.json()["username"] == "newuser"
Routers and Organization
APIRouter for Modular Code
from fastapi import APIRouter, Depends
router = APIRouter(
prefix="/users",
tags=["users"],
dependencies=[Depends(verify_token)],
responses={404: {"description": "Not found"}},
)
@router.get("/")
async def list_users():
return [{"username": "user1"}, {"username": "user2"}]
@router.get("/{user_id}")
async def get_user(user_id: int):
return {"user_id": user_id}
from routers import users
app = FastAPI()
app.include_router(users.router)
Best Practices
1. Project Structure
my_fastapi_project/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── users.py
│ │ └── items.py
│ ├── dependencies/
│ │ ├── __init__.py
│ │ ├── auth.py
│ │ └── database.py
│ └── utils/
│ ├── __init__.py
│ └── security.py
├── tests/
│ ├── __init__.py
│ ├── test_users.py
│ └── test_items.py
├── requirements.txt
├── .env
└── README.md
2. Configuration Management
from pydantic import BaseSettings
class Settings(BaseSettings):
app_name: str = "My FastAPI App"
database_url: str
secret_key: str
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
class Config:
env_file = ".env"
settings = Settings()
3. Documentation
app = FastAPI(
title="My API",
description="This is a very custom API",
version="1.0.0",
openapi_tags=[
{
"name": "users",
"description": "Operations with users.",
},
{
"name": "items",
"description": "Manage items.",
},
]
)
@app.post(
"/items/",
response_model=Item,
tags=["items"],
summary="Create an item",
description="Create an item with all the information",
response_description="The created item",
)
async def create_item(item: Item):
return item
Production Deployment
Docker Setup
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY ./app /app
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
Run with Gunicorn and Uvicorn Workers
gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
Skill Version: 1.0.0
Last Updated: October 2025
Skill Category: Backend Development, API Development, Python
Compatible With: FastAPI 0.100+, Python 3.7+, Pydantic 2.0+