MCP’s 2025 authorization specification standardized how MCP servers authenticate remote clients, built entirely on OAuth 2.1 rather than a bespoke scheme. This matters because it means any client already capable of standard OAuth flows, which is most enterprise software, can authorize against an MCP server with no MCP-specific auth code, only standard OAuth libraries, and it means token lifecycle (issuance, expiry, refresh, revocation) inherits decades of hardened practice rather than being reinvented per-server.
Why OAuth 2.1, Not a Custom Scheme
Before this spec update, early MCP deployments often used ad hoc API keys or bespoke tokens, which worked for simple internal setups but did not interoperate with enterprise identity providers (Okta, Azure AD, internal SSO) or support delegated, scoped, revocable access. OAuth 2.1 (the 2020s consolidation of OAuth 2.0 best practices: mandatory PKCE, no implicit grant, stricter redirect URI matching) gives MCP a battle-tested authorization model that enterprise identity teams already understand and already have infrastructure for. It also means an MCP server does not need to implement its own user login, password storage, or session management, all of that is delegated to whatever identity provider the enterprise already trusts, and the MCP server’s only job is validating tokens that provider issues.
The Discovery-and-Authorize Flow, Including Dynamic Client Registration
An MCP client first discovers the server’s associated authorization server via a well-known metadata endpoint, then performs a standard OAuth 2.1 authorization code flow with PKCE, and finally attaches the resulting access token as a bearer token on every subsequent MCP request. For clients that have never connected to a given authorization server before, the flow can also include dynamic client registration (RFC 7591), letting a new client obtain its own client ID and secret automatically rather than requiring an administrator to manually pre-register every possible client application ahead of time, which matters in an ecosystem where any conforming agent framework might be the one connecting.
sequenceDiagram
participant Client
participant MCP as MCP Server
participant AS as Authorization Server
Client->>MCP: Request without token
MCP-->>Client: 401 + WWW-Authenticate (points to AS metadata)
Client->>AS: GET /.well-known/oauth-authorization-server
AS-->>Client: Metadata (authorize/token endpoints, registration endpoint)
opt Client not yet registered with this AS
Client->>AS: POST /register (dynamic client registration)
AS-->>Client: client_id (+ client_secret if confidential)
end
Client->>AS: Authorization request (PKCE code_challenge)
AS-->>Client: Authorization code (after user consent)
Client->>AS: Token request (code + code_verifier)
AS-->>Client: Access token (+ refresh token)
Client->>MCP: Request with Authorization: Bearer <token>
MCP-->>Client: 200 OK (tool result)
style Client fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style MCP fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style AS fill:#fff7ed,stroke:#f59e0b,color:#0F172AValidating Bearer Tokens on the Server
The MCP server’s job is not to run its own authorization server (in most enterprise setups, an existing identity provider fills that role), but to validate incoming bearer tokens on every request and return proper 401 responses when validation fails. Efficient validation of JWT-based tokens hinges on caching the identity provider’s public signing keys (its JWKS) rather than fetching them on every single request.
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import httpx, time
from jose import jwt
_jwks_cache: dict = {"keys": None, "fetched_at": 0}
JWKS_TTL_SECONDS = 3600
async def _get_jwks() -> dict:
"""Cache the IdP's public keys; refresh hourly, or immediately on a
signature failure with an unrecognized key id (handles key rotation)."""
if _jwks_cache["keys"] is None or time.time() - _jwks_cache["fetched_at"] > JWKS_TTL_SECONDS:
resp = await httpx.AsyncClient().get("https://idp.acmecorp.internal/.well-known/jwks.json")
_jwks_cache["keys"] = resp.json()
_jwks_cache["fetched_at"] = time.time()
return _jwks_cache["keys"]
class OAuthBearerMiddleware(BaseHTTPMiddleware):
"""Validates OAuth 2.1 bearer tokens against the enterprise IdP's JWKS."""
async def dispatch(self, request, call_next):
if request.url.path == "/health":
return await call_next(request)
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return self._unauthorized()
token = auth_header.removeprefix("Bearer ")
try:
claims = await self._validate_jwt(token)
except Exception:
return self._unauthorized()
request.state.subject = claims["sub"]
request.state.scopes = claims.get("scope", "").split()
return await call_next(request)
def _unauthorized(self):
return JSONResponse(
status_code=401,
content={"error": "invalid_token"},
headers={
"WWW-Authenticate": (
'Bearer realm="enterprise-mcp", '
'as_uri="https://idp.acmecorp.internal/.well-known/oauth-authorization-server"'
)
},
)
async def _validate_jwt(self, token: str) -> dict:
jwks = await _get_jwks()
header = jwt.get_unverified_header(token)
key = next((k for k in jwks["keys"] if k["kid"] == header["kid"]), None)
if key is None:
# Key rotation may have happened since our last cache refresh; force one retry.
_jwks_cache["keys"] = None
jwks = await _get_jwks()
key = next((k for k in jwks["keys"] if k["kid"] == header["kid"]), None)
claims = jwt.decode(
token, key, algorithms=["RS256"],
audience="enterprise-mcp-server", issuer="https://idp.acmecorp.internal",
)
return claims # exp, aud, and iss are all verified by jwt.decode itselfThe kid-miss retry above matters in practice: identity providers rotate their signing keys periodically, and a server holding a stale cached JWKS will otherwise reject every token signed with the new key until its next scheduled refresh, forcing a single retry against a fresh JWKS fetch specifically on a kid mismatch closes that gap without requiring the cache TTL to be set uncomfortably short for the common case.
Token Expiry and Refresh Across a Long-Lived Session
Access tokens are deliberately short-lived (commonly 15 minutes to an hour) specifically to limit the damage window if one leaks, refresh tokens (typically much longer-lived, sometimes with their own rotation policy) exist so the client does not need to re-prompt a human for credentials every time the access token expires. This split matters for how an MCP client is built: a client performing a long agent task spanning several access-token lifetimes needs to detect a 401 response mid-task, silently exchange its refresh token for a new access token, and retry the failed request, all without surfacing anything to the end user or interrupting the agent’s reasoning loop.
class OAuthAwareMCPClient:
"""Client-side: transparently refreshes an expired access token mid-session."""
async def call_tool(self, name: str, arguments: dict) -> dict:
try:
return await self._do_call(name, arguments, self.access_token)
except UnauthorizedError:
self.access_token = await self._refresh_access_token(self.refresh_token)
return await self._do_call(name, arguments, self.access_token) # Retry once with the new token
async def _refresh_access_token(self, refresh_token: str) -> str:
resp = await httpx.AsyncClient().post(TOKEN_ENDPOINT, data={
"grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": self.client_id,
})
resp.raise_for_status()
return resp.json()["access_token"]The server-side contract that makes this work correctly is strict: it must reject an expired token every time, with no leniency or grace window, since any leniency defeats the entire point of issuing short-lived tokens in the first place. The scope claim extracted by the middleware above is what the next lesson and Lesson 12 use to enforce per-tool authorization, a valid token proves who is calling, but scopes decide what they are allowed to call.