OAuth 2.0 Authorization Server framework for MCP servers. Issue and manage tokens that protect MCP tool access.
Pair with mcp-authflow-resource on the resource server side.
- Token storage with PostgreSQL and in-memory backends
- RFC 6749 standardized OAuth error responses
- RFC 7523
private_key_jwtclient authentication with algorithm allowlist and JTI replay protection (Redis or in-memory) - RFC 7636 PKCE verification (
S256+plain, with an opt-in S256-only policy) and input validation for the token endpoint - RFC 8628 Device Authorization Grant — sans-IO polling state machine and code generators
- Sliding-window rate limiting for token endpoints
- Input validation for client IDs and scopes
- CORS helpers with origin allowlisting
- Async-first design, built on Starlette
pip install mcp-authflow
# With PostgreSQL token storage (production)
pip install mcp-authflow[postgres]Build an OAuth authorization server that issues tokens for MCP clients:
import secrets
import time
from contextlib import asynccontextmanager
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
from mcp_authflow.rate_limiting import SlidingWindowRateLimiter
from mcp_authflow.responses import invalid_request, rate_limit_exceeded
from mcp_authflow.storage import MemoryTokenStorage
from mcp_authflow.validation import parse_scope_field, validate_client_id
# --- Setup ---
storage = MemoryTokenStorage() # Use PostgresTokenStorage for production
limiter = SlidingWindowRateLimiter(requests_per_window=60, window_seconds=3600)
# --- Token endpoint ---
async def token_endpoint(request: Request) -> JSONResponse:
form = await request.form()
client_id = str(form.get("client_id", ""))
# Rate limit per client
if not await limiter.is_allowed(client_id):
return rate_limit_exceeded(
"Too many requests",
retry_after=await limiter.get_retry_after(client_id),
)
# Validate client
if not validate_client_id(client_id):
return invalid_request("Invalid client_id format")
# Issue token
token = secrets.token_urlsafe(32)
scopes = parse_scope_field(form.get("scope"))
expires_at = int(time.time()) + 3600
await storage.store_token(
token=token,
client_id=client_id,
scopes=scopes.split(),
expires_at=expires_at,
resource=str(form.get("resource", "")),
)
return JSONResponse({
"access_token": token,
"token_type": "bearer",
"expires_in": 3600,
"scope": scopes,
})
# --- Introspection endpoint (called by resource servers) ---
async def introspect_endpoint(request: Request) -> JSONResponse:
form = await request.form()
token = str(form.get("token", ""))
token_data = await storage.load_token(token)
if not token_data or token_data["expires_at"] < time.time():
return JSONResponse({"active": False})
return JSONResponse({
"active": True,
"client_id": token_data["client_id"],
"scope": " ".join(token_data["scopes"]),
"exp": token_data["expires_at"],
"aud": token_data.get("resource", ""),
})
@asynccontextmanager
async def lifespan(app):
await storage.initialize()
yield
await storage.close()
app = Starlette(
routes=[
Route("/token", token_endpoint, methods=["POST"]),
Route("/introspect", introspect_endpoint, methods=["POST"]),
],
lifespan=lifespan,
)Run with: uvicorn myapp:app --port 8000
MCP Client (Claude, etc.)
|
1. Authorization request
|
v
+---------------------+
| Auth Server | <-- this package
| (mcp-authflow) |
| |
| /token | 2. Issues access token
| /introspect | 4. Validates token
+---------------------+
^
|
4. Token introspection (RFC 7662)
|
+---------------------+
| Resource Server | <-- mcp-authflow-resource
| (MCP tools) |
| |
| 3. Client calls |
| MCP tools with |
| Bearer token |
+---------------------+
- MCP client authenticates with the auth server
- Auth server issues an access token (stored in PostgreSQL or memory)
- Client calls MCP tools on the resource server with the Bearer token
- Resource server validates the token by calling the auth server's
/introspectendpoint
Abstract base class with two implementations:
from mcp_authflow.storage import MemoryTokenStorage, PostgresTokenStorage
# In-memory (development/testing)
storage = MemoryTokenStorage()
# PostgreSQL (production) -- requires `postgres` extra
storage = PostgresTokenStorage(database_url="postgresql://user:pass@host/db")
# Or reads DATABASE_URL env var if no argument provided
storage = PostgresTokenStorage()
await storage.initialize() # Open the connection pool (in-memory needs no setup)PostgresTokenStorage does not create or migrate its schema — it expects
the tables to already exist, so you stay in control of migrations. Apply this
DDL (e.g. via your migration tool) before first use:
CREATE TABLE IF NOT EXISTS mcp_access_tokens (
token TEXT PRIMARY KEY, -- SHA-256 digest of the access token, not the raw value
client_id TEXT NOT NULL,
scopes TEXT NOT NULL DEFAULT '',
resource TEXT,
expires_at TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
user_id INTEGER
);
-- Speeds up expiry checks on load and the `cleanup_expired_tokens` sweep
-- (`DELETE ... WHERE expires_at < now()`), which would otherwise seq-scan.
CREATE INDEX IF NOT EXISTS idx_mcp_access_tokens_expires_at
ON mcp_access_tokens (expires_at);
-- Only needed if you use the refresh-token methods.
CREATE TABLE IF NOT EXISTS mcp_refresh_tokens (
token TEXT PRIMARY KEY, -- SHA-256 digest of the refresh token, not the raw value
client_id TEXT NOT NULL,
scopes TEXT NOT NULL DEFAULT '',
resource TEXT,
expires_at TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
user_id INTEGER
);
CREATE INDEX IF NOT EXISTS idx_mcp_refresh_tokens_expires_at
ON mcp_refresh_tokens (expires_at);On an already-large, live table, build the index with CREATE INDEX CONCURRENTLY (run outside a transaction) so the migration does not take an
ACCESS EXCLUSIVE lock that blocks concurrent auth traffic:
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_mcp_access_tokens_expires_at
ON mcp_access_tokens (expires_at);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_mcp_refresh_tokens_expires_at
ON mcp_refresh_tokens (expires_at);The DDL above uses CREATE TABLE IF NOT EXISTS, which is a no-op when the table
already exists. That is the right behaviour for a fresh install, but it means
re-running the DDL after upgrading the library does not add any columns a
newer release introduced — Postgres silently keeps your existing table as-is,
and no error is raised. If a later version then references a column your table
is missing, backend queries fail at runtime with UndefinedColumnError.
To stay ahead of this:
-
Every release documents the schema it expects. All columns shown above have shipped since the DDL was first published, so the current minimum schema is simply the base tables above. When a future release adds, drops, or changes a column, this README and the CHANGELOG will call it out and ship a copy-paste
ALTER TABLErecipe next to the baseCREATE. -
Apply the upgrade DDL, don't just re-run
CREATE. Upgrade blocks useADD COLUMN IF NOT EXISTSso they are safe to run more than once and safe on a table that predates or already has the column. The template for such a block looks like:-- Upgrade to <version>: adds <column> to the token tables. ALTER TABLE mcp_access_tokens ADD COLUMN IF NOT EXISTS <column> <type>; ALTER TABLE mcp_refresh_tokens ADD COLUMN IF NOT EXISTS <column> <type>;
There are no such blocks yet — the schema has not changed since it was first published. This section is where they will appear when it does.
As a backstop, initialize() performs a lightweight information_schema check
on the token tables that already exist and raises a clear error naming any
required column that is missing, so schema drift fails fast at startup instead
of surfacing as a mid-request UndefinedColumnError. Tables you have not
created are left alone (the mcp_refresh_tokens table is only needed if you use
the refresh-token methods).
Tokens are hashed at rest: the token column holds the SHA-256 hex digest of
the token, never the raw secret, so a database compromise does not leak
replayable credentials. Hashing is internal — the store_token / load_token
API still takes and returns the raw token. If you are upgrading a deployment
that previously stored raw tokens, treat this as a breaking schema change: the
digest is 64 hex characters, so existing rows will no longer match on lookup.
Perform an expand-contract migration (rehash existing tokens, or expire and
reissue them) as part of the upgrade.
Storage interface:
| Method | Description |
|---|---|
store_token(token, client_id, scopes, expires_at, resource?, user_id?) |
Store an access token |
load_token(token) -> dict | None |
Look up a token |
delete_token(token) |
Revoke a token |
cleanup_expired_tokens() -> int |
Purge expired tokens, returns count |
get_token_count() -> int |
Count active tokens |
store_refresh_token(...) |
Store a refresh token (same interface) |
load_refresh_token(token) -> dict | None |
Look up a refresh token |
delete_refresh_token(token) |
Revoke a refresh token |
cleanup_expired_refresh_tokens() -> int |
Purge expired refresh tokens, returns count |
Token data returned by load_token():
{
"token": str,
"client_id": str,
"scopes": list[str],
"resource": str | None, # RFC 8707 resource binding
"expires_at": int, # Unix timestamp
"created_at": int, # Unix timestamp
"user_id": int | None,
}Standardized error helpers following RFC 6749 (and the device-flow /
registration extensions). Each returns a ready-to-send Starlette
JSONResponse:
from mcp_authflow.responses import (
oauth_error, # helper the others build on (default 400)
invalid_request, # 400 - Missing/invalid parameters
invalid_client, # 401 - Authentication failure
invalid_grant, # 400 - Expired/invalid code or token
invalid_scope, # 400 - Scope violation
unsupported_grant_type, # 400 - Unsupported grant_type (RFC 6749 §5.2)
access_denied, # 400 - User/AS denied the request
invalid_redirect_uri, # 400 - Bad redirect_uri (RFC 7591 §3.2.2)
authorization_pending, # 400 - Device flow: keep polling (RFC 8628 §3.5)
slow_down, # 400 or 429 - Device flow: poll slower
expired_token, # 400 - Device flow: device_code expired
pkce_required, # 400 - PKCE is required for this client
rate_limit_exceeded, # 429 - Too many requests
server_error, # 500 (or 502/504) - Internal error
backend_timeout, # 504 - Upstream timeout
backend_connection_error, # 502 - Upstream connection failure
backend_invalid_response, # 502 - Malformed upstream response
backend_oauth_error, # passthrough of an upstream OAuth error dict
)Each returns a Starlette JSONResponse with the appropriate status code and Cache-Control: no-store header.
from mcp_authflow.rate_limiting import SlidingWindowRateLimiter
limiter = SlidingWindowRateLimiter(
requests_per_window=60, # Max requests per window
window_seconds=3600, # Window duration (1 hour)
)
if not await limiter.is_allowed(client_id):
retry_after = await limiter.get_retry_after(client_id) # Seconds until next allowed requestfrom mcp_authflow.validation import validate_client_id, parse_scope_field
validate_client_id("my-client-123") # True (alphanumeric + hyphens/underscores)
validate_client_id("") # False
parse_scope_field("read write") # "read write"
parse_scope_field(["read", "write"]) # "read write"
parse_scope_field(None) # "read" (default)from mcp_authflow.cors import parse_allowed_origins, build_cors_headers
# Reads ALLOWED_MCP_ORIGINS env var (comma-separated)
origins = parse_allowed_origins()
# Returns CORS headers if request origin is in allowlist
headers = build_cors_headers(request, origins)| Env Variable | Description | Default |
|---|---|---|
DATABASE_URL |
PostgreSQL connection string (for PostgresTokenStorage) |
Required for postgres |
ALLOWED_MCP_ORIGINS |
Comma-separated allowed CORS origins | Empty (no CORS) |
MIT