Starlette is the ASGI framework that powers FastAPI. pip install starlette uvicorn. App: from starlette.applications import Starlette. Routes: routes=[Route("/", homepage), Mount("/api", app=api_app)]. Request: async def homepage(request): return JSONResponse({"hello":"world"}). Run: uvicorn app:app. Response types: JSONResponse(data, status_code=200, headers={}), HTMLResponse(html), PlainTextResponse(text), RedirectResponse(url, status_code=302), StreamingResponse(generator(), media_type="text/event-stream"). Request body: await request.json(), await request.body(), await request.form(). Path params: request.path_params["id"]. Query: request.query_params["page"]. Headers: request.headers["Authorization"]. Background task: BackgroundTask(fn, *args) — runs after response sent. response = JSONResponse({...}, background=BackgroundTask(send_email, to=...)). Static files: Mount("/static", app=StaticFiles(directory="static"), name="static"). Templates: from starlette.templating import Jinja2Templates; templates = Jinja2Templates(directory="templates"). return templates.TemplateResponse("page.html", {"request": request, "data": data}). Middleware: @app.middleware("http") async def add_header(request, call_next): response = await call_next(request); response.headers["X-Time"] = "..."; return response. app.add_middleware(CORSMiddleware, allow_origins=["*"]). app.add_middleware(SessionMiddleware, secret_key="key"). WebSocket: async def ws_handler(websocket): await websocket.accept(); data = await websocket.receive_text(); await websocket.send_text(data). Route: WebSocketRoute("/ws", ws_handler). Lifespan: @asynccontextmanager async def lifespan(app): ... yield .... Starlette(lifespan=lifespan). Test: from starlette.testclient import TestClient; client = TestClient(app); response = client.get("/"). Exception handlers: exception_handlers={404: not_found_handler, Exception: server_error_handler}. Claude Code generates Starlette routes, middleware stacks, WebSocket handlers, and streaming response endpoints.
CLAUDE.md for Starlette
## Starlette Stack
- Version: starlette >= 0.37 | pip install "starlette[full]" uvicorn
- App: Starlette(routes=[Route(path, endpoint), Mount(prefix, app=sub)])
- Response: JSONResponse | HTMLResponse | StreamingResponse | RedirectResponse
- Middleware: add_middleware(CORSMiddleware,...) | @app.middleware("http")
- Background: BackgroundTask(fn, *args) attached to any response
- WebSocket: WebSocketRoute("/ws", handler) + websocket.accept/send/receive
- Test: TestClient(app) — synchronous, no uvicorn needed
- Lifespan: @asynccontextmanager async def lifespan(app): init; yield; cleanupStarlette ASGI Web Pipeline
# app/main.py — Starlette application with routing, middleware, and WebSockets
from __future__ import annotations
import asyncio
import json
import time
from contextlib import asynccontextmanager
from typing import AsyncIterator
from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.requests import Request
from starlette.responses import (
HTMLResponse,
JSONResponse,
PlainTextResponse,
RedirectResponse,
StreamingResponse,
)
from starlette.routing import Mount, Route, WebSocketRoute
from starlette.staticfiles import StaticFiles
from starlette.testclient import TestClient
from starlette.websockets import WebSocket
# ─────────────────────────────────────────────────────────────────────────────
# State — shared across requests via app.state
# ─────────────────────────────────────────────────────────────────────────────
# Populated in lifespan, accessed via request.app.state
class AppState:
db_pool: object = None
cache: dict = {}
# ─────────────────────────────────────────────────────────────────────────────
# Lifespan — startup and shutdown
# ─────────────────────────────────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
"""Initialise resources on startup, tear them down on shutdown."""
app.state.cache = {}
app.state.start_time = time.time()
print("App started")
yield
print("App shutting down")
# ─────────────────────────────────────────────────────────────────────────────
# Endpoints — plain async functions
# ─────────────────────────────────────────────────────────────────────────────
async def homepage(request: Request) -> HTMLResponse:
return HTMLResponse("""
<html><body>
<h1>Starlette Demo</h1>
<a href="/api/users">Users API</a>
</body></html>
""")
async def health(request: Request) -> JSONResponse:
uptime = time.time() - getattr(request.app.state, "start_time", time.time())
return JSONResponse({
"status": "ok",
"uptime_seconds": round(uptime, 2),
"cache_entries": len(getattr(request.app.state, "cache", {})),
})
# ─────────────────────────────────────────────────────────────────────────────
# User endpoints
# ─────────────────────────────────────────────────────────────────────────────
FAKE_USERS = {
1: {"id": 1, "name": "Alice", "email": "[email protected]"},
2: {"id": 2, "name": "Bob", "email": "[email protected]"},
}
async def list_users(request: Request) -> JSONResponse:
page = int(request.query_params.get("page", "1"))
page_size = int(request.query_params.get("page_size", "20"))
users = list(FAKE_USERS.values())
return JSONResponse({
"items": users,
"total": len(users),
"page": page,
})
async def get_user(request: Request) -> JSONResponse:
user_id = int(request.path_params["user_id"])
user = FAKE_USERS.get(user_id)
if user is None:
return JSONResponse({"error": "User not found"}, status_code=404)
return JSONResponse(user)
async def create_user(request: Request) -> JSONResponse:
try:
data = await request.json()
except Exception:
return JSONResponse({"error": "Invalid JSON"}, status_code=400)
if not data.get("name") or not data.get("email"):
return JSONResponse({"error": "name and email are required"}, status_code=422)
new_id = max(FAKE_USERS) + 1
user = {"id": new_id, **data}
FAKE_USERS[new_id] = user
# Send welcome email in the background after response is sent
bg = BackgroundTask(_send_welcome_email, user["email"], user["name"])
return JSONResponse(user, status_code=201, background=bg)
async def _send_welcome_email(email: str, name: str) -> None:
"""Runs after the HTTP response is sent — doesn't block the client."""
await asyncio.sleep(0.1) # simulate email API call
print(f"Welcome email sent to {name} <{email}>")
async def redirect_to_users(request: Request) -> RedirectResponse:
return RedirectResponse(url="/api/users", status_code=302)
# ─────────────────────────────────────────────────────────────────────────────
# Streaming responses — SSE and large data
# ─────────────────────────────────────────────────────────────────────────────
async def stream_events(request: Request) -> StreamingResponse:
"""Server-Sent Events — streams newline-delimited JSON to the browser."""
async def event_generator() -> AsyncIterator[str]:
for i in range(5):
event = json.dumps({"seq": i, "ts": time.time()})
yield f"data: {event}\n\n"
await asyncio.sleep(0.5)
yield "data: {\"done\": true}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)
async def stream_large_file(request: Request) -> StreamingResponse:
"""Stream a large response without buffering in memory."""
async def generate() -> AsyncIterator[bytes]:
for chunk_num in range(10):
yield f"chunk-{chunk_num}\n".encode()
await asyncio.sleep(0) # yield event loop
return StreamingResponse(generate(), media_type="text/plain")
# ─────────────────────────────────────────────────────────────────────────────
# WebSocket
# ─────────────────────────────────────────────────────────────────────────────
async def websocket_chat(websocket: WebSocket) -> None:
"""Echo server with connection lifecycle handling."""
await websocket.accept()
try:
while True:
msg = await websocket.receive_text()
echo = f"echo: {msg}"
await websocket.send_text(echo)
except Exception:
pass
finally:
try:
await websocket.close()
except Exception:
pass
async def websocket_json(websocket: WebSocket) -> None:
"""JSON WebSocket — parse and send structured messages."""
await websocket.accept()
try:
while True:
data = await websocket.receive_json()
await websocket.send_json({"received": data, "ts": time.time()})
except Exception:
pass
# ─────────────────────────────────────────────────────────────────────────────
# Exception handlers
# ─────────────────────────────────────────────────────────────────────────────
async def not_found(request: Request, exc: Exception) -> JSONResponse:
return JSONResponse({"error": "Not found", "path": str(request.url.path)},
status_code=404)
async def server_error(request: Request, exc: Exception) -> JSONResponse:
return JSONResponse({"error": "Internal server error"}, status_code=500)
# ─────────────────────────────────────────────────────────────────────────────
# Custom middleware — add request timing header
# ─────────────────────────────────────────────────────────────────────────────
class TimingMiddleware:
def __init__(self, app) -> None:
self.app = app
async def __call__(self, scope, receive, send) -> None:
if scope["type"] == "http":
start = time.perf_counter()
async def send_wrapper(message):
if message["type"] == "http.response.start":
headers = dict(message.get("headers", []))
elapsed = time.perf_counter() - start
headers[b"x-response-time"] = f"{elapsed * 1000:.1f}ms".encode()
message = {**message, "headers": list(headers.items())}
await send(message)
await self.app(scope, receive, send_wrapper)
else:
await self.app(scope, receive, send)
# ─────────────────────────────────────────────────────────────────────────────
# Application assembly
# ─────────────────────────────────────────────────────────────────────────────
api_routes = [
Route("/users", list_users, methods=["GET"]),
Route("/users/{user_id:int}", get_user, methods=["GET"]),
Route("/users", create_user, methods=["POST"]),
Route("/stream", stream_events, methods=["GET"]),
Route("/stream/file", stream_large_file, methods=["GET"]),
]
routes = [
Route("/", homepage),
Route("/health", health),
Route("/redirect", redirect_to_users),
Mount("/api", routes=api_routes),
WebSocketRoute("/ws/chat", websocket_chat),
WebSocketRoute("/ws/json", websocket_json),
]
middleware = [
Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
Middleware(SessionMiddleware, secret_key="change-me-in-production"),
Middleware(TimingMiddleware),
]
app = Starlette(
debug=False,
routes=routes,
middleware=middleware,
exception_handlers={404: not_found, 500: server_error},
lifespan=lifespan,
)
# ─────────────────────────────────────────────────────────────────────────────
# Tests using TestClient (no server needed)
# ─────────────────────────────────────────────────────────────────────────────
def test_health():
client = TestClient(app)
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "ok"
def test_list_users():
client = TestClient(app)
response = client.get("/api/users")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert data["total"] >= 2
def test_create_user():
client = TestClient(app)
response = client.post("/api/users", json={"name": "Carol", "email": "[email protected]"})
assert response.status_code == 201
assert response.json()["name"] == "Carol"
def test_websocket():
client = TestClient(app)
with client.websocket_connect("/ws/chat") as ws:
ws.send_text("hello")
reply = ws.receive_text()
assert reply == "echo: hello"
if __name__ == "__main__":
import uvicorn
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)For the Flask alternative — Flask uses synchronous WSGI which spawns a new thread per request, making each requests.get() inside a handler block a thread, while Starlette’s async endpoints share one event loop: await httpx.get(url) yields during the network call so 100 concurrent requests need far fewer threads — and BackgroundTask runs post-response work on the same event loop without spawning a thread at all. For the FastAPI alternative — FastAPI is built on top of Starlette and adds dependency injection, automatic OpenAPI docs, and Pydantic request validation while Starlette gives direct access to Request.body() and raw ASGI scope for use cases like proxy servers, SSE endpoints needing StreamingResponse with custom chunking, or WebSocket servers where you control the full connection lifecycle without FastAPI’s dependency system overhead. The Claude Skills 360 bundle includes Starlette skill sets covering Route and Mount routing, JSONResponse and StreamingResponse, BackgroundTask post-response work, WebSocket accept/send/receive, custom ASGI middleware, CORSMiddleware and SessionMiddleware, Jinja2 template responses, lifespan context for startup/shutdown, exception handlers, and TestClient for fast synchronous testing. Start with the free tier to try ASGI web framework code generation.