API Reference

This page documents Responder’s public Python API. For usage examples and explanations, see the Quick Start and Feature Tour.

The API Class

The central object of every Responder application. It holds your routes, middleware, templates, and configuration. Create one at the top of your module and use it to define your entire web service.

Quick example:

import responder

api = responder.API(
    title="My Service",           # OpenAPI title
    version="1.0",                # OpenAPI version
    openapi="3.0.2",              # enable OpenAPI
    docs_route="/docs",           # Swagger UI at /docs
    cors=True,                    # enable CORS
    allowed_hosts=["example.com"],
)

Note

Cookie sessions are on by default (sessions="auto"). With no key set, Responder mints a random per-process signing key at startup and logs a warning — fine for a single dev process, but multi-worker or multi-instance deploys need a stable key so signed cookies validate everywhere. Set one with secret_key= or the RESPONDER_SECRET_KEY environment variable:

python -c "import secrets; print(secrets.token_urlsafe(32))"

secret_key="NOTASECRET" (the old public default) now raises SessionConfigError, and sessions=True with no key raises as well. Pass sessions=False for a stateless service. See Configuration for the full secret-key and session-cookie story.

class responder.API(*, debug=False, title=None, version=None, description=None, terms_of_service=None, contact=None, license=None, openapi=None, openapi_servers=None, openapi_route='/schema.yml', static_dir=<object object>, static_route='/static', implicit_static_fallback=False, templates_dir='templates', auto_escape=True, secret_key=None, enable_hsts=False, security_headers=False, docs_route=None, cors=False, cors_params={'allow_credentials': False, 'allow_headers': (), 'allow_methods': ('GET', ), 'allow_origin_regex': None, 'allow_origins': (), 'expose_headers': (), 'max_age': 600}, allowed_hosts=None, openapi_theme='swagger_ui', lifespan=None, gzip=True, request_id=False, enable_logging=False, trust_proxy_headers=False, csrf=False, redirect_slashes=True, max_request_size=104857600, auto_etag=False, auto_vary=True, request_timeout=None, ws_idle_timeout=None, trace_dispatch=False, sessions='auto', session_backend=None, session_cookie=None, session_https_only=None, session_same_site='lax', session_max_age=1209600, metrics_route=None, metrics_buckets=None, health_route=None, encoder=None, json_ensure_ascii=False, json_decimal='string', problem_details=True, problem_handler=None, auth=None)

The primary web-service class.

Parameters:

• static_dir – The directory to use for static files (default static). Mounted at static_route only if it exists — it is never created for you. Passing a static_dir explicitly that doesn’t exist raises FileNotFoundError.

• templates_dir – The directory to use for templates (default templates). It is not created for you.

• auto_escape – If True, HTML and XML templates will automatically be escaped.

• enable_hsts – If True, redirect HTTP requests to HTTPS and send a Strict-Transport-Security header.

• security_headers – If True, add common security headers (nosniff, X-Frame-Options, Referrer-Policy) to every response; pass a dict of SecurityHeadersMiddleware options to customize (e.g. content_security_policy).

• gzip – If True (the default), compress responses with GZip.

• openapi_theme – OpenAPI documentation theme, must be one of elements, rapidoc, redoc, swagger_ui

add_dependency(name, provider, *, scope='request')

Register a dependency provider under an explicit name.

Parameters:

• name – The view parameter name to inject as.

• provider – The provider function (sync/async function or generator).

• scope – "request" (default) or "app".

add_event_handler(event_type, handler)

Adds an event handler to the API.

Parameters:

• event_type – A string in (“startup”, “shutdown”)

• handler – The function to run. Can be either a function or a coroutine.

add_exception_handler(exc_class_or_status_code, handler)

Register a handler for an exception type or status code.

handler is a Responder-style (req, resp, exc) callable (sync or async). A handler for 500/Exception installs the catch-all server-error handler (ignored under debug=True, which shows the traceback); any other exception/status routes through the exception middleware.

add_health_check(name, check)

Register a readiness check run by the health endpoint.

check is a sync or async callable; it passes unless it returns False or raises. The endpoint returns 200 when every check passes and 503 otherwise, with per-check JSON. The route (default /health, or the health_route= you set) is added on first use.

Parameters:

• name – A label for the check, used as its key in the JSON body.

• check – The check callable.

add_middleware(middleware_cls, **middleware_config)

Add ASGI middleware to the application (valid after construction).

User middleware sits just inside ServerErrorMiddleware (so its errors are caught and rendered) and the most-recently-added runs first. To wrap everything (including error rendering), wrap the API object: asgi = MyMiddleware(api).

Parameters:

• middleware_cls – A Starlette-compatible middleware class.

• middleware_config – Keyword arguments passed to the constructor.

add_route(route=None, endpoint=None, *, default=False, static=True, check_existing=True, websocket=False, before_request=False, methods=None, name=None)

Adds a route to the API.

Parameters:

• route – A string representation of the route.

• endpoint – The endpoint for the route – can be a callable, or a class.

• default – If True, all unknown requests will route to this view.

• static – If True and no endpoint was passed, render static/index.html as a legacy default route only when implicit_static_fallback=True.

• methods – Optional list of HTTP methods (e.g. ["GET", "POST"]).

• name – Optional route name for url_for() reverse lookup.

Calling add_route() without an endpoint now raises by default. Pass an endpoint explicitly, or serve static assets via static_dir/static_route. Apps that need the old fallback during migration can pass API(implicit_static_fallback=True).

add_security_scheme(name, scheme=None, *, default=False)

Register an OpenAPI security scheme (enables Swagger’s Authorize button).

Accepts either an auth helper carrying its own definition (add_security_scheme(BearerAuth(...))) or an explicit name plus a scheme dict. With default=True the scheme becomes a global requirement applied to every documented operation.

Parameters:

• name – The scheme name, or an auth helper object.

• scheme – The OpenAPI security-scheme dict (omit when passing a helper).

• default – If True, require this scheme on all operations.

after_request(f=None)

Register a function to run after every request.

Works both bare and called: @api.after_request or @api.after_request().

Usage:

@api.after_request
def add_request_id(req, resp):
    resp.headers["X-Request-ID"] = str(uuid.uuid4())

property app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

The assembled ASGI middleware stack, built lazily on first access.

property async_requests

An async test client wired to the ASGI app (httpx.AsyncClient).

The async mirror of requests; async with also runs the app’s lifespan. See responder.testing.AsyncTestClient.

before_request(websocket=False)

Register a function to run before every request.

If the hook sets resp.status_code, the route handler is skipped and the response is sent immediately (short-circuiting).

Parameters:

websocket – If True, register as a WebSocket before-request hook instead of HTTP.

Usage:

@api.before_request()
def check_auth(req, resp):
    if "Authorization" not in req.headers:
        resp.status_code = 401
        resp.media = {"error": "unauthorized"}

build_middleware_stack() → Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

Assemble the full ASGI stack from the collected configuration.

Outermost → innermost: logging/request-id → metrics → ServerError → user middleware → trusted-host → hsts → cors → sessions → gzip → ExceptionMiddleware → router. ServerErrorMiddleware is the outermost application layer (it catches errors from every middleware below it), while the observability tier wraps even it so a rendered 500 still carries X-Request-ID and is logged with its real status.

delete(route=None, **options)

Register a route for DELETE (sugar for route(methods=["DELETE"])).

dependency(name=None, *, scope='request')

Register a dependency provider, injected into views by parameter name.

Any view parameter (beyond req and resp) whose name matches a registered dependency receives the provider’s value. Providers may be sync or async functions, or generators — code after yield runs as teardown once the response is sent. Providers accepting a parameter receive the current Request. Each dependency is resolved at most once per request. Path parameters take precedence over dependencies of the same name.

