Cross-cutting behavior
Request Pipeline
Tenant context, request timing, app-wide response headers, or an app-owned gate belong in
src/middleware.py — an ordered chain the application owns — instead
of edits to main.py or copies across routes.
Framework pipeline — main.py
Transport and security invariants: request id, security headers, public files, maintenance mode, rate limiting, sessions, CSRF, and auth.
Application pipeline — src/middleware.py
Your ordered chain around pages, route.py endpoints, and RPC calls. Never reorder framework middleware for an app concern.
Writing middleware
The file is optional. When present it must define middleware, a list of async functions or Middleware(...) entries.
# src/middleware.py
from casp.request_pipeline import Middleware
async def tenant(request, call_next):
request.state.tenant = request.headers.get("x-tenant", "default")
return await call_next(request)
async def timing(request, call_next):
import time
started = time.perf_counter()
response = await call_next(request)
response.headers["Server-Timing"] = f"app;dur={(time.perf_counter() - started) * 1000:.1f}"
return response
middleware = [
tenant,
Middleware(timing, paths=["/dashboard", "/api"], exclude=["/api/webhooks"]),
]
-
Each entry is
async def name(request, call_next); a sync function is a startup error. -
Work before
call_nextruns in list order; work after it unwinds in reverse. -
request.statevalues reach pages,route.pyhandlers, and RPC actions. -
pathslimits an entry to URL subtrees (/dashboardcovers/dashboard/users, not/dashboards);excluderemoves subtrees. - Streaming responses pass through — do not read the response body in middleware.
- Loaded once at startup; the dev stack restarts Python when it changes.
Short circuits
Returning a response without calling call_next skips the page, endpoint,
or RPC, skips layouts, and bypasses the page cache; the request id and security headers still apply.
Return the shape the caller expects — JSON for APIs and RPC, HTML or a redirect for pages.
from starlette.responses import JSONResponse
async def require_tenant(request, call_next):
if request.url.path.startswith("/api/") and "x-tenant" not in request.headers:
return JSONResponse({"error": "Missing tenant"}, status_code=400)
return await call_next(request)
Do not recreate the framework
Never use a short circuit to rebuild authentication, CSRF, or rate limiting, and use maintenance mode rather than a hand-built 503 gate. Keep business validation and resource authorization in the handler, where the typed input and loaded record exist.
Scope and order
request id
-> security headers
-> public files
-> request records and metrics
-> maintenance mode
-> rate limit
-> missing public asset / body limit
-> session -> CSRF cookie -> auth
-> application pipeline (src/middleware.py)
-> RPC -> page / route.py
The application pipeline never runs for:
-
existing files under
public/; - WebSocket connections;
-
/health,/ready, and/mcp; - requests already refused (rate limit, maintenance, private-route redirect).
Because it runs after auth, auth.get_payload() and
allows(...) work inside it. Middleware that gates an action must use the same
ability the handler authorizes with. An exception raised here becomes an error response through the global handlers.
Authorization
Per-action and per-record policies.
Maintenance Mode
The built-in 503 gate with a bypass secret.