FastAPI Development (Hybrid)

FastAPI Development Guide (Hybrid Knowledge System)

This skill provides curated FastAPI guidance with intelligent escalation to Context7 MCP for advanced topics.

Knowledge Tier System

This skill operates on a tiered knowledge model:

Tier 1: This Skill (Use First)

Covers 90% of common FastAPI tasks with curated, tested patterns:

• Application setup and structure
• Pydantic models and validation
• CRUD operations and routers
• Dependency injection
• OAuth2/JWT authentication
• Error handling
• Middleware
• Background tasks
• Testing basics

Tier 2: Context7 MCP (Escalate When Needed)

IMPORTANT: Escalate to Context7 MCP server when the user's query matches ANY of these conditions:

Automatic Escalation Triggers

1. Version-Specific Questions

   • "FastAPI 0.100+", "latest version", "new in FastAPI"
   • "Pydantic v2", "migration from v1"
   • Use: mcp__context7__query-docs with /websites/fastapi_tiangolo

2. Advanced Topics Not Covered Here

   • WebSockets beyond basics
   • GraphQL integration
   • OpenTelemetry/tracing
   • Custom OpenAPI schema modifications
   • Strawberry GraphQL
   • Server-Sent Events (SSE)

3. Specific Error Messages

   • When user pastes an error message
   • Query Context7 with the exact error text

4. Third-Party Integrations

   • Specific database drivers (asyncpg, aiomysql)
   • Message queues (Celery, RQ, ARQ)
   • Caching (Redis, memcached)
   • Search (Elasticsearch, Meilisearch)

5. User Explicitly Requests

   • "show me the official docs"
   • "what does the documentation say"
   • "latest best practices"
   • "more details about X"

Escalation Protocol

When escalating to Context7, use this pattern:

1. First, acknowledge what you know from this skill
2. Then fetch additional details:
   - Resolve library: mcp__context7__resolve-library-id("fastapi", "user's specific question")
   - Query docs: mcp__context7__query-docs("/websites/fastapi_tiangolo", "specific topic")
3. Synthesize skill knowledge + Context7 results
4. Provide unified response with both curated pattern AND latest docs

Tier 1 Knowledge: Core Patterns

Application Structure (Curated)

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application entry point
│   ├── config.py            # Settings (pydantic-settings)
│   ├── dependencies.py      # Shared dependencies
│   ├── models/              # Pydantic schemas
│   ├── routers/             # API route handlers
│   ├── services/            # Business logic
│   └── db/                  # Database (SQLAlchemy)
├── tests/
└── pyproject.toml

Basic Application Setup

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up...")
    yield
    # Shutdown
    print("Shutting down...")

app = FastAPI(
    title="My API",
    description="API description",
    version="1.0.0",
    lifespan=lifespan,
)

Pydantic Models (v2 Syntax)

from pydantic import BaseModel, Field, EmailStr, ConfigDict
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50)

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime
    is_active: bool = True

    model_config = ConfigDict(from_attributes=True)

ESCALATION NOTE: For Pydantic v1 to v2 migration details, query Context7: mcp__context7__query-docs("/pydantic/pydantic", "migration from v1 to v2")

Path Operations

from fastapi import Path, Query, Body, HTTPException, status
from typing import Annotated

@app.get("/users/{user_id}")
async def get_user(
    user_id: Annotated[int, Path(gt=0, description="User ID")],
    include_posts: Annotated[bool, Query(description="Include posts")] = False,
):
    user = await get_user_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    return await create_new_user(user)

APIRouter Organization

from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

@router.get("/")
async def list_items():
    return []

# In main.py
app.include_router(router, prefix="/api/v1")

Dependency Injection

from fastapi import Depends
from typing import Annotated

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        await db.close()

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[Session, Depends(get_db)],
) -> User:
    # Validate token and return user
    ...

@app.get("/me")
async def get_me(user: Annotated[User, Depends(get_current_user)]):
    return user

OAuth2 + JWT Authentication

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        username = payload.get("sub")
        if not username:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await get_user_by_username(username)
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user

ESCALATION NOTE: For OAuth2 scopes, API key auth, or third-party OAuth providers, query Context7 for comprehensive examples.

Error Handling

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

# Custom exception
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Item {exc.item_id} not found"},
    )

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors()},
    )

Middleware

from fastapi.middleware.cors import CORSMiddleware
import time

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start)
    return response

Background Tasks

from fastapi import BackgroundTasks

async def send_email(email: str, message: str):
    # Email sending logic
    pass

@app.post("/notify/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Notification queued"}

Tier 2 Topics: Escalate to Context7

The following topics require Context7 MCP lookup for comprehensive guidance:

WebSockets (Advanced)

Basic pattern provided, escalate for:

• Connection managers
• Room/channel patterns
• Authentication in WebSockets
• Scaling with Redis pub/sub

# Basic WebSocket (Tier 1)
from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

ESCALATE: For connection managers, rooms, or scaling patterns → mcp__context7__query-docs("/websites/fastapi_tiangolo", "websocket connection manager broadcast")

Database Patterns

Escalate for:

• Specific async drivers (asyncpg, aiomysql, motor)
• Connection pooling configuration
• Multi-database setups
• Database migrations with Alembic

ESCALATE: mcp__context7__query-docs("/websites/fastapi_tiangolo", "sqlalchemy async session")

Testing (Advanced)

Escalate for:

• Async test fixtures
• Database test isolation
• Mocking external services
• Integration testing patterns

ESCALATE: mcp__context7__query-docs("/websites/fastapi_tiangolo", "testing async database")

Deployment

Escalate for:

• Docker configuration
• Kubernetes deployment
• Gunicorn/Uvicorn workers
• Health checks for orchestrators

ESCALATE: mcp__context7__query-docs("/websites/fastapi_tiangolo", "deployment docker gunicorn")

Quick Reference

Additional Resources

Reference Files (Tier 1 Deep-Dives)

• references/authentication.md - Complete auth patterns
• references/database.md - SQLAlchemy async patterns
• references/testing.md - Testing strategies

Context7 Libraries for Escalation

• FastAPI: /websites/fastapi_tiangolo
• Pydantic: /pydantic/pydantic
• SQLAlchemy: /sqlalchemy/sqlalchemy
• Starlette: /encode/starlette