Custom Middleware

Overview

Middleware wraps every request and response. It runs before your handler sees the request and after your handler returns, so it's where cross-cutting work lives: logging, auth, rate limiting, timing, security headers.

A middleware is any async callable matchingasync def mw(request, next) -> Response. Call await next(request)to pass control down [the request pipeline], then inspect or replace what comes back.

Reach for custom middleware when the work applies to many routes. For one route, do it in the handler. Chirp already ships the common middleware — sessions, CSRF, security headers — so check Built-in Middleware before writing your own.

Define a middleware

The shortest middleware is a function. Use a class when you need configuration or state.

import time
from chirp import Request, Response, Next

async def timing(request: Request, next: Next) -> Response:
    start = time.monotonic()
    response = await next(request)
    elapsed = time.monotonic() - start
    return response.with_header("X-Response-Time", f"{elapsed:.3f}s")

app.add_middleware(timing)

import threading
import time
from chirp import Request, Response, Next

class RateLimiter:
    def __init__(self, max_requests: int, window: float) -> None:
        self.max_requests = max_requests
        self.window = window
        self._counts: dict[str, list[float]] = {}
        self._lock = threading.Lock()

    async def __call__(self, request: Request, next: Next) -> Response:
        client_ip = request.headers.get("X-Forwarded-For", "unknown")

        with self._lock:
            now = time.monotonic()
            hits = self._counts.setdefault(client_ip, [])
            # Remove expired entries
            hits[:] = [t for t in hits if now - t < self.window]

            if len(hits) >= self.max_requests:
                return Response("Too Many Requests").with_status(429)
            hits.append(now)

        return await next(request)

app.add_middleware(RateLimiter(max_requests=100, window=60.0))

Common patterns

Request logging

async def request_logger(request: Request, next: Next) -> Response:
    print(f"→ {request.method} {request.path}")
    response = await next(request)
    print(f"← {response.status} {request.path}")
    return response

Error handling

async def error_boundary(request: Request, next: Next) -> Response:
    try:
        return await next(request)
    except Exception as e:
        print(f"Error: {e}")
        return Response("Internal Server Error").with_status(500)

Request context

Useg, the request-scoped namespace, to pass data between middleware and handlers:

from chirp import g

async def load_user(request: Request, next: Next) -> Response:
    token = request.cookies.get("session_token")
    if token:
        g.user = await get_user_from_token(token)
    else:
        g.user = None
    return await next(request)

Then in handlers:

from chirp import Redirect, Template, g

@app.route("/profile")
def profile():
    if not g.user:
        return Redirect("/login")
    return Template("profile.html", user=g.user)

g is backed by a ContextVar, so each request gets its own namespace.

Conditional middleware

Skip middleware for certain paths:

async def auth_required(request: Request, next: Next) -> Response:
    public_paths = {"/", "/login", "/health"}
    if request.path in public_paths:
        return await next(request)

    if not request.cookies.get("session"):
        return Redirect("/login")

    return await next(request)

Response transformation

For standard security headers, add the built-in [[docs/build-apps/request-pipeline/builtin|SecurityHeadersMiddleware]] rather than rolling your own:

from chirp.middleware import SecurityHeadersMiddleware

app.add_middleware(SecurityHeadersMiddleware())

async def add_security_headers(request: Request, next: Next) -> Response:
    response = await next(request)
    return (
        response
        .with_header("X-Content-Type-Options", "nosniff")
        .with_header("X-Frame-Options", "DENY")
        .with_header("Referrer-Policy", "strict-origin-when-cross-origin")
    )

Next steps

• [Request Pipeline] — how the chain composes and orders
• Built-in Middleware — what ships with Chirp
• Thread Safety — free-threading patterns for shared state