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

•
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
•
Advanced Topics Not Covered Here
- •WebSockets beyond basics
- •GraphQL integration
- •OpenTelemetry/tracing
- •Custom OpenAPI schema modifications
- •Strawberry GraphQL
- •Server-Sent Events (SSE)
•
Specific Error Messages
- •When user pastes an error message
- •Query Context7 with the exact error text
•
Third-Party Integrations
- •Specific database drivers (asyncpg, aiomysql)
- •Message queues (Celery, RQ, ARQ)
- •Caching (Redis, memcached)
- •Search (Elasticsearch, Meilisearch)
•
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:

code

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)

code

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

python

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)

python

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

python

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

python

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

python

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

python

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

python

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

python

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

python

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

python

# 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

Task	Tier	Action
Create endpoint	1	Use patterns above
Add Pydantic model	1	Use patterns above
JWT auth	1	Use patterns above
WebSocket rooms	2	Escalate to Context7
Pydantic v2 migration	2	Escalate to Context7
Specific error message	2	Escalate to Context7
Third-party integration	2	Escalate to Context7
Latest FastAPI features	2	Escalate to Context7

FastAPI Development (Hybrid)

FastAPI Development Guide (Hybrid Knowledge System)

Knowledge Tier System

Tier 1: This Skill (Use First)

Tier 2: Context7 MCP (Escalate When Needed)

Automatic Escalation Triggers

Escalation Protocol

Tier 1 Knowledge: Core Patterns

Application Structure (Curated)

Basic Application Setup

Pydantic Models (v2 Syntax)

Path Operations

APIRouter Organization

Dependency Injection

OAuth2 + JWT Authentication

Error Handling

Middleware

Background Tasks

Tier 2 Topics: Escalate to Context7

WebSockets (Advanced)

Database Patterns

Testing (Advanced)

Deployment

Quick Reference

Additional Resources

Reference Files (Tier 1 Deep-Dives)

Context7 Libraries for Escalation