Implement MCP server authentication and authorization

Introduction

When you deploy an MCP tool server without verifying caller identity, any agent — legitimate or adversarial — can connect, enumerate your tool catalog, and serve poisoned tool descriptions before a single capability check runs. OAuth2 client credentials combined with scoped capability tokens close this gap by ensuring every incoming connection carries a signed JWT that names exactly which tools the caller may invoke. By the end of this lesson, you'll be able to implement a JWT-validating ASGI middleware layer that blocks unauthenticated requests before any MCP protocol message is processed, extract capability claims to enforce per-tool access control, and bind tokens to a specific server identity to prevent replay attacks against impersonating servers.

Key terminology

• Capability token: A JWT containing a capabilities claim that specifies which MCP tools the bearer may invoke and with what permission levels, enforcing per-tool access control at the server boundary.
• Server binding: The practice of restricting a token's validity to a single MCP server instance by embedding the server's identity in the JWT aud claim, preventing cross-server token reuse.
• Required-claims enforcement: The options={"require": [...]} setting passed to jwt.decode that rejects any token missing exp, aud, sub, or capabilities before claim extraction runs, so malformed tokens never reach the handler.
• Bearer token: The JWT carried in the Authorization: Bearer <token> header that the middleware extracts and validates before any MCP protocol message is processed.
• JWKS endpoint: A URL served by the authorization server that returns the current set of public keys used to verify JWT signatures, enabling key rotation without server redeployment.

Concepts

Why Authentication Must Run Before the MCP Protocol Layer

An MCP server exposes a tool catalog through protocol messages — tools/list is typically the first exchange after a connection is established. If authentication is deferred until after that handshake, an adversarial agent can enumerate available tools and receive poisoned descriptions before a single identity check runs. Placing validation in an ASGI middleware layer solves this by intercepting the raw HTTP request stream, meaning no MCP protocol message is ever parsed or dispatched unless the caller has already presented a valid signed token. The /health path is the only exemption, deliberately kept unauthenticated so infrastructure probes can reach it without credentials (see Code Walkthrough).

JWKS-Based Signature Validation and Audience Binding

Rather than distributing static public keys to each server, the middleware delegates key retrieval to a live JWKS endpoint via PyJWKClient. When a token arrives, get_signing_key_from_jwt extracts the key ID from the JWT header and fetches the matching public key from the authorization server's JWKS URL — with cache_keys=True so repeated calls reuse the cached key until rotation forces a refresh. This avoids the operational burden of re-deploying every MCP server when keys rotate.

The decoded payload is accepted only when jwt.decode confirms the aud claim matches self.server_id. This audience check is the concrete mechanism that prevents replay attacks: a JWT issued for server-A embeds "aud": "server-A", so presenting it to server-B (whose self.server_id is "server-B") raises jwt.InvalidAudienceError and the middleware returns 401 invalid_token. Required claims — exp, aud, sub, and capabilities — are enforced through options={"require": [...]}, so a token missing any of them is rejected without reaching claim extraction.

Propagating Capability Claims for Per-Tool Enforcement

Once the token passes validation, MCPAuthMiddleware attaches the resulting CapabilityClaims instance to request.state.capabilities. This pattern avoids re-parsing or re-validating the JWT on every subsequent call within the same request: the middleware runs once, and all downstream handlers read from the already-decoded, already-validated object.

Tool handlers use check_tool_capability to consult the tools dict inside CapabilityClaims. Each entry maps a tool name to a list of permission levels — for example, {"file_reader": ["read"], "file_writer": ["read", "write"]} — allowing handlers to enforce not just whether a tool is accessible but what operations the caller may perform. A capabilities attribute that is None (meaning the middleware was bypassed, which should never happen in production) triggers an immediate False return, ensuring the handler fails closed rather than open (see Code Walkthrough).

Code Walkthrough

Building on the JWKS signature validation and audience binding concepts above, the following code wires OAuth2 JWT validation and capability enforcement into a single ASGI middleware layer that protects every MCP connection before any protocol message reaches the server.

The MCPAuthMiddleware class intercepts each HTTP request, extracts the Bearer token, validates its signature against the authorization server's JWKS endpoint, and confirms the aud claim matches this server's registered identity. That audience check is the concrete mechanism that blocks a valid token issued for one MCP server from being replayed against an impersonating server. Decoded capability claims are attached to request.state so downstream tool handlers can enforce per-tool access without re-parsing the token.