Parameters:

• name – The injection name. Defaults to the provider’s __name__.

• scope – "request" (default) resolves per request; "app" resolves once on first use and caches for the application’s lifetime — generator teardown then runs at shutdown. App-scoped providers cannot take parameters.

Usage:

@api.dependency()
async def db():
    conn = await create_connection()
    yield conn
    await conn.close()

@api.route("/users/{id:int}")
async def get_user(req, resp, *, id, db):
    resp.media = await db.fetch_user(id)

An app-scoped dependency, shared across all requests:

@api.dependency(scope="app")
async def pool():
    pool = await create_pool()
    yield pool
    await pool.close()  # runs at application shutdown

dependency_overrides(**overrides)

Temporarily override dependencies (for tests); restores on exit.

Each value may be a provider (a callable, with full sub-dependency and request injection) or a bare value, which is wrapped automatically. Overrides are request-scoped, so they replace and bypass the cache of an app-scoped dependency too:

with api.dependency_overrides(db=fake_db):
    api.requests.get("/users")

Parameters:

overrides – name=provider_or_value pairs to override.

exception_handler(exception_cls)

Register a handler for a specific exception type.

Usage:

@api.exception_handler(ValueError)
async def handle_value_error(req, resp, exc):
    resp.status_code = 400
    resp.media = {"error": str(exc)}

generate_client(path=None, *, class_name='APIClient', language='python')

Generate a client from this app’s OpenAPI schema.

Returns source code by default. When path is provided, writes the client module there and returns the resulting Path.

Usage:

source = api.generate_client(class_name="AcmeClient")
api.generate_client("clients/acme.py", class_name="AcmeClient")
api.generate_client(
    "clients/acme.ts",
    class_name="AcmeClient",
    language="typescript",
)

get(route=None, **options)

Register a route for GET (sugar for route(methods=["GET"])).

graphql(route='/graphql', *, schema, graphiql=True, introspection=True, max_depth=None, partial_data_status=200)

Mount a GraphQL API at the given route.

Usage:

import graphene

class Query(graphene.ObjectType):
    hello = graphene.String(name=graphene.String(default_value="stranger"))
    def resolve_hello(self, info, name):
        return f"Hello {name}"

api.graphql("/graphql", schema=graphene.Schema(query=Query))

For production, disable the in-browser IDE and introspection and cap query depth:

api.graphql(
    "/graphql", schema=schema,
    graphiql=api.debug, introspection=api.debug, max_depth=10,
)

Parameters:

• route – The URL path for the GraphQL endpoint.

• schema – A Graphene schema instance.

• graphiql – Serve the in-browser GraphiQL IDE for HTML GET requests (default True).

• introspection – Allow schema-introspection queries (default True).

• max_depth – Reject queries nested deeper than this (default unlimited).

• partial_data_status – HTTP status for GraphQL responses that contain both data and errors. 200 (default) follows the GraphQL-over-HTTP spec; 400 preserves Responder 8.x behavior.

group(prefix)

Create a route group with a shared URL prefix.

Usage:

v1 = api.group("/v1")

@v1.route("/users")
def list_users(req, resp):
    resp.media = []

@v1.route("/users/{id:int}")
def get_user(req, resp, *, id):
    resp.media = {"id": id}

For routes declared in separate modules (without an API instance), use responder.Router with include_router() instead.

head(route=None, **options)

Register a route for HEAD (sugar for route(methods=["HEAD"])).

include_router(router: Router, *, prefix: str = '', tags: list[str] | None = None, dependencies: Any = None, auth: Any = <object object>) → None

Attach a standalone responder.Router’s recorded declarations.

Each recorded route is replayed through route(), so auth inheritance, Depends guards, and OpenAPI metadata behave exactly as if the route had been declared on the API directly. Group-level values compose: prefixes concatenate, tags merge (group tags first), dependencies run before route-level ones, and a route’s own auth= wins over the router’s, which wins over the one given here.

Inclusion is a snapshot — routes declared on the router afterwards are not picked up. Including the same router at two prefixes is fine, but including the same view function twice with different effective auth/tags/dependencies raises, since route metadata is attached to the view itself.

Parameters:

• router – The responder.Router to include.

• prefix – URL prefix prepended to the router’s own prefix.

• tags – OpenAPI tags merged into every included route.

• dependencies – Depends(...) guards run before every included route.

• auth – Auth helper(s) for included routes that don’t set their own.

middleware(middleware_type='http')

Register function-style HTTP middleware (decorator).

The decorated function receives the Starlette Request and a call_next callable, and returns the response to send — no ASGI class boilerplate required:

import time

