MCP Authorization: OAuth 2.1, CIMD and a Real Identity Provider

What You’ll Build

An MCP server on HTTP is reachable by anyone who can reach its port. This lesson puts OAuth in front of it. It assumes you know the HTTP transport from Streamable HTTP, Testing and the MCP Inspector. For the general ideas behind tokens, scopes, PKCE and audiences, without MCP, read OAuth and Token-Based Authorization first.

• Required MCP concepts: the 401 that starts discovery, Protected Resource Metadata, a token verifier that checks issuer, audience, expiry and scope, the SDK client running the whole flow with a Client ID Metadata Document, and the difference between CIMD and Dynamic Client Registration.
• Optional production infrastructure: token refresh, a real identity provider (Keycloak), and where DPoP stands in the SDK.

Everything runs in one container, with pip install mcp==2.3.0 httpx pyjwt cryptography on python:3.12-slim. The SDK’s OAuth client is built on httpx2, which mcp==2.3.0 installs itself. The Keycloak part adds a Keycloak container. No cloud account is involved.

Part 1: Required MCP Concepts

The lab: three small servers

The flow needs three parties. The MCP server (the resource server) holds the tools. The authorization server issues tokens. The client’s metadata document is a small JSON file served over HTTPS, because the SDK only accepts an HTTPS client_metadata_url. First, a self-signed certificate for https://localhost:

# make_cert.py - a self-signed TLS certificate for https://localhost (lab only)
import datetime
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import NameOID

key = ec.generate_private_key(ec.SECP256R1())
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
now = datetime.datetime.now(datetime.timezone.utc)
cert = (x509.CertificateBuilder().subject_name(name).issuer_name(name)
        .public_key(key.public_key()).serial_number(x509.random_serial_number())
        .not_valid_before(now).not_valid_after(now + datetime.timedelta(days=7))
        .add_extension(x509.SubjectAlternativeName([x509.DNSName("localhost")]), critical=False)
        .sign(key, hashes.SHA256()))
