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