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
    secret_key="change-me",       # session signing key
    allowed_hosts=["example.com"],
)

class responder.API(*, debug=False, title=None, version=None, description=None, terms_of_service=None, contact=None, license=None, openapi=None, openapi_route='/schema.yml', static_dir='static', static_route='/static', templates_dir='templates', auto_escape=True, secret_key='NOTASECRET', enable_hsts=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)

The primary web-service class.

Parameters:

• static_dir – The directory to use for static files. Will be created for you if it doesn’t already exist.

• templates_dir – The directory to use for templates. Will be created for you if it doesn’t already exist.

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

• enable_hsts – If True, send all responses to HTTPS URLs.

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

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

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_middleware(middleware_cls, **middleware_config)

Add ASGI middleware to the application.

Middleware wraps the entire application and can inspect or modify every request and response. Middleware is applied in reverse order — the last middleware added runs first.

Parameters:

• middleware_cls – A Starlette-compatible middleware class.

• middleware_config – Keyword arguments passed to the middleware constructor.

Usage:

from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
api.add_middleware(HTTPSRedirectMiddleware)

add_route(route=None, endpoint=None, *, default=False, static=True, check_existing=True, websocket=False, before_request=False, methods=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”. Also, it will become a default route.

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

after_request()

Register a function to run after every request.

Usage:

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

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"}

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)}

graphql(route='/graphql', *, schema)

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))

Parameters:

• route – The URL path for the GraphQL endpoint.

• schema – A Graphene schema instance.

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}

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.

on_event(event_type: str, **args)

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():
    ...

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, to test all known routes against.

redirect(resp, location, *, set_text=True, status_code=301)

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.

property requests

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

route(route=None, *, request_model=None, response_model=None, **options)

Decorator for creating new routes around function and class definitions.

Usage:

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

With Pydantic models for OpenAPI documentation:

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"],
            request_model=ItemIn, response_model=ItemOut)
async def create_item(req, resp):
    data = await req.media()
    resp.media = {"id": 1, **data}

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, **options)

Run the application with uvicorn.

If the PORT environment variable is set, requests will be served on that port automatically to all known hosts.

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.

• options – Additional keyword arguments to send to uvicorn.run().

session(base_url='http://;')

Testing HTTP client. Returns a Starlette TestClient instance, able to send HTTP requests to the Responder application.

Parameters:

base_url – The base URL for the test client.

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.

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).

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:

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

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

# JSON body
data = await req.media()

# Form data
form = await req.media("form")

# File uploads
files = await req.media("files")

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

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)

Returns True if the incoming Request accepts the given content_type.

property apparent_encoding

The apparent encoding, detected automatically. Must be awaited.

Uses chardet for detection if installed, otherwise falls 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

The cookies sent in the Request, as a dictionary.

property encoding

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

property full_url

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

property headers

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

property is_json

Returns True if the request content type is JSON.

property is_secure

True if the request was made over HTTPS.

async media(format: str | Callable = None)

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.

property method

The incoming HTTP method used for the request, lower-cased.

property mimetype

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

property params

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

property path_params: dict

The path parameters extracted from the URL route.

property session

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

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()

property text

The Request body, as unicode. Must be awaited.

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
resp.file("path/to/file.pdf")       # auto content-type
resp.stream_file("large/export.csv") # streamed

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

class responder.Response(req, *, formats)

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 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.

file(path, *, content_type=None)

Serve a file from disk as the response.

Parameters:

• path – Path to the file to serve.

• content_type – Optional MIME type override.

property ok

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

redirect(location, *, set_text=True, status_code=301)

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 – The HTTP status code (default 301).

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

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.

Usage:

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

sse(func, *args, **kwargs)

Set up Server-Sent Events streaming.

Usage:

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

Each yielded dict can have: data, event, id, retry. Yielding a string is treated as data.

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)

Stream a file without loading it entirely into memory.

Parameters:

• path – Path to the file.

• content_type – Optional MIME type override.

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

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.

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")

# 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.

Parameters:

f – The function to run.

Returns:

A concurrent.futures.Future for the result.

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.

Rate Limiter

In-memory token bucket rate limiter. Limits requests per client IP address 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, and Retry-After (when limited).

class responder.ext.ratelimit.RateLimiter(requests=100, period=60)

Token bucket rate limiter.

Usage:

from responder.ext.ratelimit import RateLimiter

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

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

Or use the shorthand:

limiter = RateLimiter(requests=100, period=60)
limiter.install(api)

check(req, resp)

Check rate limit. Sets 429 status if exceeded.

install(api)

Install as a before_request hook on the API.

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)