@api.middleware("http")
async def add_timing(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Response-Time"] = f"{elapsed:.4f}s"
    return response

Synchronous functions work too: they are offloaded to the threadpool (like sync views), and the call_next they receive is a plain blocking callable.

Function middleware registers through the same stack as add_middleware(), so the two compose freely. Order: among user middleware — function-style or class-based alike — the most-recently-registered is the outermost and sees the request first (and the response last). Non-HTTP traffic (WebSockets, lifespan) passes through untouched.

Parameters:

middleware_type – Only "http" is supported (the default). The decorator may also be applied bare (@api.middleware).

mount(route, app)

Mounts an WSGI / ASGI application at a given route.

Parameters:

• route – String representation of the route to be used (shouldn’t be parameterized).

• app – The other WSGI / ASGI app.

ndjson(route=None, *, item_model=None, **options)

Register a typed newline-delimited JSON GET route.

on_event(event_type: str, **args: Any) → Callable[[...], Any]

Decorator for registering functions or coroutines to run at certain events Supported events: startup, shutdown

Usage:

@api.on_event('startup')
async def open_database_connection_pool():
    ...

@api.on_event('shutdown')
async def close_database_connection_pool():
    ...

options(route=None, **options)

Register a route for OPTIONS (sugar for route(methods=["OPTIONS"])).

patch(route=None, **options)

Register a route for PATCH (sugar for route(methods=["PATCH"])).

path_matches_route(path)

Given a path portion of a URL, tests that it matches against any registered route.

Parameters:

path – The path portion of a URL (e.g. "/hello"), to test all known routes against. An ASGI scope mapping is also accepted.

policy(name, auth)

Create and register a named auth policy.

The returned policy can be passed anywhere auth= accepts an auth helper. Naming a policy makes route declarations describe intent while preserving the wrapped auth scheme’s runtime and OpenAPI behavior:

admin = api.policy("admin", bearer.requires("admin"))

@api.get("/admin", auth=admin)
def dashboard(req, resp, *, user): ...

post(route=None, **options)

Register a route for POST (sugar for route(methods=["POST"])).

put(route=None, **options)

Register a route for PUT (sugar for route(methods=["PUT"])).

redirect(resp: Response, location: str, *, set_text: bool = True, status_code: int | None = None, permanent: bool = False, allow_external: bool = True) → None

Redirects a given response to a given location.

Parameters:

• resp – The Response to mutate.

• location – The location of the redirect.

• set_text – If True, sets the Redirect body content automatically.

• status_code – an API.status_codes attribute, or an integer, representing the HTTP status code of the redirect. Defaults to 307 (308 with permanent=True).

• permanent – If True, send a 308 Permanent Redirect instead of the default 307.

• allow_external – If False, refuse (with a 400) to redirect to an external URL — pass this for user-supplied locations.

property requests

A test client connected to the ASGI app. Lazily initialized.

route(route=None, *, response_model=None, params_model=None, include_in_schema=True, security=None, tags=None, summary=None, description=None, operation_id=None, deprecated=None, responses=None, examples=None, response_examples=None, openapi_extra=None, status_code=None, before=None, after=None, auth=<object object>, csrf=None, dependencies=None, _stream_mode=None, _stream_model=None, _stream_heartbeat=None, **options)

Decorator for creating new routes around function and class definitions.

Usage:

@api.route("/hello")
def hello(req, resp):
    resp.text = "hello, world!"

A route can declare its default success status with status_code=; resp.status_code is pre-seeded with it before the handler runs (assigning resp.status_code in the handler still wins), and the OpenAPI document keys the success response under it instead of 200:

@api.route("/items", methods=["POST"], status_code=201)
async def create_item(req, resp, *, item: ItemIn):
    resp.media = {"id": 1, **item.model_dump()}

Return annotations provide response validation and OpenAPI documentation without repeating the model in the decorator:

from pydantic import BaseModel

class ItemIn(BaseModel):
    name: str
    price: float

class ItemOut(BaseModel):
    id: int
    name: str
    price: float

@api.route("/items", methods=["POST"])
async def create_item(req, resp, *, item: ItemIn) -> ItemOut:
    return ItemOut(id=1, **item.model_dump())

Pydantic TypeAdapter-compatible annotations such as list[ItemOut] and ItemOut | ErrorOut also infer. Pass response_model=False to disable inference for one route.

Additional statuses can declare their own validated contract with responses={404: ErrorOut}. Use responses={404: {"model": ErrorOut, "description": "Not found"}} to add OpenAPI metadata to that contract.

Query parameters validate the same way with params_model — invalid queries get a 422, valid ones land on req.state.validated_params:

class SearchParams(BaseModel):
    q: str
    limit: int = 10

@api.route("/search", params_model=SearchParams)
async def search(req, resp):
    params = req.state.validated_params
    resp.media = {"q": params.q, "limit": params.limit}

run(**kwargs)

Run the application. Shorthand for serve() that inherits the debug setting.

Parameters:

kwargs – Keyword arguments passed through to serve().

schema(name, **options)

Decorator for creating new routes around function and class definitions.

Usage:

from marshmallow import Schema, fields
@api.schema("Pet")
class PetSchema(Schema):
    name = fields.Str()

serve(*, address=None, port=None, debug=False, server='uvicorn', port_precedence='explicit', **options)

Run the application with an ASGI server.

If the PORT environment variable is set and no explicit port= is provided, requests will be served on that port automatically to all known hosts. If both are set and disagree, the explicit port= wins by default.

Parameters:

• address – The address to bind to.

• port – The port to bind to. If none is provided, one will be selected at random.

• debug – Whether to run application in debug mode.

• server – Server backend to use: "uvicorn" (default) or "granian".

• port_precedence – "explicit" (default) lets an explicit port= win over a conflicting PORT environment variable; "env" preserves the legacy behavior.

• options – Additional keyword arguments to send to the selected server.

sse(route=None, *, event_model=None, heartbeat=None, **options)

Register a typed Server-Sent Events GET route.

The event data contract is inferred from AsyncIterator[Item] or AsyncIterator[SSE[Item]]. Pass event_model=Item when the return annotation cannot express it.

state

Application-level state. Set values at startup, read them anywhere (handlers can reach it via req.api.state).

property static_app

The Starlette StaticFiles application for serving static assets.

template(filename, *args, **kwargs)

Render a Jinja2 template file with the provided values.

Parameters:

• filename – The filename of the jinja2 template, in templates_dir.

• *args – Data to pass into the template.

• **kwargs – Data to pass into the template.

template_string(source, *args, **kwargs)

Render a Jinja2 template string with the provided values.

Parameters:

• source – The template to use, a Jinja2 template string.

• *args – Data to pass into the template.

• **kwargs – Data to pass into the template.

test_client(base_url='http://;', **options)

Return a Starlette TestClient connected to this app.

This is the configurable form of requests; pass normal Starlette TestClient options such as raise_server_exceptions=False or a custom base_url.

url_for(endpoint, **params)

Given an endpoint, returns a rendered URL for its route.

Parameters:

• endpoint – The route endpoint you’re searching for.

• params – Data to pass into the URL generator (for parameterized URLs).

websocket_route(route=None, **options)

Register a WebSocket route (sugar for route(websocket=True)).

Request

The request object is passed into every view as the first argument. It gives you access to everything the client sent — headers, query parameters, the request body, cookies, and more.

Most properties are synchronous, but reading the body requires await because it involves I/O.

Common patterns:

# The request method — UPPERCASE: "GET", "POST", ...
if req.method == "POST":
    ...

# Headers (case-insensitive)
token = req.headers.get("Authorization")

# Every raw line of a repeated header, in order
hops = req.headers.get_list("X-Forwarded-For")

# Query parameters: /search?q=python&page=2
query = req.params["q"]

# JSON body (async handlers)
data = await req.media()

# ...or synchronously, from a sync handler
data = req.media_sync()
body = req.text_sync

# Form data and file uploads
form = await req.media("form")
files = await req.media("files")

# Client info
ip, port = req.client
is_https = req.is_secure

Note

req.method is a plain UPPERCASE str ("GET", "POST"). Comparisons are case-sensitive, so compare against uppercase literals: req.method == "GET".

await req.media("files") returns {name: UploadFile} (streamed, spooled to disk). See Parameter Markers for the typed File() form.

For reading typed query parameters, headers, and cookies straight off the signature, see Parameter Markers below.

class responder.Request(scope, receive, api=None, formats=None)

An HTTP request, passed to each view as the first argument.

Provides access to headers, cookies, query parameters, the request body, session data, and more. Most properties are synchronous; reading the body (via content, text, or media()) requires await.

accepts(content_type: str) → bool

Whether the client’s Accept header allows content_type.

Honors media ranges (*/*, type/*) and q-values. Per RFC 9110 §12.5.1 the most specific matching range governs, so an explicit q=0 range (not acceptable) excludes the type even when a broader range (*/* or type/*) would match. An absent Accept header accepts anything. content_type may be a full media type (application/json) or a bare subtype token (json).

property apparent_encoding

The apparent encoding, detected automatically. Must be awaited.

Valid UTF-8 bodies (the overwhelmingly common case) are recognized with a cheap strict decode. Otherwise, uses chardet for detection if installed (off the event loop, since detection is CPU-bound), falling back to UTF-8.

property client

The client’s address as a (host, port) named tuple, or None.

property content

The Request body, as bytes. Must be awaited.

property cookies: dict

The cookies sent in the Request, as a dictionary.

property csrf_input

A hidden <input> carrying csrf_token, for HTML forms.

Returns markup-safe HTML, so it renders unescaped in autoescaped Jinja templates: pass the request into the template context and drop {{ req.csrf_input }} inside each <form method="post">.

property csrf_token

The session’s CSRF token, minted on first access.

Embed it in responses (or read it client-side) and send it back on unsafe requests in an X-CSRF-Token header or csrf_token form field. Requires sessions; see session.

property encoding

The encoding of the Request’s body. Can be set, manually. Must be awaited.

property full_url: str

The full URL of the Request, query parameters and all.

property headers: CaseInsensitiveDict

A case-insensitive dictionary, containing all headers sent in the Request.

Single-value access (req.headers["X-Forwarded-For"]) returns the last value received for a repeated header; use req.headers.get_list("X-Forwarded-For") to read every raw line, in order.

property is_json: bool

Returns True if the request content type is JSON.

property is_secure: bool

True if the request was made over HTTPS.

property last_event_id: str | None

The SSE Last-Event-ID header sent by a reconnecting client.

Use it to resume an event stream from where the client left off.

async media(format: str | Callable | None = None) → Any

Renders incoming json/yaml/form data as Python objects. Must be awaited.

Parameters:

format – The name of the format being used. Alternatively, accepts a custom callable for the format type.

media_sync(format: str | Callable | None = None) → Any

Synchronous media(), for use from sync handlers.

Responder runs sync handlers in a worker thread, so this safely bridges to the event loop to read and parse the body. Calling it from an async handler raises — use await req.media() there instead.

Usage:

@api.route("/", methods=["POST"])
def create(req, resp):
    data = req.media_sync()

property method: str

The HTTP method, UPPER-cased ("GET", "POST", …).

Comparisons are case-sensitive; compare against uppercase literals (req.method == "GET"). Use .lower() for the lowercase string.

property mimetype: str

The MIME type of the request body, from the Content-Type header.

property params: QueryDict

A dictionary of the parsed query parameters used for the Request.

property path_params: dict

The path parameters extracted from the URL route.

preferred_media_type(candidates)

Pick the candidate media type the client’s Accept header prefers.

Candidates are ranked by the q-value of the most specific matching media range (RFC 9110 §12.5.1); ties — and requests without an Accept header — keep the order of candidates, so put the server’s preferred default first. Returns None when the client accepts none of them.

Usage:

preferred = req.preferred_media_type(
    ["application/json", "text/csv"]
)

Parameters:

candidates – An iterable of media types (application/json) or bare subtype tokens (json).

property session

The session data, in dict form, from the Request.

Requires sessions enabled (a secret_key or session_backend, with sessions != False); raises RuntimeError otherwise.

property state: State

Use the state to store additional information.

This can be a very helpful feature, if you want to hand over information from a middelware or a route decorator to the actual route handler.

Usage: request.state.time_started = time.time()

async stream()

Iterate over the raw request body in chunks, without buffering.

Useful for large uploads. Once streamed, the body cannot be read again via content, text, or media().

Usage:

@api.route("/upload", methods=["POST"])
async def upload(req, resp):
    async with await anyio.open_file(path, "wb") as f:
        async for chunk in req.stream():
            await f.write(chunk)

property text

The Request body, as unicode. Must be awaited.

property text_sync

Synchronous text, for use from sync handlers (see media_sync()).

property url

The parsed URL of the Request.

Response

The response object is passed into every view as the second argument. Mutate it to control what gets sent back to the client — the body, status code, headers, and cookies.

Common patterns:

resp.text = "plain text"            # text/plain
resp.html = "<h1>Hello</h1>"        # text/html
resp.media = {"key": "value"}       # application/json
resp.content = b"raw bytes"         # application/octet-stream

# Serve files. Pass root= to jail a user-supplied path under a directory —
# a "../" or symlink escape returns 404 instead of leaking the filesystem:
resp.file("reports/q3.pdf", root="exports")       # auto content-type
resp.stream_file("exports/big.csv", root="exports")  # streamed

resp.status_code = 201
resp.headers["X-Custom"] = "value"
resp.cookies["session"] = "abc123"

# Common response helpers:
resp.created({"id": 1}, location="/items/1")  # 201 + Location
resp.no_content()                            # 204 + empty body
resp.problem(409, "Already exists")          # application/problem+json

# Redirect (external targets allowed by default; pass
# allow_external=False to refuse off-site URLs):
resp.redirect("/dashboard")

Note

Handlers can also return the body Flask-style instead of mutating resp: return body, return body, status, or return body, status, headers. Pydantic models and dataclasses serialize natively. A supported return annotation validates the body and supplies its OpenAPI/generated-client type; use response_model=False to opt out. Status-specific contracts use responses={404: ErrorModel}, or a mapping with model plus additional OpenAPI metadata. resp.session is a read/write view of req.session and raises RuntimeError when the app is built with sessions=False.

class responder.Response(req, *, formats, auto_etag=False, auto_vary=False)

An HTTP response, passed to each view as the second argument.

Mutate this object to control what gets sent back to the client. Set text, html, media, or content to define the body. Use headers and set_cookie() to control metadata.

Variables:

• text – Set the response body as plain text (sets Content-Type: text/plain).

• html – Set the response body as HTML (sets Content-Type: text/html).

• media – Set a Python object (dict, list) to be serialized as JSON (or negotiated format).

• content – Set the raw response body as bytes.

• status_code – The HTTP status code (e.g. 200, 404). Defaults to 200 if not set.

• headers – A case-insensitive (case-preserving) dict of response headers.

• cookies – A SimpleCookie holding cookies to set on the response.

• session – A dict of session data. Changes are persisted in a signed cookie.

• etag – Entity tag for the response. When the request’s If-None-Match matches on GET/HEAD, an automatic 304 Not Modified is sent instead of the body. On state-changing methods, If-Match / If-None-Match preconditions are enforced against it with an automatic 412 Precondition Failed.

• last_modified – A datetime (or HTTP-date string) for Last-Modified. Honors If-Modified-Since with automatic 304 responses, and If-Unmodified-Since on state-changing methods with automatic 412 responses.

background(func, *args, **kwargs)

Schedule a task to run after the response has been sent.

Unlike api.background (which runs immediately in a thread pool), tasks scheduled here are deferred until the client has the response, so they never delay it. Sync and async functions both work. Multiple tasks run in the order scheduled.

Usage:

@api.route("/signup", methods=["POST"])
async def signup(req, resp):
    resp.media = {"ok": True}
    resp.background(send_welcome_email, "user@example.com")

cache_control(**directives)

Set the Cache-Control header from keyword directives.

Underscores become hyphens; True renders a bare directive, other values render name=value:

resp.cache_control(public=True, max_age=3600)
# Cache-Control: public, max-age=3600

resp.cache_control(no_store=True)
# Cache-Control: no-store

created(media=None, *, location=None, headers=None)

Mark the response as 201 Created.

Optionally set a JSON-serializable body and a Location header:

resp.created({"id": item.id}, location=f"/items/{item.id}")

delete_cookie(key, *, path='/', domain=None, secure=False, httponly=True, samesite='lax')

Expire a cookie on the client (empty value, Max-Age=0).

Pass the same path/domain the cookie was set with so the browser matches and drops it.

Usage:

resp.delete_cookie("token")

download(path, *, filename=None, content_type=None, root=None, conditional=True)

Serve a file as an attachment, prompting the browser to download.

Streams the file (with range support, so downloads are resumable) and sets Content-Disposition.

Parameters:

• path – Path to the file to serve.

• filename – Download name presented to the client. Defaults to the file’s own name.

• content_type – Optional MIME type override.

• root – If given, path is resolved under this directory and any escape attempt yields a 404 (see file()).

file(path, *, content_type=None, root=None, conditional=True)

Serve a file from disk as the response.

Supports HTTP range requests (Range: bytes=...) with 206 partial responses. The file’s bytes are read in a worker thread when the response is sent, so calling this from an async handler never blocks the event loop.

Parameters:

• path – Path to the file to serve.

• content_type – Optional MIME type override.

• root – If given, path is resolved under this directory and any attempt to escape it (via .. or a symlink) yields a 404 — use this whenever path is built from user input.

• conditional – If True (default), set a stat-based ETag and Last-Modified so conditional requests get a 304 (and the file’s bytes aren’t read to compute it).

no_content(*, headers=None)

Mark the response as 204 No Content and clear any response body.

property ok

True if the status code is in the 2xx range (success).

Reads as 200 until a status code has been set, so it never raises.

problem(status_code, detail=None, *, title=None, type=None, instance=None, errors=None, headers=None, **extensions)

Send an RFC 9457 application/problem+json response.

API-level problem_handler enrichment and request IDs are applied, then explicit helper arguments and extension members are layered on top.

redirect(location: str, *, set_text: bool = True, status_code: int | None = None, permanent: bool = False, allow_external: bool = True) → None

Redirect the client to a different URL.

Parameters:

• location – The URL to redirect to.

• set_text – If True, set a default redirect message as the body.

• status_code – An explicit HTTP status code. When omitted, the redirect is a method-preserving 307 Temporary Redirect (or 308 Permanent Redirect with permanent=True).

• permanent – If True, send a 308 Permanent Redirect instead of the default 307. Ignored when an explicit status_code is given.

• allow_external – If False, refuse (with a 400) to redirect to an absolute or protocol-relative URL — pass this whenever location comes from user input, to prevent open redirects.

render(template, *args, **kwargs)

Render a Jinja2 template as the HTML response body.

Shorthand for resp.html = api.template(...), using the owning API’s templates_dir.

Parameters:

• template – The template filename.

• *args – Data to pass into the template.

• **kwargs – Data to pass into the template.

Usage:

@api.route("/")
def home(req, resp):
    resp.render("home.html", user="kenneth")

reset_for_error()

Discard a previously prepared success response before framework errors.

property session

The session dict (delegates to the request; requires sessions on).

set_cookie(key, value='', expires=None, path='/', domain=None, max_age=None, secure=False, httponly=True, samesite='lax')

Set a cookie on the response with full control over directives.

Parameters:

• key – The cookie name.

• value – The cookie value.

• expires – Expiration date string (e.g. "Thu, 01 Jan 2026 00:00:00 GMT").

• path – URL path the cookie applies to (default "/").

• domain – Domain the cookie is valid for.

• max_age – Maximum age in seconds before the cookie expires.

• secure – If True, cookie is only sent over HTTPS.

• httponly – If True (default), cookie is inaccessible to JavaScript.

• samesite – Cross-site behavior: "lax" (default), "strict", "none", or None to omit the directive.

Usage:

resp.set_cookie(
    "token", value="abc123",
    max_age=3600, secure=True, httponly=True,
)

sse(func=None, *args, heartbeat=None, **kwargs)

Set up a Server-Sent Events response from an async generator.

Each yielded value becomes one event:

• a dict with any of data, event, id, retry (data that is a dict/list is JSON-encoded);

• {"comment": "..."} for a comment/keepalive line;

• a str (or anything else) is sent as the data field;

• bytes are written verbatim (pre-formatted frame).

Usage:

@api.route("/events")
async def events(req, resp):
    @resp.sse
    async def stream():
        for i in range(10):
            yield {"data": {"n": i}, "id": i}

Pass heartbeat= (seconds) to emit a keepalive comment during idle periods, so proxies don’t drop the connection:

@resp.sse(heartbeat=15)
async def stream(): ...

On reconnect the client sends the last id it saw; read it with Request.last_event_id.

Parameters:

• func – The async generator function (or omit when using the @resp.sse(heartbeat=...) form).

• heartbeat – Idle keepalive interval in seconds (None = off).

property status_code_safe: int

Return the status code, raising RuntimeError if it hasn’t been set.

stream(func, *args, **kwargs)

Set up a streaming response from an async generator function.

The generator yields chunks of bytes that are sent to the client as they are produced, without buffering the full response in memory.

Usage:

@api.route("/stream")
async def stream_data(req, resp):
    @resp.stream
    async def body():
        for i in range(10):
            yield f"chunk {i}\n".encode()

Parameters:

func – An async generator function that yields response chunks.

stream_file(path, *, content_type=None, chunk_size=8192, root=None, conditional=True)

Stream a file without loading it entirely into memory.

Supports HTTP range requests (Range: bytes=...) with 206 partial responses, enabling video seeking and resumable downloads.

Parameters:

• path – Path to the file.

• content_type – Optional MIME type override.

• chunk_size – Size of chunks to read (default 8192 bytes).

• root – If given, path is resolved under this directory and any attempt to escape it (via .. or a symlink) yields a 404 — use this whenever path is built from user input.

• conditional – If True (default), set a stat-based ETag and Last-Modified so conditional requests get a 304.

vary(*values)

Add one or more field names to the response Vary header.

Tokens are merged with any existing Vary value and de-duplicated case-insensitively:

resp.vary("Accept", "Accept-Language")

Typed Streams

Use API.sse() and API.ndjson() for item-by-item validation, serialization, OpenAPI schemas, and generated streaming clients. SSE is the optional metadata envelope for a typed Server-Sent Event.

class responder.SSE(data: T | None = None, event: str | None = None, id: str | int | None = None, retry: int | None = None, comment: str | None = None)

One typed Server-Sent Event with optional protocol metadata.

Yield the data model directly when no metadata is needed. Use SSE for named events, resumable ids, retry hints, or comment-only keepalives:

yield SSE(BuildEvent(...), event="progress", id=build.id)
yield SSE(comment="still working")

Parameter Markers

Inject validated query parameters, headers, cookies, and path parameters straight into a handler’s signature. A marker goes in the default slot of a keyword-only argument — it is not a decorator:

from responder import Query, Header

@api.route("/search")
def search(req, resp, *,
           q: str = Query(...),            # required
           limit: int = Query(10),         # optional, defaults to 10
           tags: list[str] = Query(...),   # repeated keys: ?tags=a&tags=b
           token: str = Header(None, alias="X-Token")):
    resp.media = {"q": q, "limit": limit, "tags": tags}

Query(...) (an Ellipsis) marks a required parameter; Query(value) supplies a default. Each value is coerced to the parameter’s type annotation with Pydantic; a missing required value or a coercion failure returns 422 Unprocessable Entity with a body of {"errors": [...]} aggregating every failing parameter.

• Query() reads the query string. A list / list[int] annotation collects repeated keys.

• Header() reads request headers; the parameter name is matched with underscores converted to dashes (user_agent → user-agent) unless you pass alias=.

• Cookie() reads cookies by name (no underscore conversion).

• Path() re-validates or renames a path parameter (Path(..., alias="uid")). A path parameter always wins over a same-named query/header/cookie marker.

Markers also accept Pydantic field constraints, which are enforced at runtime (returning 422 on violation) and emitted into the schema, along with description= and deprecated=:

@api.route("/search")
def search(req, resp, *,
           q: str = Query(..., min_length=3, description="search term"),
           limit: int = Query(10, ge=1, le=100)):
    ...

An unknown keyword (a typo such as Query(dafault=5)) raises immediately.

Markers may also be written in PEP 593 Annotated form, which keeps the parameter’s default value in the usual slot:

from typing import Annotated

def search(req, resp, *, q: Annotated[str, Query(min_length=3)] = "all"):
    ...

Markers also drive the generated OpenAPI parameters and add an automatic 422 to validating routes. For full request/response validation with Pydantic models, see Building a REST API.

File uploads and form fields

File() injects an uploaded file as an UploadFile (read it with await f.read(), or stream it in chunks — large uploads are spooled to disk, not held in memory). Form() injects a form field (urlencoded or multipart), coerced and validated like Query():

from responder import File, Form, UploadFile

@api.route("/upload", methods=["POST"])
async def upload(req, resp, *,
                 document: UploadFile = File(...),
                 title: str = Form(...),
                 tags: list[str] = Form([])):
    saved = await document.save("/srv/uploads/document.bin")
    resp.media = {"name": document.filename, "path": str(saved)}

A sequence annotation (list[UploadFile]) collects multiple files sent under one field name. These routes generate a multipart/form-data (or application/x-www-form-urlencoded) request body in OpenAPI, so the interactive docs show a file picker. await req.media("files") returns the same UploadFile objects keyed by field name.

UploadFile.save(path) streams the upload to disk and returns the resulting Path. Pass create_parents=True to create the parent directory first, or seek_start=False if you intentionally want to save from the file’s current read position.

Binding a whole form to a Pydantic model

Instead of one marker per field, a parameter annotated with a Pydantic model and a Form(...) default binds the entire parsed form (urlencoded or multipart) in one shot — fields are coerced by the model, model defaults apply, and a validation failure returns the same 422 payload as JSON body models, with each error located at ["form", <field>]:

from pydantic import BaseModel
from responder import Form

class ProfileForm(BaseModel):
    name: str
    age: int = 0
    tags: list[str] = []          # repeated form keys collect into lists

@api.route("/profiles", methods=["POST"])
async def create(req, resp, *, profile: ProfileForm = Form(...)):
    resp.media = profile.model_dump()

Uploaded files bind to model fields declared as UploadFile — Pydantic requires arbitrary_types_allowed for that:

from pydantic import ConfigDict
from responder import UploadFile

class AvatarForm(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    name: str
    avatar: UploadFile

Field aliases are honored (the form key is the alias), and Form(None) makes the whole model optional: an empty form yields the default instead of a 422. Markers — including form models — work identically on class-based view methods, and Query/Header/Cookie/Path markers also resolve on WebSocket handlers from the handshake request (a validation failure there closes the socket with code 1008).

Explicit dependencies

Use Depends() when one route needs a local provider and you don’t want to register it app-wide with api.dependency:

from responder import Depends

def current_user(req):
    return decode_user(req.headers.get("Authorization"))

@api.route("/me")
def me(req, resp, *, user=Depends(current_user)):
    resp.media = {"user": user}

The provider follows the same rules as registered dependencies: it may be sync, async, a generator, or an async generator, and it may receive the current request or registered dependencies.

Use dependencies=[Depends(...)] when a dependency is a guard or setup step and the handler does not need its return value. It participates in dependency registration/caching, sub-dependency resolution, and generator teardown:

def require_user(req):
    if "Authorization" not in req.headers:
        responder.abort(401, detail="Not authenticated")

@api.route("/private", dependencies=[Depends(require_user)])
def private(req, resp):
    resp.media = {"ok": True}

Use before= / after= for raw route hooks: imperative code that should run around the handler without entering the dependency graph. Hooks may short-circuit by setting resp.status_code and are best for request/response mutation or simple guards that do not need dependency caching, sub-dependencies, OpenAPI security integration, or generator teardown:

def require_json(req, resp):
    if not req.is_json:
        resp.status_code = 415
        resp.media = {"error": "JSON required"}

@api.route("/events", before=require_json)
def events(req, resp):
    resp.media = req.headers

The three paths are intentionally distinct:

• Handler parameters with Depends(...) resolve a value and pass it into the handler.

• Route dependencies=[Depends(...)] resolve graph-aware providers for their side effects and ignore the return value.

• Route before= / after= hooks are raw hooks; they are not dependency providers.

Execution order for a route is:

1. Global before_request hooks

2. Per-route before hooks

3. Per-route auth helpers

4. Route validation/input parsing

5. Side-effect dependencies

6. Handler view(s)

7. Response-model validation

8. Per-route and global after hooks

responder.Query(default=Ellipsis, **kwargs)

Mark a parameter as coming from the query string.

responder.Header(default=Ellipsis, **kwargs)

Mark a parameter as coming from a request header (_ → - by default).

responder.Cookie(default=Ellipsis, **kwargs)

Mark a parameter as coming from a request cookie.

responder.Path(default=Ellipsis, **kwargs)

Mark a parameter as coming from a path parameter.

responder.Form(default=Ellipsis, **kwargs)

Mark a parameter as a form field (urlencoded or multipart body).

responder.File(default=Ellipsis, **kwargs)

Mark a parameter as an uploaded file; injects an UploadFile (or a list of them for a sequence annotation).

responder.Depends(provider)

Inject the value returned by provider into a handler parameter.

Unlike named api.dependency(...) providers, Depends is local to one handler and does not require registration on the app.

Route Groups

Group related routes under a shared URL prefix — useful for API versioning and organizing large applications:

v1 = api.group("/v1")

@v1.route("/users")
def list_users(req, resp):
    resp.media = []

class responder.api.RouteGroup(api, prefix)

A group of routes with a shared URL prefix.

Before-request hooks registered on a group only run for requests whose path falls under the group’s prefix.

before_request(websocket=False)

Register a hook that runs before requests under this group’s prefix.

Standalone Routers

responder.Router records route declarations without an API instance, so routes can live in separate modules (like Flask blueprints) and be attached later — with prefix composition, nesting, and group-level tags / dependencies / auth. See Composing Apps with Routers for the guide.

from responder import Router

router = Router(prefix="/users", tags=["users"])

@router.get("/{user_id:int}")
def get_user(req, resp, *, user_id):
    resp.media = {"id": user_id}

api.include_router(router, prefix="/v1")

class responder.Router(prefix: str = '', *, tags: Iterable[str] | None = None, dependencies: Any = None, auth: Any = <object object>)

A standalone, composable collection of route declarations.

Unlike api.group(), a Router needs no API at creation: it records declarations and replays them when included, so routes can be declared in separate modules without circular imports:

# users.py
from responder import Router

router = Router(prefix="/users", tags=["users"])

@router.route("/{user_id:int}")
def get_user(req, resp, *, user_id):
    resp.media = {"id": user_id}

# app.py
api.include_router(users.router, prefix="/v1")  # GET /v1/users/{user_id}

Routers nest via include_router(); prefixes compose, tags merge (group tags first, duplicates dropped), dependencies concatenate (group guards run first), and the innermost explicit auth wins. Inclusion is a snapshot: routes declared after a router has been included are not picked up by that earlier inclusion.

Parameters:

• prefix – URL prefix for every route in this router (e.g. "/users").

• tags – OpenAPI tags applied to every route in this router.

• dependencies – Depends(...) guards run before every route in this router.

• auth – Auth helper(s) required by every route in this router (individual routes may still override with their own auth=).

add_route(route: str | None = None, endpoint: Any = None, *, tags: Iterable[str] | None = None, dependencies: Any = None, auth: Any = <object object>, before_request: bool = False, websocket: bool = False, **options: Any) → None

Record a route declaration (imperative form of route()).

before_request(websocket: Any = False) → Callable

Record a before-request hook scoped to this router’s mounted prefix.

Works both bare and called (@router.before_request or @router.before_request()). Once included, the hook only runs for request paths under the prefix the router was mounted at; if the composed prefix is empty, it runs for every request.

Parameters:

websocket – If True, register as a WebSocket hook instead of HTTP.

delete(route: str | None = None, **options: Any) → Callable

Record a DELETE route (sugar for route(methods=["DELETE"])).

get(route: str | None = None, **options: Any) → Callable

Record a GET route (sugar for route(methods=["GET"])).

head(route: str | None = None, **options: Any) → Callable

Record a HEAD route (sugar for route(methods=["HEAD"])).

include_router(router: Router, *, prefix: str = '', tags: Iterable[str] | None = None, dependencies: Any = None, auth: Any = <object object>) → None

Nest another router’s declarations under this one.

The child’s declarations are copied at call time (a snapshot): routes added to the child afterwards are not seen. Prefixes compose, tags merge, dependencies concatenate, and a child route’s own auth wins over the values given here.

Parameters:

• router – The Router to include.

• prefix – Extra URL prefix, prepended to the child’s own prefix.

• tags – OpenAPI tags to merge into the included routes.

• dependencies – Depends(...) guards prepended to the included routes’ dependencies.

• auth – Auth helper(s) for included routes that don’t set their own.

ndjson(route: str | None = None, *, item_model: Any = None, **options: Any) → Callable

Record a typed newline-delimited JSON route.

options(route: str | None = None, **options: Any) → Callable

Record an OPTIONS route (sugar for route(methods=["OPTIONS"])).

patch(route: str | None = None, **options: Any) → Callable

Record a PATCH route (sugar for route(methods=["PATCH"])).

post(route: str | None = None, **options: Any) → Callable

Record a POST route (sugar for route(methods=["POST"])).

put(route: str | None = None, **options: Any) → Callable

Record a PUT route (sugar for route(methods=["PUT"])).

route(route: str | None = None, *, tags: Iterable[str] | None = None, dependencies: Any = None, auth: Any = <object object>, **options: Any) → Callable

Decorator recording a route declaration; matches API.route().

All keyword options (methods=, response_model=, summary=, …) are stored and replayed through API.route() at inclusion time.

sse(route: str | None = None, *, event_model: Any = None, heartbeat: float | None = None, **options: Any) → Callable

Record a typed Server-Sent Events route.

websocket_route(route: str | None = None, **options: Any) → Callable

Record a WebSocket route (sugar for route(websocket=True)).

Route-Local Hooks and Auth

Use global before_request / after_request hooks for behavior that applies to the whole app. For behavior that belongs to one endpoint, pass before= and after= to the route decorator:

def require_json(req, resp):
    if not req.is_json:
        resp.status_code = 415
        resp.media = {"error": "JSON required"}

def add_audit_header(req, resp):
    resp.headers["X-Audited"] = "1"

@api.post("/events", before=require_json, after=add_audit_header)
async def events(req, resp):
    resp.media = await req.media()

Authentication helpers from responder.ext.auth can be attached directly with auth=. Responder enforces the scheme, registers OpenAPI security when OpenAPI is enabled, stores the principal on req.state.user / req.state.auth, and injects it into a user, principal, or auth handler parameter:

from responder.ext.auth import BearerAuth

bearer = BearerAuth(tokens=["s3cret"])

@api.get("/me", auth=bearer)
def me(req, resp, *, user):
    resp.media = {"user": user}

Use API(auth=bearer) when most routes share the same auth scheme. Routes inherit the app auth by default; pass auth=None on public routes such as /login or /health.

Use api.policy(name, auth) to give a reusable auth requirement an application-facing name while keeping the wrapped scheme’s runtime and OpenAPI behavior unchanged:

admin = api.policy("admin", bearer.requires("admin"))

@api.get("/admin", auth=admin)
def admin_dashboard(req, resp, *, user):
    resp.media = {"user": user}

Use auth.optional() when credentials should be accepted but not required. Missing credentials inject None into user / principal / auth; invalid credentials still return 401. OpenAPI documents both anonymous and authenticated access:

optional = bearer.optional()

@api.get("/maybe", auth=optional)
def maybe(req, resp, *, user):
    resp.media = {"user": user}

Use auth.requires(...) or ScopedAuth when a route needs lightweight scope or role checks after authentication. Scopes are read from a principal’s scopes or roles attribute/key, either as a space-delimited string or an iterable. Missing scopes return 403 and OpenAPI security requirements include the required scopes:

admin = bearer.requires("items:write")

@api.post("/items", auth=admin)
def create_item(req, resp, *, user):
    resp.media = {"user": user}

Background Queue

Run tasks in background threads without blocking the response. Available as api.background:

@api.route("/submit")
async def submit(req, resp):
    data = await req.media()

    @api.background.task
    def process(data):
        # runs in a thread pool
        ...

    process(data)
    resp.media = {"status": "accepted"}

class responder.background.BackgroundQueue(n=None)

A queue for running tasks in background threads.

Uses a ThreadPoolExecutor sized to the number of CPUs. Access it via api.background.

Usage:

# As a decorator — fire and forget
@api.background.task
def send_email(to, subject):
    ...

send_email("user@example.com", "Hello")

# Async functions work too — run via ``asyncio.run`` on the worker
@api.background.task
async def refresh_cache():
    ...

# Direct submission
future = api.background.run(send_email, "user@example.com", "Hello")

# As a callable (supports async functions)
await api.background(send_email, "user@example.com", "Hello")

run(f, *args, **kwargs)

Submit a function to run in a background thread.

Async functions are supported: they are driven to completion on the worker thread via asyncio.run (previously they silently produced a never-awaited coroutine and the job never ran).

Parameters:

f – The function to run.

Returns:

A concurrent.futures.Future for the result.

shutdown(wait=True)

Stop accepting new tasks and, by default, drain in-flight ones.

Called automatically at application shutdown so fire-and-forget tasks submitted via run()/task() are given a chance to finish rather than being abandoned when the process exits.

Parameters:

wait – Block until running tasks complete (default True).

task(f)

Decorator that wraps a function to run in the background thread pool.

The decorated function returns a Future instead of blocking. Exceptions are printed to stderr via traceback.

Parameters:

f – The function to wrap.

Query Dict

A dictionary subclass for query string parameters with multi-value support. Behaves like a normal dict for single values, but supports getlist() for parameters that appear multiple times (e.g. ?tag=a&tag=b).

class responder.models.QueryDict(query_string)

A dictionary for query string parameters that handles multi-value keys.

Single-value access returns the last value for a key. Use get_list() to retrieve all values for a multi-value parameter.

get(key, default=None)

Return the last data value for the passed key. If key doesn’t exist or value is an empty list, return default.

get_list(key, default=None)

Return the list of values for the key. If key doesn’t exist, return a default value.

items()

Yield (key, value) pairs, where value is the last item in the list associated with the key.

items_list()

Yield (key, value) pairs, where value is the the list.

setdefault(key, default=None)

Like dict.setdefault, but stores scalars via __setitem__() so single-value reads return them unchanged.

update(other=None, /, **kwargs)

Update from a mapping/iterable and keyword args, storing scalar values via __setitem__() (lists are stored as-is). Merging from another QueryDict copies each key’s full stored list, so multi-value parameters survive intact.

Headers Dict

The case-insensitive (case-preserving) mapping behind req.headers and resp.headers. Lookups match header names case-insensitively; on request headers, single-value access returns the last value received for a repeated header, while get_list() returns every raw line, in order:

hops = req.headers.get_list("X-Forwarded-For")
# ["203.0.113.7", "198.51.100.2"] — one entry per proxy hop

class responder.models.CaseInsensitiveDict(data: Any = None, /, **kwargs: Any)

A case-insensitive, case-preserving dict for HTTP headers.

Lookups, membership tests, deletion, and updates match keys case-insensitively, while iteration preserves the casing each key was last set with — so d["content-type"] = ... replaces an existing Content-Type entry rather than adding a second one.

HTTP allows the same header name to appear on multiple lines; when the mapping is built from such raw pairs (as req.headers is), single-value access returns the last value while get_list() returns every value, in the order received.

get_list(key: str, default: Any = None) → Any

Return every value received for key, in order.

HTTP permits a header to appear on multiple lines (proxies append separate X-Forwarded-For/Via lines; HTTP/2 clients may split headers similarly), and single-value access returns only the last one. This returns them all, in the order they arrived. A missing key returns default (or [] when no default is given).

Usage:

hops = req.headers.get_list("X-Forwarded-For")

Rate Limiter

Sliding-window rate limiter (fixed-window with the Redis backends). Limits requests per client IP address — or per anything, via key= — and returns 429 Too Many Requests when exceeded:

from responder.ext.ratelimit import RateLimiter

limiter = RateLimiter(requests=100, period=60)  # 100 req/min
limiter.install(api)

Response headers: X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset (seconds until the window resets), and Retry-After (when limited).

Pass key= (a req -> str callable) to bucket by API key or user instead of client IP, and fail_open=True to let requests through with a warning when the backend is unreachable (the default answers 503).

The in-memory backend is per-process. For multi-worker or distributed deploys, pass a shared store via backend= — RedisBackend (sync) or AsyncRedisBackend (async, via redis.asyncio):

from responder.ext.ratelimit import RateLimiter, AsyncRedisBackend

limiter = RateLimiter(requests=100, period=60, backend=AsyncRedisBackend())

class responder.ext.ratelimit.RateLimiter(requests=100, period=60, backend=None, trust_proxy_headers=False, key=None, fail_open=False)

Per-client request rate limiter.

The limiting algorithm depends on the backend: MemoryBackend (the default) is a sliding window over the last period seconds, while RedisBackend/AsyncRedisBackend count within fixed windows.

Usage:

from responder.ext.ratelimit import RateLimiter

limiter = RateLimiter(requests=100, period=60)  # 100 req/min
limiter.install(api)

Enforcement can also be hand-rolled in a before-request hook when you need custom logic around it:

@api.route(before_request=True)
def rate_limit(req, resp):
    limiter.check(req, resp)

Prefer install() (or limit()) when you don’t: they enforce the same way, but a manual check() call is invisible to the OpenAPI generator, so the schema won’t document the 429/503 responses.

To rate-limit a single route, apply limit() beneath @api.route. Give each route its own RateLimiter for an independent budget:

expensive_limiter = RateLimiter(requests=5, period=60)

@api.route("/expensive")
@expensive_limiter.limit
async def expensive(req, resp):
    ...

async acheck(req, resp)

Check the rate limit, awaiting an async backend (or off-loading a sync one to a thread). Works with any backend.

check(req, resp)

Check the rate limit synchronously. Sets a 429 if exceeded.

Requires a sync backend (hit); use acheck() for async-only backends.

install(api)

Install as a before_request hook on the API (async, any backend).

limit(f)

Decorator that rate-limits a single route handler.

Apply beneath @api.route(). When the limit is exceeded, the handler is skipped and a 429 response is returned. Works on both function views (req, resp) and class-based-view methods (self, req, resp) — the request/response are located by type among the positional arguments, so a bound method’s leading self does not shift them.

Status Code Helpers

Convenience functions for checking which category a status code falls into. Useful in middleware and after-request hooks:

from responder.status_codes import is_200, is_400, is_500

@api.after_request()
def log_errors(req, resp):
    if is_400(resp.status_code) or is_500(resp.status_code):
        print(f"Error: {req.method} {req.url.path} -> {resp.status_code}")

responder.status_codes.is_100(status_code)

responder.status_codes.is_200(status_code)

responder.status_codes.is_300(status_code)

responder.status_codes.is_400(status_code)

responder.status_codes.is_500(status_code)

Errors and Exceptions

Use abort() to short-circuit a request with a rendered HTTP error from anywhere in a handler, hook, or dependency — no Starlette import required. Unlike setting resp.status_code, it halts the handler:

from responder import abort

@api.route("/admin")
def admin(req, resp):
    if not req.session.get("is_admin"):
        abort(403, detail="Forbidden")

responder.abort(status_code, *, detail=None, headers=None, title=None, type=None, instance=None, errors=None, **extensions)

Short-circuit the request with an HTTP error response.

Raises an HTTPException that Responder renders (as JSON or text per the client’s Accept header). Use it to bail out from anywhere in a handler or dependency without importing Starlette directly.

Usage:

from responder import abort

@api.route("/admin")
def admin(req, resp):
    if not req.session.get("is_admin"):
        abort(403, detail="Forbidden")

Passing any of the RFC 9457 members — title, type, instance, errors, or extension keywords — raises Problem instead, carrying them into the rendered problem-details payload:

abort(
    409,
    detail="Plan quota exhausted",
    type="https://api.example.com/errors/quota-exceeded",
    balance=0,
)

Parameters:

• status_code – The HTTP status code (e.g. 404).

• detail – Optional error message; defaults to the status phrase.

• headers – Optional dict of headers to attach to the error response.

• title – Optional problem summary; defaults to the status phrase.

• type – Optional URI identifying the problem type.

• instance – Optional URI for this occurrence; defaults to the request path in the rendered payload.

• errors – Optional list of structured error dicts.

• extensions – Extra keyword arguments become top-level extension members of the problem payload.

For typed problem catalogs, raise Problem — an HTTPException subclass carrying the full set of RFC 9457 members (type, title, instance, and arbitrary extension members) into the rendered application/problem+json payload. It flows through the same machinery as framework errors, including API(problem_handler=...) enrichment and the app’s JSON encoder:

from responder import Problem

@api.route("/quota")
def quota(req, resp):
    raise Problem(
        409,
        "You have used all 100 requests for today.",
        title="Quota Exceeded",
        type="https://api.example.com/errors/quota-exceeded",
        balance=0,
    )

Passing any RFC 9457 member to abort() raises a Problem for you, so abort(409, type="...", balance=0) is equivalent.

class responder.Problem(status_code: int, detail: str | None = None, *, title: str | None = None, type: str | None = None, instance: str | None = None, errors: list[dict] | None = None, headers: dict[str, str] | None = None, **extensions: Any)

A raisable RFC 9457 problem-details error.

Raise it from any view, hook, or dependency to short-circuit the request with a full application/problem+json response, rendered through the same pipeline as framework errors — content negotiation, API-level problem_handler enrichment, request IDs, and the app’s JSON encoder:

from responder import Problem

@api.route("/quota")
def quota(req, resp):
    raise Problem(
        409,
        "You have used all 100 requests for today.",
        title="Quota Exceeded",
        type="https://api.example.com/errors/quota-exceeded",
        balance=0,
    )

Because it subclasses Starlette’s HTTPException, existing exception handlers and middleware treat it exactly like abort(). When instance is omitted, the rendered payload defaults it to the request path per the RFC 9457 recommendation. With API(problem_details=False) the legacy content-negotiated error format is used instead and only status_code/detail apply.

Parameters:

• status_code – The HTTP status code (e.g. 409).

• detail – Human-readable explanation of this occurrence; defaults to the status phrase.

• title – Short summary of the problem type; defaults to the status phrase.

• type – URI identifying the problem type (default about:blank).

• instance – URI for this specific occurrence; defaults to the request path when rendered.

• errors – Optional list of structured error dicts (the same shape validation failures use).

• headers – Optional dict of headers to attach to the error response.

• extensions – Any extra keyword arguments become top-level extension members of the problem payload.

Dependency injection raises the following at request time when a provider graph is misconfigured — cycles, illegal scopes, or unresolvable parameters. Catch the base DependencyError to cover all four. (Registration mistakes, such as a reserved name or a bad scope, raise plain ValueError instead.) See the Feature Tour for the dependency-injection guide.

exception responder.DependencyError

Base class for dependency-injection configuration/resolution errors.

exception responder.DependencyCycleError

A dependency depends on itself (directly or transitively).

exception responder.DependencyScopeError

An app-scoped dependency illegally depends on the request or a request-scoped dependency.

exception responder.DependencyResolutionError

A dependency parameter is neither the request nor a registered dependency.

The sessions extension raises SessionConfigError for an unsafe or contradictory configuration — for example secret_key="NOTASECRET", or sessions=True with no key set.

exception responder.ext.sessions.SessionConfigError

Raised for an unsafe or contradictory session configuration.

Type Aliases

Convenience aliases in responder.types for annotating your own handlers, hooks, and dependency providers:

from responder.types import Handler, Hook, Dependency

responder.types.Handler

A view handler. Receives (req, resp, **extras) and may return None or a body value (JSON scalar/dict/list/str/bytes/model, or a Flask-style (body, status[, headers]) tuple). Sync or async.

alias of Callable[[…], Any]

responder.types.Hook

A before/after-request hook, called (req, resp). Sync or async.

alias of Callable[[Request, Response], Any]

responder.types.Dependency

A dependency provider. Called with no arguments or the current Request; may be a function, coroutine, or (async) generator with teardown.

alias of Callable[[…], Any]