Skip to content

Responses

RFC 6749 OAuth error response helpers. Each function returns a Starlette JSONResponse with the appropriate status code and a Cache-Control: no-store header.

responses

OAuth error response helpers for MCP Auth server.

This module provides standardized error response functions following OAuth 2.0 error response format (RFC 6749).

oauth_error 

oauth_error(
    error: str,
    description: str,
    status_code: int = 400,
    extra_headers: dict[str, str] | None = None,
) -> JSONResponse

Create a standard OAuth error response.

Parameters:

Name Type Description Default
error str

OAuth error code (e.g., "invalid_request", "server_error")

 required
description str

Human-readable error description

 required
status_code int

HTTP status code (default 400)

 400
extra_headers dict[str, str] | None

Additional headers to include (e.g., Retry-After)

 None

Returns:

Type Description
JSONResponse

JSONResponse with OAuth error format
Source code in mcp_authflow/responses.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def oauth_error(
    error: str,
    description: str,
    status_code: int = 400,
    extra_headers: dict[str, str] | None = None,
) -> JSONResponse:
    """Create a standard OAuth error response.

    Args:
        error: OAuth error code (e.g., "invalid_request", "server_error")
        description: Human-readable error description
        status_code: HTTP status code (default 400)
        extra_headers: Additional headers to include (e.g., Retry-After)

    Returns:
        JSONResponse with OAuth error format
    """
    headers = OAUTH_NO_CACHE_HEADERS.copy()
    if extra_headers:
        headers.update(extra_headers)

    return JSONResponse(
        {"error": error, "error_description": description},
        status_code=status_code,
        headers=headers,
    )

invalid_request 

invalid_request(description: str) -> JSONResponse

Create an invalid_request error response (400).

Use for: missing required parameters, invalid parameter format.

Source code in mcp_authflow/responses.py 
42
43
44
45
46
47
def invalid_request(description: str) -> JSONResponse:
    """Create an invalid_request error response (400).

    Use for: missing required parameters, invalid parameter format.
    """
    return oauth_error("invalid_request", description, 400)

invalid_client 

invalid_client(description: str) -> JSONResponse

Create an invalid_client error response (401).

Use for: client authentication failed, unknown client.

Source code in mcp_authflow/responses.py 
50
51
52
53
54
55
def invalid_client(description: str) -> JSONResponse:
    """Create an invalid_client error response (401).

    Use for: client authentication failed, unknown client.
    """
    return oauth_error("invalid_client", description, 401)

slow_down 

slow_down(
    description: str, retry_after: int | None = None
) -> JSONResponse

Create a slow_down error response (400, RFC 8628 §3.5).

Use for: rate limiting during device flow polling.

The status code is always 400, like the other device-flow polling errors — RFC 8628 §3.5 defines slow_down as a token-endpoint error response, so it uses the RFC 6749 §5.2 status. For a generic 429 on a non-polling endpoint use :func:rate_limit_exceeded instead.

Parameters:

Name Type Description Default
description str

Error description

 required
retry_after int | None

Optional retry-after value in seconds

 None
Source code in mcp_authflow/responses.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def slow_down(description: str, retry_after: int | None = None) -> JSONResponse:
    """Create a slow_down error response (400, RFC 8628 §3.5).

    Use for: rate limiting during device flow polling.

    The status code is always 400, like the other device-flow polling errors —
    RFC 8628 §3.5 defines ``slow_down`` as a token-endpoint error response, so
    it uses the RFC 6749 §5.2 status. For a generic 429 on a non-polling
    endpoint use :func:`rate_limit_exceeded` instead.

    Args:
        description: Error description
        retry_after: Optional retry-after value in seconds
    """
    extra_headers = {"Retry-After": str(retry_after)} if retry_after else None
    return oauth_error("slow_down", description, 400, extra_headers)

rate_limit_exceeded 

rate_limit_exceeded(
    description: str, retry_after: int | None = None
) -> JSONResponse

Create a rate limit exceeded error response (429).

Use for: too many requests to a generic endpoint (registration, introspection, etc.).

Emits the too_many_requests error code so it does not collide with the device-flow polling signal produced by :func:slow_down. RFC 8628 §3.5 reserves slow_down specifically for device-flow token polling, so a generic 429 must use a distinct code to avoid pushing clients into device-flow backoff.

Parameters:

Name Type Description Default
description str

Error description

 required
retry_after int | None

Optional retry-after value in seconds

 None
Source code in mcp_authflow/responses.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def rate_limit_exceeded(description: str, retry_after: int | None = None) -> JSONResponse:
    """Create a rate limit exceeded error response (429).

    Use for: too many requests to a generic endpoint (registration,
    introspection, etc.).

    Emits the ``too_many_requests`` error code so it does not collide with the
    device-flow polling signal produced by :func:`slow_down`. RFC 8628 §3.5
    reserves ``slow_down`` specifically for device-flow token polling, so a
    generic 429 must use a distinct code to avoid pushing clients into
    device-flow backoff.

    Args:
        description: Error description
        retry_after: Optional retry-after value in seconds
    """
    extra_headers = {"Retry-After": str(retry_after)} if retry_after else None
    return oauth_error("too_many_requests", description, 429, extra_headers)

server_error 

server_error(
    description: str, status_code: int = 500
) -> JSONResponse

Create a server_error response.

Use for: internal server errors, backend failures.

Parameters:

Name Type Description Default
description str

Error description

 required
status_code int

HTTP status code (500, 502, 504, etc.)

 500
Source code in mcp_authflow/responses.py 
 96
 97
 98
 99
100
101
102
103
104
105
def server_error(description: str, status_code: int = 500) -> JSONResponse:
    """Create a server_error response.

    Use for: internal server errors, backend failures.

    Args:
        description: Error description
        status_code: HTTP status code (500, 502, 504, etc.)
    """
    return oauth_error("server_error", description, status_code)

backend_timeout 

backend_timeout() -> JSONResponse

Create a backend timeout error response (504).

Source code in mcp_authflow/responses.py 
108
109
110
def backend_timeout() -> JSONResponse:
    """Create a backend timeout error response (504)."""
    return server_error("Backend timeout", 504)

backend_connection_error 

backend_connection_error() -> JSONResponse

Create a backend connection error response (502).

Source code in mcp_authflow/responses.py 
113
114
115
def backend_connection_error() -> JSONResponse:
    """Create a backend connection error response (502)."""
    return server_error("Backend connection error", 502)

backend_invalid_response 

backend_invalid_response() -> JSONResponse

Create an invalid backend response error (502).

Source code in mcp_authflow/responses.py 
118
119
120
def backend_invalid_response() -> JSONResponse:
    """Create an invalid backend response error (502)."""
    return server_error("Invalid response from backend", 502)

invalid_grant 

invalid_grant(description: str) -> JSONResponse

Create an invalid_grant error response (400).

Use for: invalid or expired authorization codes, refresh tokens, or device codes.

Source code in mcp_authflow/responses.py 
123
124
125
126
127
128
def invalid_grant(description: str) -> JSONResponse:
    """Create an invalid_grant error response (400).

    Use for: invalid or expired authorization codes, refresh tokens, or device codes.
    """
    return oauth_error("invalid_grant", description, 400)

invalid_scope 

invalid_scope(description: str) -> JSONResponse

Create an invalid_scope error response (400).

Use for: requested scope is invalid, unknown, or exceeds what was granted.

Source code in mcp_authflow/responses.py 
131
132
133
134
135
136
def invalid_scope(description: str) -> JSONResponse:
    """Create an invalid_scope error response (400).

    Use for: requested scope is invalid, unknown, or exceeds what was granted.
    """
    return oauth_error("invalid_scope", description, 400)

unsupported_grant_type 

unsupported_grant_type(description: str) -> JSONResponse

Create an unsupported_grant_type error response (400, RFC 6749 §5.2).

Use for: token endpoint received a grant_type the AS does not support.

Source code in mcp_authflow/responses.py 
139
140
141
142
143
144
def unsupported_grant_type(description: str) -> JSONResponse:
    """Create an unsupported_grant_type error response (400, RFC 6749 §5.2).

    Use for: token endpoint received a grant_type the AS does not support.
    """
    return oauth_error("unsupported_grant_type", description, 400)

access_denied 

access_denied(description: str) -> JSONResponse

Create an access_denied error response (400, RFC 6749 §4.1.2.1 / §5.2).

Use for: resource owner or AS denied the request (e.g., user declined consent at the /authorize step, or device flow authorize_device was called with action=deny).