open("cimd.pem", "wb").write(cert.public_bytes(serialization.Encoding.PEM))
open("cimd.key", "wb").write(key.private_bytes(serialization.Encoding.PEM,
     serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
print("wrote cimd.pem and cimd.key for localhost, valid 7 days")

$ python make_cert.py
wrote cimd.pem and cimd.key for localhost, valid 7 days

The metadata document says who the client is and where it may be redirected:

# cimd_host.py - publish the client's metadata document over HTTPS on port 8443
import http.server
import json
import ssl

DOC = {
    "client_id": "https://localhost:8443/client.json",
    "client_name": "Lesson CLI",
    "redirect_uris": ["http://localhost:3000/callback"],
    "grant_types": ["authorization_code", "refresh_token"],
    "response_types": ["code"],
    "token_endpoint_auth_method": "none",
}


class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps(DOC).encode()
        self.send_response(200 if self.path == "/client.json" else 404)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = http.server.HTTPServer(("localhost", 8443), Handler)
tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
tls.load_cert_chain("cimd.pem", "cimd.key")
server.socket = tls.wrap_socket(server.socket, server_side=True)
server.serve_forever()

The authorization server is about 75 lines. It’s infrastructure for the lab, not the subject of the lesson: it auto-approves every request (there is no login page), signs tokens with an ES256 key it publishes at /jwks, accepts both CIMD and Dynamic Client Registration, and supports refresh tokens. Read it if you want to see what an authorization server does; otherwise copy it as is.

# auth_server.py - a minimal OAuth 2.1 authorization server for the lab (port 9000)
import base64, hashlib, json, os, secrets, time
import httpx, jwt, uvicorn
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec
from starlette.applications import Starlette
from starlette.responses import JSONResponse, RedirectResponse
from starlette.routing import Route

ISSUER, TTL = "http://localhost:9000", int(os.environ.get("ACCESS_TTL", "300"))
KEY = ec.generate_private_key(ec.SECP256R1())
open("as_key.pem", "wb").write(KEY.private_bytes(serialization.Encoding.PEM,
     serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
JWK = json.loads(jwt.algorithms.ECAlgorithm.to_jwk(KEY.public_key())) | {"kid": "lab-1", "alg": "ES256"}
clients, codes, refresh_tokens = {}, {}, {}


async def metadata(request):
    return JSONResponse({
        "issuer": ISSUER, "authorization_endpoint": f"{ISSUER}/authorize",
        "token_endpoint": f"{ISSUER}/token", "registration_endpoint": f"{ISSUER}/register",
        "jwks_uri": f"{ISSUER}/jwks", "response_types_supported": ["code"],
        "grant_types_supported": ["authorization_code", "refresh_token"],
        "code_challenge_methods_supported": ["S256"], "token_endpoint_auth_methods_supported": ["none"],
        "client_id_metadata_document_supported": os.environ.get("CIMD", "1") == "1"})


async def register(request):  # Dynamic Client Registration (RFC 7591)
    client_id = f"dcr-{len(clients) + 1}"
    clients[client_id] = (await request.json()) | {"client_id": client_id}
    print(f"AS: registered {client_id} for {clients[client_id]['client_name']!r} (DCR)")
    return JSONResponse(clients[client_id], status_code=201)


async def authorize(request):  # auto-approves: no login page in the lab
    q = dict(request.query_params)
    if q["client_id"].startswith("https://"):  # Client ID Metadata Document
        doc = httpx.get(q["client_id"], verify="cimd.pem").json()
        print(f"AS: fetched CIMD for {doc['client_name']!r}")
    else:
        doc = clients[q["client_id"]]
    if doc["client_id"] != q["client_id"] or q["redirect_uri"] not in doc["redirect_uris"]:
        return JSONResponse({"error": "invalid_request"}, status_code=400)
    code = secrets.token_urlsafe(16)
    codes[code] = q
    return RedirectResponse(f"{q['redirect_uri']}?code={code}&state={q['state']}", status_code=302)


def issue(client_id, scope, resource):
    now = int(time.time())
    claims = {"iss": ISSUER, "aud": resource, "client_id": client_id, "scope": scope, "iat": now, "exp": now + TTL}
    access = jwt.encode(claims, KEY, "ES256", headers={"kid": "lab-1"})
    refresh = secrets.token_urlsafe(16)
    refresh_tokens[refresh] = (client_id, scope, resource)
    return JSONResponse({"access_token": access, "token_type": "Bearer", "expires_in": TTL,
                         "refresh_token": refresh, "scope": scope})


async def token(request):
    form = await request.form()
    if form["grant_type"] == "refresh_token":
        client_id, scope, resource = refresh_tokens.pop(form["refresh_token"])  # one use: rotation
        print(f"AS: refresh_token grant for {client_id}, old refresh token revoked")
        return issue(client_id, scope, resource)
    q = codes.pop(form["code"])
    digest = hashlib.sha256(form["code_verifier"].encode()).digest()
    if base64.urlsafe_b64encode(digest).rstrip(b"=").decode() != q["code_challenge"]:
        return JSONResponse({"error": "invalid_grant"}, status_code=400)
    print(f"AS: issued token aud={q['resource']} scope={q.get('scope')!r}")
    return issue(q["client_id"], q.get("scope", ""), q["resource"])


routes = [Route("/.well-known/oauth-authorization-server", metadata), Route("/jwks", lambda r: JSONResponse({"keys": [JWK]})),
          Route("/register", register, methods=["POST"]), Route("/authorize", authorize),
          Route("/token", token, methods=["POST"])]
uvicorn.run(Starlette(routes=routes), host="localhost", port=9000, log_level="warning")

Start the CIMD host and the authorization server in the background:

python cimd_host.py &
python -u auth_server.py > as.log 2>&1 &

Protect the server

# protected_server.py - the calc server, now requiring an OAuth access token (port 8001)
import jwt
from mcp.server import MCPServer
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings

ISSUER, RESOURCE = "http://localhost:9000", "http://localhost:8001/mcp"
jwks = jwt.PyJWKClient(f"{ISSUER}/jwks")


class JWTVerifier:
    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            key = jwks.get_signing_key_from_jwt(token)
            c = jwt.decode(token, key, algorithms=["ES256"], issuer=ISSUER, options={"verify_aud": False})
        except jwt.PyJWTError:
            return None  # bad signature, wrong issuer or expired: 401
        return AccessToken(token=token, client_id=c["client_id"], scopes=c["scope"].split(),
                           expires_at=c["exp"], resource=c["aud"])


auth = AuthSettings(issuer_url=ISSUER, resource_server_url=RESOURCE,
                    required_scopes=["calc"], validate_token_resource=True)
mcp = MCPServer("calc", token_verifier=JWTVerifier(), auth=auth)


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="localhost", port=8001)

Two pieces do the work. JWTVerifier checks the token’s signature against the published key, its issuer and its expiry, and reports the audience it was issued for (aud) as resource. AuthSettings tells the SDK which authorization server to advertise, which scope every request needs, and, with validate_token_resource=True, to refuse tokens whose resource isn’t this server. Start it with python -u protected_server.py > rs.log 2>&1 &.

The 401 that starts everything

# probe.py - what does an MCP request without a token get back?
import httpx

headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
r = httpx.post("http://localhost:8001/mcp", headers=headers, json={})
print(r.status_code, r.reason_phrase)
print("WWW-Authenticate:", r.headers["www-authenticate"])
url = r.headers["www-authenticate"].split('resource_metadata="')[1].rstrip('"')
print(httpx.get(url).json())

$ python probe.py
401 Unauthorized
WWW-Authenticate: Bearer error="invalid_token", error_description="Authentication required", resource_metadata="http://localhost:8001/.well-known/oauth-protected-resource/mcp"
{'resource': 'http://localhost:8001/mcp', 'authorization_servers': ['http://localhost:9000'], 'scopes_supported': ['calc'], 'bearer_methods_supported': ['header']}

A request with no token fails, and the failure tells the client what to do next. The WWW-Authenticate header points at the server’s Protected Resource Metadata (RFC 9728). That document names the authorization server to use and the scope to ask for. A client needs only the MCP server’s URL; it finds everything else from this 401.

The client runs the whole flow

# memory_storage.py - keep tokens and client info in memory (a real client saves them)
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken


class MemoryStorage:
    def __init__(self):
        self.tokens: OAuthToken | None = None
        self.client_info: OAuthClientInformationFull | None = None

    async def get_tokens(self):
        return self.tokens

    async def set_tokens(self, tokens):
        self.tokens = tokens

    async def get_client_info(self):
        return self.client_info

    async def set_client_info(self, client_info):
        self.client_info = client_info

# oauth_client.py - log in with OAuth, then call the protected tool
import asyncio, sys
import httpx2
from mcp import Client
from mcp.client.auth import OAuthClientProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import AuthorizationCodeResult, OAuthClientMetadata
from memory_storage import MemoryStorage

SERVER = "http://localhost:8001/mcp"
redirect = {}


async def open_browser(url: str) -> None:  # stands in for the user's browser
    print("client: authorize at", url.split("?")[0])
    redirect["to"] = httpx2.get(url).headers["location"]


async def read_callback() -> AuthorizationCodeResult:
    query = dict(httpx2.URL(redirect["to"]).params)
    return AuthorizationCodeResult(code=query["code"], state=query["state"])


async def main(use_cimd: bool):
    meta = OAuthClientMetadata(client_name="Lesson CLI", redirect_uris=["http://localhost:3000/callback"],
                               grant_types=["authorization_code", "refresh_token"], token_endpoint_auth_method="none")
    auth = OAuthClientProvider(SERVER, meta, MemoryStorage(), open_browser, read_callback,
                               client_metadata_url="https://localhost:8443/client.json" if use_cimd else None)
    async with Client(streamable_http_client(SERVER, http_client=httpx2.AsyncClient(auth=auth))) as client:
        print("add ->", (await client.call_tool("add", {"a": 2, "b": 3})).structured_content)


if __name__ == "__main__":
    asyncio.run(main(use_cimd="--dcr" not in sys.argv))

$ python oauth_client.py && cat as.log
client: authorize at http://localhost:9000/authorize
add -> {'result': 5}
AS: fetched CIMD for 'Lesson CLI'
AS: issued token aud=http://localhost:8001/mcp scope='calc'

One call_tool triggered the whole OAuth flow inside OAuthClientProvider. The client got the 401, read the resource metadata, then the authorization server’s metadata. It sent the user to /authorize with a PKCE challenge, and swapped the code for a token. Then it retried the original request with Authorization: Bearer .... open_browser and read_callback stand in for the browser: a real client opens the URL and listens on the redirect URI. The authorization server’s log shows the CIMD step: it fetched the client’s metadata from https://localhost:8443/client.json instead of looking the client up in its own database.

Tokens that aren’t good enough

These tokens are signed with the authorization server’s real key, so only the claims differ.

# wrong_tokens.py - real tokens from the same issuer, but not good enough for this server
import time
import httpx, jwt

key = open("as_key.pem").read()  # the lab's signing key, to mint test tokens
now = int(time.time())
base = {"iss": "http://localhost:9000", "client_id": "test", "iat": now, "exp": now + 60}
cases = {
    "for another server": base | {"aud": "http://localhost:8002/mcp", "scope": "calc"},
    "missing scope calc": base | {"aud": "http://localhost:8001/mcp", "scope": "read"},
    "expired": base | {"aud": "http://localhost:8001/mcp", "scope": "calc", "exp": now - 5},
}
for label, claims in cases.items():
    token = jwt.encode(claims, key, "ES256", headers={"kid": "lab-1"})
    r = httpx.post("http://localhost:8001/mcp", json={}, headers={"Authorization": f"Bearer {token}",
                   "Content-Type": "application/json", "Accept": "application/json, text/event-stream"})
    print(f"{label}: {r.status_code}", r.headers["www-authenticate"].split(", resource_metadata")[0])

$ python wrong_tokens.py
for another server: 401 Bearer error="invalid_token", error_description="Authentication required"
missing scope calc: 403 Bearer error="insufficient_scope", error_description="Required scope: calc"
expired: 401 Bearer error="invalid_token", error_description="Authentication required"

The SDK sends two different answers. A token for another server and an expired token both get 401 invalid_token: the client has to get a new token. A token without the calc scope gets 403 insufficient_scope: the token is fine, it just isn’t enough, and the SDK client responds by asking for the missing scope (a step-up). The audience check is what stops one MCP server from replaying a token it received to a different server; it’s the reason RFC 8707 resource indicators are part of MCP authorization.

CIMD versus Dynamic Client Registration

An authorization server has to know a client before it issues a token. MCP clients can introduce themselves in two ways. Run the same client with --dcr, which leaves out client_metadata_url:

$ python oauth_client.py --dcr && tail -2 as.log
client: authorize at http://localhost:9000/authorize
add -> {'result': 5}
AS: registered dcr-1 for 'Lesson CLI' (DCR)
AS: issued token aud=http://localhost:8001/mcp scope='calc'

With Dynamic Client Registration (RFC 7591), the client POSTs its metadata to /register and gets a new client_id (dcr-1). The authorization server now stores a record for this client, and for every other client that ever registers. With CIMD, the client_id is the URL of the metadata document. There is nothing to store, and the client’s owner can update its name or redirect URIs in one place. That is why the 2026-07-28 specification prefers CIMD.

The SDK picks between them for you. Restart the authorization server with CIMD=0 so its metadata says client_id_metadata_document_supported: false, and run the CIMD client unchanged:

$ python oauth_client.py && cat as.log
client: authorize at http://localhost:9000/authorize
add -> {'result': 5}
AS: registered dcr-1 for 'Lesson CLI' (DCR)
AS: issued token aud=http://localhost:8001/mcp scope='calc'

The client was given a metadata URL, but the server doesn’t support CIMD, so the SDK fell back to registering. You’ll see this with identity providers that predate CIMD, as the Keycloak section below shows.

Part 2: Optional Production Infrastructure

Token refresh

Short-lived access tokens limit the damage a leaked token can do, but nobody wants to log in every five minutes. Restart the authorization server with ACCESS_TTL=2 (2-second tokens) and keep the same session going:

# refresh_client.py - outlive a 2-second access token without logging in again
import asyncio
import httpx2
from mcp import Client
from mcp.client.auth import OAuthClientProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientMetadata
from memory_storage import MemoryStorage
from oauth_client import SERVER, open_browser, read_callback


async def main():
    meta = OAuthClientMetadata(client_name="Lesson CLI", redirect_uris=["http://localhost:3000/callback"],
                               grant_types=["authorization_code", "refresh_token"], token_endpoint_auth_method="none")
    storage = MemoryStorage()
    auth = OAuthClientProvider(SERVER, meta, storage, open_browser, read_callback,
                               client_metadata_url="https://localhost:8443/client.json")
    async with Client(streamable_http_client(SERVER, http_client=httpx2.AsyncClient(auth=auth))) as client:
        seen = set()
        for n in range(2):
            result = await client.call_tool("add", {"a": n, "b": 3})
            seen.add(storage.tokens.refresh_token)
            print(f"call {n + 1}: add -> {result.structured_content}, refresh tokens seen: {len(seen)}")
            await asyncio.sleep(3)  # longer than the 2-second access token


asyncio.run(main())

$ python refresh_client.py && cat as.log
client: authorize at http://localhost:9000/authorize
call 1: add -> {'result': 3}, refresh tokens seen: 1
call 2: add -> {'result': 4}, refresh tokens seen: 2
AS: fetched CIMD for 'Lesson CLI'
AS: issued token aud=http://localhost:8001/mcp scope='calc'
AS: refresh_token grant for https://localhost:8443/client.json, old refresh token revoked

The second call came three seconds after the first, after the access token expired. The client didn’t log in again: before sending the request it saw the token had expired and swapped its refresh token for a new pair. The lab server rotates refresh tokens, revoking the old one as it issues the new one, which is why the client saw two different refresh tokens. A stolen refresh token that has already been used is then worthless, and a reuse attempt is a signal worth alerting on.

A real identity provider: Keycloak

In production the authorization server is your identity provider: Keycloak, Okta, Entra ID, Auth0 and so on. Here is the same server trusting Keycloak 26.8.1, run locally on a Docker network:

docker network create mcp-lab
docker run -d --name keycloak --network mcp-lab -e KC_BOOTSTRAP_ADMIN_USERNAME=admin \
  -e KC_BOOTSTRAP_ADMIN_PASSWORD=admin-lab-only quay.io/keycloak/keycloak:26.8.1 start-dev

Run the next scripts in a container on the same network. Run kc_setup.py (in the collapsible block below) once first: it creates the mcp realm these scripts query, and without it they fail with KeyError: 'issuer'. Then ask Keycloak what it supports:

# kc_metadata.py - what does Keycloak advertise to an MCP client?
import httpx

meta = httpx.get("http://keycloak:8080/realms/mcp/.well-known/oauth-authorization-server").json()
print("issuer:", meta["issuer"])
print("registration_endpoint (DCR):", meta.get("registration_endpoint"))
print("client_id_metadata_document_supported (CIMD):", meta.get("client_id_metadata_document_supported"))
print("PKCE methods:", meta.get("code_challenge_methods_supported"))
print("DPoP algs:", meta.get("dpop_signing_alg_values_supported"))

$ python kc_metadata.py
issuer: http://keycloak:8080/realms/mcp
registration_endpoint (DCR): http://keycloak:8080/realms/mcp/clients-registrations/openid-connect
client_id_metadata_document_supported (CIMD): None
PKCE methods: ['plain', 'S256']
DPoP algs: ['PS384', 'RS384', 'EdDSA', 'ES384', 'ES256', 'RS256', 'ES512', 'PS256', 'PS512', 'RS512']

Keycloak supports PKCE, DPoP and Dynamic Client Registration, but doesn’t advertise CIMD. An MCP client pointed at it falls back to DCR, exactly as in the CIMD=0 run above. And DCR with an identity provider is rarely open:

# kc_dcr.py - can an unknown MCP client register itself with Keycloak (DCR)?
import httpx

r = httpx.post("http://keycloak:8080/realms/mcp/clients-registrations/openid-connect", json={
    "client_name": "Lesson CLI", "redirect_uris": ["http://localhost:3000/callback"],
    "grant_types": ["authorization_code"], "token_endpoint_auth_method": "none"})
print(r.status_code, r.json())

$ python kc_dcr.py
403 {'error': 'insufficient_scope', 'error_description': "Policy 'Trusted Hosts' rejected request to client-registration service. Details: Host not trusted."}

Keycloak’s default policy only accepts anonymous registrations from trusted hosts. That is the right default, because otherwise anyone could create clients in your realm. It’s also the problem CIMD solves: an MCP client from an unknown vendor can’t register itself. With an identity provider like this, you register MCP clients ahead of time.

This script, the setup an administrator would do in the admin console, creates a realm, a calc scope that adds the MCP server’s URL as the token audience, and a confidential client for a service.

# kc_setup.py - create a realm, a client scope "calc" and an audience for the MCP server
import httpx

KC = "http://keycloak:8080"
token = httpx.post(f"{KC}/realms/master/protocol/openid-connect/token", data={
    "grant_type": "password", "client_id": "admin-cli",
    "username": "admin", "password": "admin-lab-only"}).json()["access_token"]
api = httpx.Client(base_url=f"{KC}/admin/realms", headers={"Authorization": f"Bearer {token}"})

api.post("", json={"realm": "mcp", "enabled": True})
api.post("/mcp/client-scopes", json={
    "name": "calc", "protocol": "openid-connect",
    "protocolMappers": [{"name": "aud", "protocol": "openid-connect",
                         "protocolMapper": "oidc-audience-mapper",
                         "config": {"included.custom.audience": "http://mcp-server:8001/mcp",
                                    "access.token.claim": "true"}}]})
api.post("/mcp/clients", json={
    "clientId": "billing-agent", "secret": "lab-only-secret", "publicClient": False,
    "serviceAccountsEnabled": True, "standardFlowEnabled": False,
    "optionalClientScopes": ["calc"]})
print("realm mcp:", api.get("/mcp").json()["enabled"])
print("clients:", [c["clientId"] for c in api.get("/mcp/clients").json() if c["clientId"] == "billing-agent"])

$ python kc_setup.py
realm mcp: True
clients: ['billing-agent']

The MCP server changes in three places: the issuer, the key URL, and the audience check, which now happens in the verifier.

# kc_server.py - the calc server, trusting tokens from Keycloak (port 8001)
import jwt
from mcp.server import MCPServer
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings

ISSUER, RESOURCE = "http://keycloak:8080/realms/mcp", "http://mcp-server:8001/mcp"
jwks = jwt.PyJWKClient(f"{ISSUER}/protocol/openid-connect/certs")


class KeycloakVerifier:
    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            key = jwks.get_signing_key_from_jwt(token)
            c = jwt.decode(token, key, algorithms=["RS256"], issuer=ISSUER, audience=RESOURCE)
        except jwt.PyJWTError as e:
            print("rejected:", type(e).__name__)
            return None
        return AccessToken(token=token, client_id=c["azp"], scopes=c["scope"].split(), expires_at=c["exp"])


auth = AuthSettings(issuer_url=ISSUER, resource_server_url=RESOURCE,
                    required_scopes=["calc"], validate_token_resource=False)  # the verifier checks aud
mcp = MCPServer("calc", token_verifier=KeycloakVerifier(), auth=auth)


@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8001)

Run it in a container named mcp-server on the mcp-lab network. The client is a service with no user, so it uses the client-credentials grant from the SDK’s extensions:

# kc_client.py - a service (no user) gets a Keycloak token and calls the tool
import asyncio
import httpx2
from mcp import Client
from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider
from mcp.client.streamable_http import streamable_http_client
from memory_storage import MemoryStorage

SERVER = "http://mcp-server:8001/mcp"


async def main():
    storage = MemoryStorage()
    auth = ClientCredentialsOAuthProvider(SERVER, storage, client_id="billing-agent",
                                          client_secret="lab-only-secret", scope="calc",
                                          issuer="http://keycloak:8080/realms/mcp")
    async with Client(streamable_http_client(SERVER, http_client=httpx2.AsyncClient(auth=auth))) as client:
        print("add ->", (await client.call_tool("add", {"a": 2, "b": 3})).structured_content)
    print("token scope:", storage.tokens.scope)


asyncio.run(main())

$ python kc_client.py
add -> {'result': 5}
token scope: calc

Same 401, same discovery, but the token came from Keycloak and was checked against Keycloak’s published RS256 keys. issuer= pins the authorization server the secret may be sent to: without it, a malicious MCP server could advertise its own authorization server and collect your client secret. The SDK warns when it’s missing. Here the audience comes from the calc scope’s audience mapper, so the verifier checks aud itself with PyJWT and validate_token_resource is False.

DPoP: not in the SDK yet

A bearer token works for whoever holds it. DPoP (RFC 9449) binds a token to a key the client holds, so a stolen token is useless without the key. OAuth and Token-Based Authorization shows a DPoP proof being made and checked. Here is where mcp 2.3.0 stands:

# dpop_support.py - where does mcp 2.3.0 mention DPoP?
import pathlib
import mcp

root = pathlib.Path(mcp.__file__).parent
for f in sorted(root.rglob("*.py")):
    for n, line in enumerate(f.read_text().splitlines(), 1):
        if "dpop" in line.lower():
            print(f"{f.relative_to(root)}:{n}: {line.strip()}")

$ python dpop_support.py
shared/auth.py:256: dpop_signing_alg_values_supported: list[str] | None = None
shared/auth.py:257: # dpop_bound_access_tokens_required default is False, but omitted here for clarity
shared/auth.py:258: dpop_bound_access_tokens_required: bool | None = None

Those three lines are the only mentions of DPoP in the whole package: two fields of authorization server metadata that the SDK parses and never uses. Its client sends no proof and its server checks none. Keycloak advertised DPoP above, but an MCP client built on this SDK can’t use it. Until it can, rely on what this lesson already enforces: short-lived tokens, refresh rotation, audience and scope checks, and TLS on every hop. The production lesson, Running MCP in Production, covers key rotation and the other operational controls.

Practice

1. Call the server by another name. Point the client at http://127.0.0.1:8001/mcp, the same server reached by IP. What happens, and why is that the safe outcome?

# p23_1.py - the client checks that the metadata really describes this server
import asyncio
import httpx2
from mcp import Client
from mcp.client.auth import OAuthClientProvider
from mcp.client.auth.exceptions import OAuthFlowError
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientMetadata
from memory_storage import MemoryStorage
from oauth_client import open_browser, read_callback

SERVER = "http://127.0.0.1:8001/mcp"  # same server, but not the URL its metadata names


async def main():
    meta = OAuthClientMetadata(client_name="Lesson CLI", redirect_uris=["http://localhost:3000/callback"])
    auth = OAuthClientProvider(SERVER, meta, MemoryStorage(), open_browser, read_callback,
                               client_metadata_url="https://localhost:8443/client.json")
    try:
        async with Client(streamable_http_client(SERVER, http_client=httpx2.AsyncClient(auth=auth))) as client:
            await client.call_tool("add", {"a": 2, "b": 3})
    except* OAuthFlowError as group:
        print("refused:", group.exceptions[0])


asyncio.run(main())

$ python p23_1.py 2>/dev/null
refused: Protected resource http://localhost:8001/mcp does not match expected http://127.0.0.1:8001/mcp

2. Forge a token. Sign a token with the right issuer, audience, scope and even the right kid, but with your own key. Send it to protected_server.py.

# p23_3.py - perfect claims, but signed with a key the authorization server never published
import time
import httpx, jwt
from cryptography.hazmat.primitives.asymmetric import ec

forger = ec.generate_private_key(ec.SECP256R1())
now = int(time.time())
claims = {"iss": "http://localhost:9000", "aud": "http://localhost:8001/mcp", "client_id": "forged",
          "scope": "calc", "iat": now, "exp": now + 60}
token = jwt.encode(claims, forger, "ES256", headers={"kid": "lab-1"})  # even the kid matches
r = httpx.post("http://localhost:8001/mcp", json={}, headers={"Authorization": f"Bearer {token}",
               "Content-Type": "application/json", "Accept": "application/json, text/event-stream"})
print("forged token:", r.status_code, r.headers["www-authenticate"].split(",")[0])

$ python p23_3.py
forged token: 401 Bearer error="invalid_token"

3. Leave out the scope at Keycloak. Get one client-credentials token with scope=calc and one without, and send each to kc_server.py in a well-formed server/discover request.

# p23_2.py - a valid Keycloak token, but without the calc scope (and so without our audience)
import httpx, jwt

KC = "http://keycloak:8080/realms/mcp/protocol/openid-connect"
META = {"io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": {}}
DISCOVER = {"jsonrpc": "2.0", "id": 1, "method": "server/discover", "params": {"_meta": META}}
HEADERS = {"Accept": "application/json, text/event-stream", "MCP-Protocol-Version": "2026-07-28",
           "Mcp-Method": "server/discover"}

for scope in ("calc", None):
    form = {"grant_type": "client_credentials", "client_id": "billing-agent",
            "client_secret": "lab-only-secret"} | ({"scope": scope} if scope else {})
    token = httpx.post(f"{KC}/token", data=form).json()["access_token"]
    aud = jwt.decode(token, options={"verify_signature": False}).get("aud")  # just to look inside
    r = httpx.post("http://mcp-server:8001/mcp", json=DISCOVER,
                   headers=HEADERS | {"Authorization": f"Bearer {token}"})
    print(f"scope={scope}: aud={aud} -> HTTP {r.status_code}")

$ python p23_2.py
scope=calc: aud=http://mcp-server:8001/mcp -> HTTP 200
scope=None: aud=None -> HTTP 401