Code snippetpython
1import jwt
2from jwt import PyJWKClient
3from dataclasses import dataclass, field
4from typing import Optional
5from starlette.middleware.base import BaseHTTPMiddleware
6from starlette.requests import Request
7from starlette.responses import JSONResponse
8
9@dataclass
10class CapabilityClaims:
11    client_id: str
12    server_id: str
13    tools: dict[str, list[str]] = field(default_factory=dict)
14    max_calls_per_minute: int = 60
15    expires_at: float = 0.0
16
17class MCPAuthMiddleware(BaseHTTPMiddleware):
18    def __init__(self, app, jwks_url: str, server_id: str):
19        super().__init__(app)
20        self.jwks_client = PyJWKClient(jwks_url, cache_keys=True)
21        self.server_id = server_id
22
23    async def dispatch(self, request: Request, call_next):
24        if request.url.path == "/health":
25            return await call_next(request)
26        auth_header = request.headers.get("Authorization")
27        if not auth_header or not auth_header.startswith("Bearer "):
28            return JSONResponse({"error": "missing_token"}, status_code=401)
29        token = auth_header.split(" ", 1)[1]
30        claims = self._validate_token(token)
31        if claims is None:
32            return JSONResponse({"error": "invalid_token"}, status_code=401)
33        request.state.capabilities = claims
34        return await call_next(request)
35
36    def _validate_token(self, token: str) -> Optional[CapabilityClaims]:
37        try:
38            signing_key = self.jwks_client.get_signing_key_from_jwt(token)
39            payload = jwt.decode(
40                token, signing_key.key,
41                algorithms=["RS256", "ES256"],
42                audience=self.server_id,
43                options={"require": ["exp", "aud", "sub", "capabilities"]},
44            )
45        except (jwt.ExpiredSignatureError, jwt.InvalidAudienceError,
46                jwt.DecodeError, jwt.MissingRequiredClaimError):
47            return None
48        return CapabilityClaims(
49            client_id=payload["sub"],
50            server_id=self.server_id,
51            tools=payload.get("capabilities", {}).get("tools", {}),
52            max_calls_per_minute=payload.get("rate_limit", 60),
53            expires_at=payload["exp"],
54        )

With the middleware registered, each tool handler enforces capability scope by reading the attached claims and refusing any tool the token does not list. The tools dict maps each permitted tool name to a list of permission levels such as ["read"] or ["read", "write"], so handlers can apply fine-grained rules beyond a simple allow/deny check.

Code snippetpython
1from starlette.requests import Request
2from starlette.responses import JSONResponse
3
4def check_tool_capability(request: Request, tool_name: str) -> bool:
5    capabilities = getattr(request.state, "capabilities", None)
6    if capabilities is None:
7        return False
8    return tool_name in capabilities.tools
9
10def handle_tool_call(request: Request, tool_name: str):
11    if not check_tool_capability(request, tool_name):
12        return JSONResponse(
13            {"error": "capability_denied", "tool": tool_name},
14            status_code=403,
15        )
16    return {"status": "ok", "tool": tool_name}

Verify by sending a connection request carrying a JWT whose aud claim does not match your server's registered identity and confirming the middleware returns 401 invalid_token before any tools/list response is generated.

Do's and Don'ts

Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.

Do's

1. ✓Do validate the aud claim against self.server_id in jwt.decode — passing audience=self.server_id as a parameter is the concrete mechanism that prevents a signed JWT issued for one MCP server from being replayed against an impersonating server; without it, the PyJWT library skips audience validation entirely and the cross-server replay attack succeeds silently.
2. ✓Do attach decoded CapabilityClaims to request.state.capabilities inside MCPAuthMiddleware.dispatch — storing the already-validated payload on request state lets every downstream handler call check_tool_capability without re-parsing or re-validating the JWT, and guarantees all access decisions are based on the same verified token.
3. ✓Do enumerate all four required claims — exp, aud, sub, and capabilities — in jwt.decode's options={"require": [...]} list — this causes MissingRequiredClaimError (caught and returned as 401 invalid_token) for any token that silently omits a capability scope or server audience, blocking unauthenticated requests before a single MCP protocol message is processed.

Don'ts

1. ✗Don't omit the audience parameter from jwt.decode — dropping it removes audience enforcement entirely, meaning a legitimate JWT issued for a different MCP server instance passes signature validation and grants full tool access to an impersonating server, which is the replay attack server_id binding is specifically designed to prevent.
2. ✗Don't push JWT validation logic into individual tool handlers instead of MCPAuthMiddleware.dispatch — if signature verification runs per-handler, unauthenticated requests can reach tools/list and receive a full tool catalog before any identity check executes, recreating the tool-enumeration vulnerability the ASGI middleware layer closes.
3. ✗Don't check only tool_name in capabilities.tools and ignore the associated permission list — each entry in the tools dict maps to a list of permission levels such as ["read"] or ["read", "write"]; skipping that list in handler logic discards the fine-grained per-tool access control the capability token encodes and silently upgrades read-only callers to write access.