Source code in mcp_authflow/responses.py 
147
148
149
150
151
152
153
154
def access_denied(description: str) -> JSONResponse:
    """Create an access_denied error response (400, RFC 6749 §4.1.2.1 / §5.2).

    Use for: resource owner or AS denied the request (e.g., user declined
    consent at the ``/authorize`` step, or device flow ``authorize_device``
    was called with action=deny).
    """
    return oauth_error("access_denied", description, 400)

invalid_redirect_uri 

invalid_redirect_uri(description: str) -> JSONResponse

Create an invalid_redirect_uri error response (400, RFC 7591 §3.2.2).

Use for: registration or authorization request supplied a redirect_uri that is not allowed for the client.

Source code in mcp_authflow/responses.py 
157
158
159
160
161
162
163
def invalid_redirect_uri(description: str) -> JSONResponse:
    """Create an invalid_redirect_uri error response (400, RFC 7591 §3.2.2).

    Use for: registration or authorization request supplied a redirect_uri
    that is not allowed for the client.
    """
    return oauth_error("invalid_redirect_uri", description, 400)

authorization_pending 

authorization_pending(
    description: str = "Authorization pending",
) -> JSONResponse

Create an authorization_pending error response (400, RFC 8628 §3.5).

Use for: device flow token polling while the user has not yet completed authorization. Clients should continue polling at the configured interval.

Source code in mcp_authflow/responses.py 
166
167
168
169
170
171
172
173
def authorization_pending(description: str = "Authorization pending") -> JSONResponse:
    """Create an authorization_pending error response (400, RFC 8628 §3.5).

    Use for: device flow token polling while the user has not yet completed
    authorization. Clients should continue polling at the configured
    ``interval``.
    """
    return oauth_error("authorization_pending", description, 400)

expired_token 

expired_token(
    description: str = "Device code has expired",
) -> JSONResponse

Create an expired_token error response (400, RFC 8628 §3.5).

Use for: device flow token polling after the device_code has expired.

Source code in mcp_authflow/responses.py 
176
177
178
179
180
181
def expired_token(description: str = "Device code has expired") -> JSONResponse:
    """Create an expired_token error response (400, RFC 8628 §3.5).

    Use for: device flow token polling after the ``device_code`` has expired.
    """
    return oauth_error("expired_token", description, 400)

pkce_required 

pkce_required(
    description: str = "PKCE is required for public clients",
) -> JSONResponse

Create an invalid_request error indicating PKCE is required (400).

Public clients (no client_secret) MUST use PKCE per OAuth 2.1 / RFC 9700. This returns the standard invalid_request OAuth error code with a description that calls out the missing PKCE binding.

Source code in mcp_authflow/responses.py 
184
185
186
187
188
189
190
191
192
193
def pkce_required(
    description: str = "PKCE is required for public clients",
) -> JSONResponse:
    """Create an invalid_request error indicating PKCE is required (400).

    Public clients (no client_secret) MUST use PKCE per OAuth 2.1 / RFC 9700.
    This returns the standard ``invalid_request`` OAuth error code with a
    description that calls out the missing PKCE binding.
    """
    return oauth_error("invalid_request", description, 400)

backend_oauth_error 

backend_oauth_error(
    error_dict: dict[str, str], status_code: int
) -> JSONResponse

Create a JSONResponse from an already-formatted OAuth error dict.

Use for: forwarding transformed backend errors that are already in {"error": "...", "error_description": "..."} format.

Parameters:

Name Type Description Default
error_dict dict[str, str]

Dict with "error" and "error_description" keys

 required
status_code int

HTTP status code from the backend response

 required

Returns:

Type Description
JSONResponse

JSONResponse with Cache-Control: no-store header
Source code in mcp_authflow/responses.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def backend_oauth_error(error_dict: dict[str, str], status_code: int) -> JSONResponse:
    """Create a JSONResponse from an already-formatted OAuth error dict.

    Use for: forwarding transformed backend errors that are already in
    {"error": "...", "error_description": "..."} format.

    Args:
        error_dict: Dict with "error" and "error_description" keys
        status_code: HTTP status code from the backend response

    Returns:
        JSONResponse with Cache-Control: no-store header
    """
    return JSONResponse(
        error_dict,
        status_code=status_code,
        headers=OAUTH_NO_CACHE_HEADERS.copy(),
    )