A stdio server is started by its client on the same machine. A remote MCP server listens on a network address, and anyone who can reach it can send requests, so it must decide who may call which tool. You will serve a docs server over Streamable HTTP as an OAuth 2.1 resource server: anyone with docs:read may call search_docs, and only a token with docs:write may call publish_page. Every check follows the MCP authorization specification, 2026-07-28. MCP authorization
When you need this. Once a docs server leaves your laptop and sits behind a URL, anyone who can reach it can send requests. A docs service is a typical case: every team’s agent may search it, but only the publishing pipeline may change it. Scoped tokens let one server answer both callers and refuse a write from a client that was only allowed to read.
Full output: run-output.txt.
Authorization has two boundaries: the endpoint requires docs:read, then ScopeChallenge checks the scope of the tool named in the JSON-RPC body. publish_page independently requires docs:write inside the tool. A valid search token therefore cannot become a write token by changing the requested tool name.
1. Set up the project
Download the complete lab ZIP, which includes the Python files, fixtures, tests and README. Extract it before running the commands below. The SDK installs everything else this lesson uses.
unzip build-a-remote-mcp-service-with-authentication-and-authorization.zip
cd mcp-remote-auth
python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements-lock.txt2. Run a test authorization server
Before any code, here is who does what. MCP authorization has three roles. The MCP server is a resource server that checks the access token on every request. The MCP client is an OAuth client that gets tokens and sends them. The authorization server issues tokens; it can be a separate product, such as your identity provider.
For a repeatable run, token_issuer.py stands in for the authorization server. It serves its metadata (RFC 8414), a JSON Web Key Set, and a token endpoint for the client_credentials grant, in which a program proves who it is with a client ID and secret. Each token is a JWT signed with a key made at start-up, and its audience is the resource the client asked for.
"""A test token issuer for this lesson. It is NOT a full OAuth 2.1 authorization server.
Usage: ISSUER_PORT=9001 CLIENTS_JSON='{...}' python token_issuer.py
It serves authorization server metadata (RFC 8414), a JSON Web Key Set and a token
endpoint for the client_credentials grant only. Every token is a JWT signed with an
ES256 key made at start-up and bound to the `resource` the client names (RFC 8707).
There is no user sign-in, consent, authorization code or PKCE.
"""
import json
import os
import secrets
import time
import jwt
import uvicorn
from cryptography.hazmat.primitives.asymmetric import ec
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
PORT = int(os.environ["ISSUER_PORT"])
ISSUER = f"http://127.0.0.1:{PORT}"
CLIENTS = json.loads(os.environ["CLIENTS_JSON"]) # {client_id: {"secret", "scopes", "resources", "lifetime"?}}
KEY = ec.generate_private_key(ec.SECP256R1())
KID = secrets.token_hex(8)
PUBLIC_JWK = {**json.loads(jwt.algorithms.ECAlgorithm.to_jwk(KEY.public_key())), "kid": KID, "use": "sig", "alg": "ES256"}
def oauth_error(status, error, description):
return JSONResponse({"error": error, "error_description": description}, status_code=status)
async def metadata(request):
return JSONResponse({
"issuer": ISSUER,
"authorization_endpoint": f"{ISSUER}/authorize", # mcp 2.3.0's metadata model requires it; it only refuses
"response_types_supported": [],
"token_endpoint": f"{ISSUER}/token",
"jwks_uri": f"{ISSUER}/jwks.json",
"grant_types_supported": ["client_credentials"],
"token_endpoint_auth_methods_supported": ["client_secret_post"],
"scopes_supported": ["docs:read", "docs:write"],
})
async def authorize(request):
return oauth_error(400, "unsupported_response_type", "this test issuer has no user sign-in; use the token endpoint")
async def jwks(request):
return JSONResponse({"keys": [PUBLIC_JWK]})
async def token(request):
form = await request.form()
client = CLIENTS.get(form.get("client_id", ""))
if form.get("grant_type") != "client_credentials":
return oauth_error(400, "unsupported_grant_type", "only client_credentials is supported")
if client is None or not secrets.compare_digest(client["secret"], form.get("client_secret", "")):
return oauth_error(401, "invalid_client", "unknown client or wrong secret")
if form.get("resource") not in client["resources"]:
return oauth_error(400, "invalid_target", "the resource is missing or not allowed for this client")
scopes = form.get("scope", "").split()
if not scopes or not set(scopes) <= set(client["scopes"]):
return oauth_error(400, "invalid_scope", "a requested scope is not allowed for this client")
now, lifetime = int(time.time()), client.get("lifetime", 300)
claims = {"iss": ISSUER, "sub": form["client_id"], "client_id": form["client_id"], "aud": form["resource"],
"scope": " ".join(scopes), "iat": now, "exp": now + lifetime, "jti": secrets.token_hex(8)}
return JSONResponse({"access_token": jwt.encode(claims, KEY, algorithm="ES256", headers={"kid": KID}),
"token_type": "Bearer", "expires_in": lifetime, "scope": " ".join(scopes)})
app = Starlette(routes=[Route("/.well-known/oauth-authorization-server", metadata), Route("/authorize", authorize),
Route("/jwks.json", jwks), Route("/token", token, methods=["POST"])])
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=PORT, log_level="warning")Note: this issuer is for tests only. It has no user sign-in, consent, authorization code or PKCE, which the specification requires for clients that act for a user, and it serves plain HTTP, while all authorization server endpoints must use HTTPS. Use an established authorization server in production. MCP authorization: security considerations
A token request without a resource. The test issuer returns invalid_target. Include the MCP resource URI in the token request so the token has the intended audience.
A token request for a resource or scope the client was not given. The issuer returns invalid_target or invalid_scope. Grant only the intended resource and scopes in the authorization server; changing an MCP tool description cannot grant them.
3. Make the MCP server a resource server
"""A remote docs server over Streamable HTTP that accepts only access tokens issued for it.
Usage: SERVICE_PORT=9002 ISSUER_URL=http://127.0.0.1:9001 python docs_service.py
The server is an OAuth 2.1 resource server. It publishes Protected Resource Metadata
(RFC 9728), checks each bearer token's signature, issuer, audience and expiry, needs
docs:read for any MCP request and docs:write for publish_page. It listens on
127.0.0.1 only and refuses a Host or Origin header that is not its own.
"""
import json
import os
from typing import TypedDict
import jwt
import uvicorn
from starlette.responses import JSONResponse
from mcp.server import MCPServer
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings
from mcp.server.mcpserver.exceptions import ToolError
from mcp.server.transport_security import TransportSecuritySettings
PORT = int(os.environ["SERVICE_PORT"])
ISSUER_URL = os.environ["ISSUER_URL"]
RESOURCE = f"http://127.0.0.1:{PORT}/mcp" # the canonical server URI; tokens must name it as their audience
METADATA_URL = f"http://127.0.0.1:{PORT}/.well-known/oauth-protected-resource/mcp"
TOOL_SCOPES = {"publish_page": "docs:write"} # extra scope per tool; docs:read is needed for everything
MAX_BODY_BYTES = 1_000_000
PAGES = {"quickstart": "Start a worker with taskq worker emails.", "retries": "A failed job is retried 3 times."}
class JwtVerifier:
"""Verify ES256 JWTs from the issuer's key set; refuse any token not issued for this server."""
def __init__(self, issuer, audience):
self.issuer, self.audience = issuer, audience
self.keys = jwt.PyJWKClient(f"{issuer}/jwks.json")
async def verify_token(self, token):
try:
key = self.keys.get_signing_key_from_jwt(token)
claims = jwt.decode(token, key.key, algorithms=["ES256"], audience=self.audience, issuer=self.issuer,
options={"require": ["exp", "iat", "aud", "iss", "sub", "client_id", "scope"]})
except jwt.PyJWTError:
return None
# A signed token can still carry a claim of the wrong type: treat it as invalid, not as a crash.
if not (isinstance(claims["client_id"], str) and claims["client_id"] and isinstance(claims["scope"], str)):
return None
return AccessToken(token=token, client_id=claims["client_id"], scopes=claims["scope"].split(),
expires_at=claims["exp"], resource=claims["aud"], subject=claims["sub"])
verifier = JwtVerifier(ISSUER_URL, RESOURCE)
server = MCPServer("docs", version="1.0.0", token_verifier=verifier,
auth=AuthSettings(issuer_url=ISSUER_URL, resource_server_url=RESOURCE,
required_scopes=["docs:read"], validate_token_resource=True))
class Page(TypedDict):
slug: str
text: str
caller: str # the client_id from the token, so each answer shows who asked
@server.tool()
def search_docs(query: str) -> list[Page]:
"""Return the docs pages that contain the query. Needs docs:read."""
caller = get_access_token().client_id
return [{"slug": s, "text": t, "caller": caller} for s, t in PAGES.items() if query.lower() in t.lower()]
@server.tool()
def publish_page(slug: str, text: str) -> Page:
"""Create or replace one docs page. Needs docs:write."""
token = get_access_token()
if "docs:write" not in token.scopes: # second check: never rely on the HTTP layer alone
raise ToolError("The token lacks docs:write.")
PAGES[slug] = text
return {"slug": slug, "text": text, "caller": token.client_id}
def refuse_duplicate_keys(pairs):
if len({key for key, _ in pairs}) != len(pairs): # two parsers could keep different copies of a repeated key
raise ValueError("duplicate key")
return dict(pairs)
async def read_body(receive):
"""The whole request body, or None once it passes MAX_BODY_BYTES."""
body = b""
while True:
message = await receive()
body += message.get("body", b"")
if message["type"] == "http.disconnect" or len(body) > MAX_BODY_BYTES:
return None
if not message.get("more_body"):
return body
def rpc_error(code, text):
return JSONResponse({"jsonrpc": "2.0", "id": None, "error": {"code": code, "message": text}}, status_code=400)
def check(body, token):
"""A refusal for this request body, or None to let the SDK run it."""
if body is None:
return JSONResponse({"error": "request body too large"}, status_code=413)
try:
message = json.loads(body, object_pairs_hook=refuse_duplicate_keys)
except (ValueError, RecursionError): # not JSON, not UTF-8, or a repeated key
return rpc_error(-32700, "Parse error")
if not isinstance(message, dict): # a batch or a bare value: the SDK runs neither
return rpc_error(-32600, "Invalid Request")
needed = tool_scope(message)
if needed and needed not in token.scopes:
challenge = (f'Bearer error="insufficient_scope", scope="{needed}", '
f'resource_metadata="{METADATA_URL}", error_description="This tool needs {needed}"')
return JSONResponse({"error": "insufficient_scope", "error_description": f"This tool needs {needed}"},
status_code=403, headers={"WWW-Authenticate": challenge})
return None
class ScopeChallenge:
"""ASGI middleware: answer 403 insufficient_scope before a tool runs when the token lacks that tool's scope.
The tool name comes from the JSON-RPC body, the same bytes the SDK runs, never from the
Mcp-Name header. Compliant modern calls must include a matching header; malformed clients can omit it.
"""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] == "http" and scope["method"] == "POST":
auth = next((v.decode("latin-1") for k, v in scope["headers"] if k == b"authorization"), "") # the first, as Starlette reads it
token = await verifier.verify_token(auth[7:]) if auth.lower().startswith("bearer ") else None
if token is not None: # without a valid token the SDK answers 401 and the body is never read
body = await read_body(receive)
if (response := check(body, token)) is not None:
return await response(scope, receive, send)
receive = replay(body, receive)
async def add_scope_to_401(message):
# The SDK's 401 names the metadata URL; the authorization spec also asks for the scope.
if message["type"] == "http.response.start" and message["status"] == 401:
message = {**message, "headers": [(k, v + b', scope="docs:read"') if k == b"www-authenticate" else (k, v)
for k, v in message["headers"]]}
await send(message)
await self.app(scope, receive, add_scope_to_401)
def tool_scope(message):
"""The extra scope a tools/call message needs, read from params.name."""
params = message.get("params")
if message.get("method") == "tools/call" and isinstance(params, dict) and isinstance(params.get("name"), str):
return TOOL_SCOPES.get(params["name"])
return None
def replay(body, receive):
"""A receive() that gives the app the body this middleware already read, then the real stream."""
sent = False
async def again():
nonlocal sent
if sent:
return await receive()
sent = True
return {"type": "http.request", "body": body, "more_body": False}
return again
app = ScopeChallenge(server.streamable_http_app(json_response=True, transport_security=TransportSecuritySettings(
enable_dns_rebinding_protection=True, allowed_hosts=[f"127.0.0.1:{PORT}"], allowed_origins=[f"http://127.0.0.1:{PORT}"])))
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=PORT, log_level="warning")Three parts of the SDK do most of the work:
AuthSettingsmakes the SDK serve Protected Resource Metadata (RFC 9728) at/.well-known/oauth-protected-resource/mcp, naming this server’s canonical URI and its authorization server. The specification requires this document so that a client can find the authorization server on its own.token_verifier=JwtVerifier(...)checks each token’s signature, issuer, audience and expiry. A token whose audience is not this server’s URI is refused: the specification says servers must only accept tokens issued for them, and must not pass a client’s token on to another API.required_scopes=["docs:read"]refuses any MCP request without that scope. Without a valid token, the SDK answers401with aWWW-Authenticateheader that names the metadata URL.
The specification also asks the server to name the needed scope in the challenge, so the client requests only that. add_scope_to_401 appends scope="docs:read" to the SDK’s 401. MCP authorization: scope selection strategy
Diagnose identity, scope and request failures
Start with the status and challenge: 401 means the caller needs acceptable credentials; 403 means valid credentials lack the required scope. Malformed or oversized requests fail before tool execution. Inspect the exact refusal below when diagnosing a failing request.
Inspect the recorded refusal cases
A request without a token. The recorded reply is 401, with resource_metadata and scope="docs:read". Follow that challenge to obtain a token instead of guessing an issuer.
A token issued for another server. The server returns 401. Obtain a token with this server’s canonical URI as its resource; a valid signature alone is insufficient.
An expired token or a token in the query string. Both are refused with 401. Refresh expired credentials and send the access token in the Authorization header: query strings persist in logs and proxies.
A token for another audience, or an expired one. verify_token returns None. Reject the request and obtain a token meant for this resource; do not fall back to accepting its signature alone.
A signed token without client_id or scope, or with a malformed one. verify_token returns None instead of raising KeyError. Keep the claim-shape checks even for correctly signed tokens.
4. Require a scope per tool
A valid token can still lack a permission. For that case the specification says the server should answer 403 with WWW-Authenticate: Bearer error="insufficient_scope", the scope the operation needs and the resource_metadata URL. MCP authorization: scope challenge handling
The SDK checks one scope list for the whole endpoint, so ScopeChallenge, a small ASGI middleware, adds the per-tool check. It reads the JSON-RPC body, finds the tool in params.name, and answers 403 before the SDK runs anything when the token lacks that tool’s scope. Then it hands the same bytes to the SDK. It authorizes the tool named in the body rather than trusting the routing header. Under protocol version 2026-07-28, a compliant tools/call request must include a matching Mcp-Name header, Base64-encoded when required. The pinned SDK rejects a missing or mismatched header with HeaderMismatch; this middleware can refuse insufficient scope before that SDK check. It also refuses invalid JSON, a repeated key, a batch, or a body larger than 1 MB before forwarding it.
publish_page checks docs:write again itself. If the middleware is ever removed or bypassed, the tool still refuses.
TransportSecuritySettings accepts only this server’s own Host and Origin, which protects a local server from DNS rebinding, and uvicorn listens on 127.0.0.1 only. MCP Streamable HTTP: security warning
Inspect per-tool refusals
A valid token without the tool’s scope. ScopeChallenge returns 403 insufficient_scope with scope="docs:write". The client can request that scope, but the authorization server still decides whether to grant it.
An Mcp-Name header that names a different tool than the body. The recorded search_docs header with a publish_page body still gets 403. Authorize the tool named in the body bytes that the SDK will execute, rather than trusting the routing header.
A body that is not JSON, repeats a key or is over 1 MB. The middleware returns 400, 400 or 413 respectively. Fix the malformed request and keep the body limit before parsing; do not forward an unreadable body to the tool app.
A batch that calls publish_page. The middleware returns 400 -32600 before invoking the app. Send one JSON-RPC request rather than hiding a write in a batch.
A request from a foreign Origin. The recorded reply is 403 Invalid Origin header. Behind HTTPS, change RESOURCE, the metadata URL, and allowed Host/Origin values together; point ISSUER_URL at the intended authorization server.
A tool reached without its scope. publish_page itself raises ToolError. Retain this check so removing or bypassing middleware does not grant writes.
5. Connect with the SDK’s OAuth client
ClientCredentialsOAuthProvider is an httpx2 auth handler. The client starts with no token. On the first 401 it reads the metadata URL from the challenge, fetches the Protected Resource Metadata, then the authorization server metadata, and asks for a token with the challenged scope and resource set to this server’s URI. issuer= pins the only authorization server the client sends its secret to.
"""Call the remote docs server as an OAuth client with the MCP SDK's client_credentials provider.
The client starts with no token and no scope. The server's 401 names its Protected Resource
Metadata; the SDK reads it, finds the authorization server and asks for a token for this
server with the scope the challenge named. A later 403 insufficient_scope makes the SDK ask
again for the union of the old and new scopes. Every HTTP request it sends is recorded.
"""
from mcp import Client, MCPError
from mcp.client.auth import OAuthFlowError, OAuthTokenError
from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider
from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client
AUTH_ERRORS = (OAuthFlowError, OAuthTokenError)
class MemoryTokenStorage:
"""Tokens in memory only. A real client keeps them in a secret store, never in logs."""
tokens = client_info = 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
async def run_client(stack, client_id, calls, clean=str, issuer=None):
"""Make `calls` [(tool, arguments)] as `client_id`; return each outcome and the HTTP requests sent.
`issuer` is the only authorization server the client sends its secret to; it defaults to the stack's issuer.
"""
sent, outcomes, storage = [], [], MemoryTokenStorage()
async def record(response):
method = response.request.headers.get("mcp-method")
target = clean(f"{response.request.method} {response.request.url.copy_with(query=None)}")
sent.append(f"{target}{f' ({method})' if method else ''} -> {response.status_code}")
auth = ClientCredentialsOAuthProvider(
server_url=stack.resource, storage=storage, client_id=client_id, client_secret=stack.clients[client_id]["secret"],
token_endpoint_auth_method="client_secret_post", issuer=issuer or stack.issuer)
http = create_mcp_http_client(auth=auth)
http.event_hooks["response"].append(record)
try:
async with http, Client(streamable_http_client(stack.resource, http_client=http)) as client:
for tool, arguments in calls:
try:
result = await client.call_tool(tool, arguments)
outcomes.append({"tool": tool, "outcome": "tool_error" if result.is_error else "result"})
except MCPError as error:
outcomes.append({"tool": tool, "outcome": "protocol_error", "detail": error.error.message})
except BaseException as error: # anyio wraps errors raised inside the transport's task group
found = [e for e in getattr(error, "exceptions", [error]) if isinstance(e, AUTH_ERRORS)]
if not found:
raise
outcomes.append({"outcome": "authorization_failed", "detail": clean(f"{type(found[0]).__name__}: {found[0]}")})
return {"client": client_id, "granted_scope": storage.tokens.scope if storage.tokens else None,
"outcomes": outcomes, "http": sent}On a 403 insufficient_scope, the SDK steps up: it asks for a new token with the union of the scopes it already had and the challenged one, and retries the request. The specification says clients acting on their own behalf, as client_credentials clients do, may step up or stop. MCP authorization: step-up authorization flow
stack.py starts the issuer and the server on free localhost ports with secrets made at run time, and record_flows.py sends every request in the recorded run:
Read stack.py and record_flows.py
"""Start the token issuer and the docs service on free localhost ports, with client secrets made at run time.
Use as a context manager: `with Stack() as stack: ...`. Both processes stop on exit.
"""
import json
import os
import secrets
import socket
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
HERE = Path(__file__).resolve().parent
def free_port():
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
return probe.getsockname()[1]
def wait_for(url, seconds=20):
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
try:
with urllib.request.urlopen(url, timeout=1):
return
except OSError:
time.sleep(0.1)
raise RuntimeError(f"{url} did not start")
class Stack:
def __init__(self):
self.issuer_port, self.service_port, other_port = free_port(), free_port(), free_port()
self.issuer = f"http://127.0.0.1:{self.issuer_port}"
self.resource = f"http://127.0.0.1:{self.service_port}/mcp"
self.other_resource = f"http://127.0.0.1:{other_port}/mcp" # some other server's URI
self.clients = {
"docs-reader": {"secret": secrets.token_urlsafe(16), "scopes": ["docs:read"],
"resources": [self.resource, self.other_resource]},
"docs-editor": {"secret": secrets.token_urlsafe(16), "scopes": ["docs:read", "docs:write"],
"resources": [self.resource]},
"short-lived": {"secret": secrets.token_urlsafe(16), "scopes": ["docs:read"],
"resources": [self.resource], "lifetime": 1},
}
self.processes = []
def __enter__(self):
issuer_env = {**os.environ, "ISSUER_PORT": str(self.issuer_port), "CLIENTS_JSON": json.dumps(self.clients)}
service_env = {**os.environ, "SERVICE_PORT": str(self.service_port), "ISSUER_URL": self.issuer} # no client secrets
try:
for script, env, ready in (
("token_issuer.py", issuer_env, f"{self.issuer}/jwks.json"),
("docs_service.py", service_env, f"http://127.0.0.1:{self.service_port}/.well-known/oauth-protected-resource/mcp")):
self.processes.append(subprocess.Popen([sys.executable, str(HERE / script)], env=env,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL))
wait_for(ready)
except BaseException:
self.__exit__()
raise
return self
def __exit__(self, *exc):
for process in reversed(self.processes):
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
self.processes = []"""Send every request in the lesson to a local issuer and docs service, and save what came back.
Usage: python record_flows.py [--write auth-flows.json]
Ports and tokens differ on every run; the saved file replaces them with {issuer}, {service} and {other}.
"""
import argparse
import asyncio
import json
import logging
import time
from pathlib import Path
import httpx2
import jwt
from client import run_client
from stack import Stack
logging.getLogger("mcp.client.auth").setLevel(logging.CRITICAL) # the client's outcome list records the same error
META = {"io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": {}}
HEADERS = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json", "MCP-Protocol-Version": "2026-07-28"}
PUBLISH = {"name": "publish_page", "arguments": {"slug": "retries", "text": "Retries are off."}}
SEARCH = {"name": "search_docs", "arguments": {"query": "worker"}}
def rpc(method, params=None):
return {"jsonrpc": "2.0", "id": 1, "method": method, "params": {**(params or {}), "_meta": META}}
class Recorder:
def __init__(self, stack):
self.stack, self.http, self.results = stack, httpx2.Client(timeout=10), {}
def clean(self, text):
other = self.stack.other_resource.split("//")[1].split("/")[0]
return (text.replace(f"127.0.0.1:{self.stack.service_port}", "{service}")
.replace(f"127.0.0.1:{self.stack.issuer_port}", "{issuer}").replace(other, "{other}"))
def keep(self, label, response):
try:
body = json.loads(self.clean(response.text))
except ValueError:
body = self.clean(response.text)
self.results[label] = {"status": response.status_code, "body": body,
"www_authenticate": self.clean(response.headers.get("www-authenticate", "")) or None}
def token(self, client, scope, resource):
return self.http.post(f"{self.stack.issuer}/token", data={
"grant_type": "client_credentials", "client_id": client, "client_secret": self.stack.clients[client]["secret"],
"scope": scope, "resource": resource})
def mcp(self, label, message, token=None, **extra):
headers = {**HEADERS, "Mcp-Method": message["method"], **extra}
if message["method"] == "tools/call":
headers.setdefault("Mcp-Name", message["params"]["name"])
if token:
headers["Authorization"] = f"Bearer {token}"
self.keep(label, self.http.post(self.stack.resource, json=message, headers=headers))
def raw_flows(rec):
s = rec.stack
rec.mcp("no token", rpc("tools/list"))
rec.keep("protected resource metadata", rec.http.get(f"http://127.0.0.1:{s.service_port}/.well-known/oauth-protected-resource/mcp"))
rec.keep("authorization server metadata", rec.http.get(f"{s.issuer}/.well-known/oauth-authorization-server"))
rec.keep("token without resource", rec.token("docs-reader", "docs:read", ""))
rec.keep("token with a scope the client lacks", rec.token("docs-reader", "docs:write", s.resource))
rec.keep("token for a resource the client was not given", rec.token("docs-editor", "docs:read", s.other_resource))
reader = rec.token("docs-reader", "docs:read", s.resource).json()["access_token"]
rec.results["reader token claims"] = {k: rec.clean(str(v)) for k, v in jwt.decode(reader, options={"verify_signature": False}).items()
if k in ("iss", "aud", "sub", "scope")}
rec.mcp("reader searches", rpc("tools/call", SEARCH), reader)
rec.mcp("reader calls publish_page", rpc("tools/call", PUBLISH), reader)
rec.mcp("Mcp-Name search_docs, body publish_page", rpc("tools/call", PUBLISH), reader, **{"Mcp-Name": "search_docs"})
raw = {**HEADERS, "Mcp-Method": "tools/call", "Authorization": f"Bearer {reader}"}
rec.keep("body that is not JSON", rec.http.post(s.resource, content=b"tools/call publish_page", headers=raw))
repeated = b'{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "search_docs", "name": "publish_page"}}'
rec.keep("body with a repeated key", rec.http.post(s.resource, content=repeated, headers=raw))
rec.keep("body over 1 MB", rec.http.post(s.resource, content=b" " * 1_000_001, headers=raw))
rec.mcp("foreign Origin header", rpc("tools/list"), reader, Origin="http://attacker.example")
other = rec.token("docs-reader", "docs:read", s.other_resource).json()["access_token"]
rec.mcp("token issued for another server", rpc("tools/list"), other)
short = rec.token("short-lived", "docs:read", s.resource).json()["access_token"]
time.sleep(2)
rec.mcp("expired token", rpc("tools/list"), short)
rec.keep("token in the query string", rec.http.post(f"{s.resource}?access_token={reader}", json=rpc("tools/list"),
headers={**HEADERS, "Mcp-Method": "tools/list"}))
rec.results["reader searches"]["body"]["result"].pop("_meta", None)
CALLS = [("search_docs", {"query": "worker"}), ("publish_page", {"slug": "retries", "text": "Retries are off."}),
("search_docs", {"query": "retries"})]
async def sdk_flows(rec):
"""The same three calls from two OAuth clients: one allowed docs:write, one not."""
rec.results["sdk clients"] = {c: await run_client(rec.stack, c, CALLS, rec.clean) for c in ("docs-editor", "docs-reader")}
elsewhere = "http://" + rec.stack.other_resource.split("//")[1].split("/")[0] # not the issuer the server names
rec.results["sdk client pinned to another issuer"] = await run_client(rec.stack, "docs-editor", CALLS[:1], rec.clean, elsewhere)
def record():
with Stack() as stack:
rec = Recorder(stack)
with rec.http:
raw_flows(rec)
asyncio.run(sdk_flows(rec))
return rec.results
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--write", type=Path)
results = record()
for label, value in results.items():
if "status" in value:
print(f"{label:46} {value['status']}")
clients = [*results["sdk clients"].values(), {**results["sdk client pinned to another issuer"], "note": " pinned to another issuer"}]
for client in clients:
print(f"SDK client {client['client']}{client.get('note', '')}: granted {client['granted_scope']!r}")
for outcome in client["outcomes"]:
print(f" {outcome.get('tool', '-'):14} {outcome['outcome']}")
for line in client["http"]:
print(f" {line}")
if (path := parser.parse_args(argv).write):
path.write_text(json.dumps(results, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(f"wrote {path}")
if __name__ == "__main__":
main().venv/bin/python record_flows.py --write auth-flows.jsonHere is an excerpt of the recorded run: one raw HTTP request per line with the status it got, then the scopes that each of three SDK clients was granted. The full output also lists every HTTP request each SDK client sent.
no token 401
protected resource metadata 200
authorization server metadata 200
token without resource 400
token with a scope the client lacks 400
token for a resource the client was not given 400
reader searches 200
reader calls publish_page 403
Mcp-Name search_docs, body publish_page 403
body that is not JSON 400
body with a repeated key 400
body over 1 MB 413
foreign Origin header 403
token issued for another server 401
expired token 401
token in the query string 401
SDK client docs-editor: granted 'docs:read docs:write'
SDK client docs-reader: granted 'docs:read'
SDK client docs-editor pinned to another issuer: granted NoneA step-up the issuer refuses. The SDK raises OAuthTokenError and the tool does not run. Report the denied permission; repeating the same token request cannot grant it.
A client pinned to another issuer. The SDK raises OAuthFlowError before a token request. Check the configured issuer against the trusted service’s metadata rather than removing the pin to make a connection succeed.
6. Read the recorded flows
The editor’s requests are the whole flow in order. Its first request got 401; it fetched the two metadata documents and a token with docs:read; search_docs worked. publish_page got 403 with scope="docs:write", the SDK asked again and was granted docs:read docs:write, and the retried call ran.
The reader made the same calls. Its step-up request got 400 invalid_scope from the authorization server, because that client may not have docs:write, so the SDK raised OAuthTokenError and the recorded trace contains no publish_page execution. The third client pinned an issuer that is not the one the server names. The SDK stopped at OAuthFlowError: Authorization server metadata issuer mismatch: http://{issuer} != http://{other} before making a token request.
This excerpt of the saved flows shows the reader’s 403 and the pinned client that stopped at the issuer check:
{
"reader calls publish_page": {
"body": {
"error": "insufficient_scope",
"error_description": "This tool needs docs:write"
},
"status": 403,
"www_authenticate": "Bearer error=\"insufficient_scope\", scope=\"docs:write\", resource_metadata=\"http://{service}/.well-known/oauth-protected-resource/mcp\", error_description=\"This tool needs docs:write\""
},
"sdk client pinned to another issuer": {
"client": "docs-editor",
"granted_scope": null,
"http": [
"POST http://{service}/mcp (server/discover) -> 401",
"GET http://{service}/.well-known/oauth-protected-resource/mcp -> 200",
"GET http://{issuer}/.well-known/oauth-authorization-server -> 200"
],
"outcomes": [
{
"detail": "OAuthFlowError: Authorization server metadata issuer mismatch: http://{issuer} != http://{other}",
"outcome": "authorization_failed"
}
]
}
}Full output: auth-flows.json.
7. Test the server
The suite starts its own issuer and server, records every flow again and compares it with the saved file. It also runs ScopeChallenge in front of a stub app, checks each token claim on its own, and calls publish_page directly with a read-only token.
.venv/bin/python -m unittest -v test_remote_auth.pytest_a_body_that_is_not_one_json_object_is_refused_before_any_tool (test_remote_auth.RecordedFlows.test_a_body_that_is_not_one_json_object_is_refused_before_any_tool) ... ok
test_a_client_pinned_to_another_issuer_never_sends_its_secret (test_remote_auth.RecordedFlows.test_a_client_pinned_to_another_issuer_never_sends_its_secret) ... ok
test_a_foreign_origin_is_refused (test_remote_auth.RecordedFlows.test_a_foreign_origin_is_refused) ... ok
test_a_refused_step_up_never_runs_the_tool (test_remote_auth.RecordedFlows.test_a_refused_step_up_never_runs_the_tool) ... ok
test_a_request_without_a_token_is_told_where_to_get_one (test_remote_auth.RecordedFlows.test_a_request_without_a_token_is_told_where_to_get_one) ... ok
test_a_token_without_the_tool_scope_gets_403_insufficient_scope (test_remote_auth.RecordedFlows.test_a_token_without_the_tool_scope_gets_403_insufficient_scope) ... ok
test_expired_and_query_string_tokens_are_refused (test_remote_auth.RecordedFlows.test_expired_and_query_string_tokens_are_refused) ... ok
test_saved_fixture_matches_this_run (test_remote_auth.RecordedFlows.test_saved_fixture_matches_this_run) ... ok
test_sdk_client_discovers_then_steps_up_to_the_union_of_scopes (test_remote_auth.RecordedFlows.test_sdk_client_discovers_then_steps_up_to_the_union_of_scopes) ... ok
test_the_issuer_refuses_what_the_client_was_not_given (test_remote_auth.RecordedFlows.test_the_issuer_refuses_what_the_client_was_not_given) ... ok
test_tokens_are_bound_to_one_resource (test_remote_auth.RecordedFlows.test_tokens_are_bound_to_one_resource) ... ok
test_a_batch_that_calls_publish_page_is_refused_before_the_app (test_remote_auth.ScopeChallengeInProcess.test_a_batch_that_calls_publish_page_is_refused_before_the_app) ... ok
test_a_call_the_token_allows_reaches_the_app_with_the_same_bytes (test_remote_auth.ScopeChallengeInProcess.test_a_call_the_token_allows_reaches_the_app_with_the_same_bytes) ... ok
test_publish_page_refuses_a_token_without_docs_write (test_remote_auth.ToolRechecksScope.test_publish_page_refuses_a_token_without_docs_write) ... ok
test_verify_token_accepts_a_token_for_this_server (test_remote_auth.VerifierClaims.test_verify_token_accepts_a_token_for_this_server) ... ok
test_verify_token_refuses_a_token_for_another_audience (test_remote_auth.VerifierClaims.test_verify_token_refuses_a_token_for_another_audience) ... ok
test_verify_token_refuses_a_token_without_client_id_or_scope (test_remote_auth.VerifierClaims.test_verify_token_refuses_a_token_without_client_id_or_scope)
A signed token for this server with a missing or malformed claim is invalid, not a KeyError. ... ok
test_verify_token_refuses_an_expired_token (test_remote_auth.VerifierClaims.test_verify_token_refuses_an_expired_token) ... ok
----------------------------------------------------------------------
Ran 18 tests in 3.143s
OKAll 18 tests passed.
Read the test source
"""Tests for the remote docs server: python -m unittest -v test_remote_auth.py
Each run starts its own token issuer and docs service on free localhost ports.
"""
import asyncio
import json
import os
import time
import unittest
from pathlib import Path
from unittest import mock
import record_flows
HERE = Path(__file__).resolve().parent
METADATA = 'resource_metadata="http://{service}/.well-known/oauth-protected-resource/mcp"'
class RecordedFlows(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.flows = record_flows.record()
cls.editor, cls.reader = (cls.flows["sdk clients"][name] for name in ("docs-editor", "docs-reader"))
def status(self, label):
return self.flows[label]["status"]
def test_a_request_without_a_token_is_told_where_to_get_one(self):
self.assertEqual(self.status("no token"), 401)
self.assertIn(METADATA, self.flows["no token"]["www_authenticate"])
self.assertIn('scope="docs:read"', self.flows["no token"]["www_authenticate"])
self.assertEqual(self.flows["protected resource metadata"]["body"], {
"resource": "http://{service}/mcp", "authorization_servers": ["http://{issuer}"],
"scopes_supported": ["docs:read"], "bearer_methods_supported": ["header"]})
def test_tokens_are_bound_to_one_resource(self):
self.assertEqual(self.flows["reader token claims"]["aud"], "http://{service}/mcp")
self.assertEqual(self.flows["token without resource"]["body"]["error"], "invalid_target")
self.assertEqual(self.status("token issued for another server"), 401)
def test_the_issuer_refuses_what_the_client_was_not_given(self):
self.assertEqual(self.flows["token for a resource the client was not given"]["body"]["error"], "invalid_target")
self.assertEqual(self.flows["token with a scope the client lacks"]["body"]["error"], "invalid_scope")
def test_expired_and_query_string_tokens_are_refused(self):
self.assertEqual((self.status("expired token"), self.status("token in the query string")), (401, 401))
def test_a_token_without_the_tool_scope_gets_403_insufficient_scope(self):
for label in ("reader calls publish_page", "Mcp-Name search_docs, body publish_page"):
self.assertEqual(self.status(label), 403, label)
self.assertTrue(self.flows[label]["www_authenticate"].startswith('Bearer error="insufficient_scope", scope="docs:write"'))
self.assertIn(METADATA, self.flows[label]["www_authenticate"])
self.assertEqual(self.status("reader searches"), 200)
def test_a_body_that_is_not_one_json_object_is_refused_before_any_tool(self):
self.assertEqual([self.status(x) for x in ("body that is not JSON", "body with a repeated key", "body over 1 MB")], [400, 400, 413])
self.assertEqual(self.flows["body with a repeated key"]["body"]["error"]["code"], -32700)
def test_a_foreign_origin_is_refused(self):
self.assertEqual((self.status("foreign Origin header"), self.flows["foreign Origin header"]["body"]), (403, "Invalid Origin header"))
def test_sdk_client_discovers_then_steps_up_to_the_union_of_scopes(self):
self.assertEqual(self.editor["granted_scope"], "docs:read docs:write")
self.assertEqual([o["outcome"] for o in self.editor["outcomes"]], ["result", "result", "result"])
self.assertEqual(self.editor["http"][:4], [
"POST http://{service}/mcp (server/discover) -> 401",
"GET http://{service}/.well-known/oauth-protected-resource/mcp -> 200",
"GET http://{issuer}/.well-known/oauth-authorization-server -> 200",
"POST http://{issuer}/token -> 200"])
self.assertIn("POST http://{service}/mcp (tools/call) -> 403", self.editor["http"])
def test_a_refused_step_up_never_runs_the_tool(self):
self.assertEqual(self.reader["granted_scope"], "docs:read")
self.assertIn("invalid_scope", self.reader["outcomes"][-1]["detail"])
self.assertEqual(self.reader["http"][-1], "POST http://{issuer}/token -> 400")
self.assertNotIn("publish_page", [o.get("tool") for o in self.reader["outcomes"]])
def test_a_client_pinned_to_another_issuer_never_sends_its_secret(self):
pinned = self.flows["sdk client pinned to another issuer"]
self.assertIn("issuer mismatch", pinned["outcomes"][0]["detail"])
self.assertFalse(any("/token" in line for line in pinned["http"]))
def test_saved_fixture_matches_this_run(self):
self.assertEqual(json.loads((HERE / "auth-flows.json").read_text(encoding="utf-8")), json.loads(json.dumps(self.flows)))
def service_module():
with mock.patch.dict(os.environ, {"SERVICE_PORT": "1", "ISSUER_URL": "http://127.0.0.1:2"}):
import docs_service
return docs_service
def reader_token():
from mcp.server.auth.provider import AccessToken
return AccessToken(token="t", client_id="docs-reader", scopes=["docs:read"], expires_at=None)
class ScopeChallengeInProcess(unittest.TestCase):
"""ScopeChallenge in front of a stub app, so a refusal can be checked to stop before the app."""
def post(self, body):
service, reached, sent = service_module(), [], []
async def app(scope, receive, send):
reached.append((await receive())["body"])
await send({"type": "http.response.start", "status": 200, "headers": []})
await send({"type": "http.response.body", "body": b"{}"})
async def receive():
return {"type": "http.request", "body": body, "more_body": False}
async def send(message):
sent.append(message)
scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": [(b"authorization", b"Bearer t")]}
with mock.patch.object(service.verifier, "verify_token", mock.AsyncMock(return_value=reader_token())):
asyncio.run(service.ScopeChallenge(app)(scope, receive, send))
return sent[0]["status"], json.loads(b"".join(m.get("body", b"") for m in sent[1:])), reached
def test_a_batch_that_calls_publish_page_is_refused_before_the_app(self):
batch = [{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "publish_page", "arguments": {}}}]
status, body, reached = self.post(json.dumps(batch).encode())
self.assertEqual((status, body["error"]["code"], reached), (400, -32600, []))
def test_a_call_the_token_allows_reaches_the_app_with_the_same_bytes(self):
call = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "search_docs"}}).encode()
status, _, reached = self.post(call)
self.assertEqual((status, reached), (200, [call]))
class VerifierClaims(unittest.TestCase):
"""verify_token() with a local signing key, so each claim check is tested on its own."""
MISSING = object() # pass as a claim value to leave the claim out of the token
def setUp(self):
import jwt
from cryptography.hazmat.primitives.asymmetric import ec
self.jwt, self.service, self.key = jwt, service_module(), ec.generate_private_key(ec.SECP256R1())
keys = mock.patch.object(self.service.verifier.keys, "get_signing_key_from_jwt", return_value=mock.Mock(key=self.key.public_key()))
keys.start()
self.addCleanup(keys.stop)
def verify(self, **changes):
now = int(time.time())
claims = {"iss": "http://127.0.0.1:2", "aud": "http://127.0.0.1:1/mcp", "sub": "docs-reader", "client_id": "docs-reader",
"scope": "docs:read", "iat": now, "exp": now + 300, **changes}
claims = {name: value for name, value in claims.items() if value is not self.MISSING}
return asyncio.run(self.service.verifier.verify_token(self.jwt.encode(claims, self.key, algorithm="ES256")))
def test_verify_token_accepts_a_token_for_this_server(self):
self.assertEqual(self.verify().scopes, ["docs:read"])
def test_verify_token_refuses_a_token_for_another_audience(self):
self.assertIsNone(self.verify(aud="http://127.0.0.1:3/mcp"))
def test_verify_token_refuses_an_expired_token(self):
self.assertIsNone(self.verify(iat=1, exp=2))
def test_verify_token_refuses_a_token_without_client_id_or_scope(self):
"""A signed token for this server with a missing or malformed claim is invalid, not a KeyError."""
for claim in ("client_id", "scope"):
with self.subTest(missing=claim):
self.assertIsNone(self.verify(**{claim: self.MISSING}))
for claim, value in (("client_id", None), ("client_id", 7), ("client_id", ""),
("scope", None), ("scope", ["docs:read"])):
with self.subTest(malformed=claim, value=value):
self.assertIsNone(self.verify(**{claim: value}))
class ToolRechecksScope(unittest.TestCase):
def test_publish_page_refuses_a_token_without_docs_write(self):
service = service_module()
from mcp.server.mcpserver.exceptions import ToolError
with mock.patch.object(service, "get_access_token", return_value=reader_token()):
with self.assertRaisesRegex(ToolError, "docs:write"):
service.publish_page("retries", "Retries are off.")
if __name__ == "__main__":
unittest.main()Check token claims and tool scopes
Four documents carry the whole flow. These are the recorded values, with ports replaced by {service} and {issuer}:
{
"401 WWW-Authenticate": "Bearer error=\"invalid_token\", error_description=\"Authentication required\", resource_metadata=\"http://{service}/.well-known/oauth-protected-resource/mcp\", scope=\"docs:read\"",
"Protected Resource Metadata": {
"authorization_servers": [
"http://{issuer}"
],
"bearer_methods_supported": [
"header"
],
"resource": "http://{service}/mcp",
"scopes_supported": [
"docs:read"
]
},
"access token claims (reader)": {
"aud": "http://{service}/mcp",
"iss": "http://{issuer}",
"scope": "docs:read",
"sub": "docs-reader"
},
"403 WWW-Authenticate": "Bearer error=\"insufficient_scope\", scope=\"docs:write\", resource_metadata=\"http://{service}/.well-known/oauth-protected-resource/mcp\", error_description=\"This tool needs docs:write\""
}The 401 tells the client where the metadata is and which scope to ask for. The metadata names the server’s canonical URI as resource and its issuer in authorization_servers. The client sends that URI as resource in the token request, and the issuer copies it into the token’s aud claim. On every request, JwtVerifier fetches the issuer’s public key by the token’s kid, checks the ES256 signature, and then checks iss, aud and exp. It also requires client_id and scope to be present and to be strings before it builds the AccessToken: a signed token without them is refused as invalid, instead of raising KeyError in the verifier. The SDK then compares the token’s resource with resource_server_url a second time and checks required_scopes. ScopeChallenge adds the per-tool scope. Only then does the request reach a tool, where get_access_token() returns the same verified token.
What an attacker could do without each check
- Without the audience check, any token from the same issuer works here. The recorded run asked the issuer for a reader token for another server,
{other}; this server refused it with401. Without the check, any other MCP server that a client sends its token to could replay that token against this one. That is the token passthrough attack the specification forbids. - Without the per-tool scope, every client allowed to search could call
publish_pageand rewrite the docs.docs:readwould mean full write access. - Without reading the scope from the body, a client could send
Mcp-Name: search_docswith a body that callspublish_page. The recorded request did exactly that and got403. - Without the Origin check, a web page in the user’s browser could send requests to a server on
127.0.0.1through DNS rebinding. The recorded foreignOrigingot403. - Without the issuer pin in the client, a malicious server could name its own authorization server in its metadata and collect the client’s secret. The pinned client stopped before any token request.
The design choices
- Local JWT checks, or token introspection.
JwtVerifierchecks each token with the issuer’s public key and no network call per request. The cost is revocation: a revoked token stays valid untilexp. Introspection (RFC 7662) asks the authorization server about every token, which can observe revocation on a fresh check but adds a request per call; caching introspection responses adds its own delay. Short lifetimes, 300 seconds intoken_issuer.py, limit the gap. client_credentials, or an authorization code with PKCE. This lesson’s clients act on their own behalf, as a CI job or a backend agent does. A client that acts for a person must use the authorization code flow with PKCE, so the token carries that person’s grant, not the program’s.- Scope per tool, or one scope per server. One scope is simpler, but then read and write access cannot be separated. A scope per tool, checked before the tool runs, follows least privilege and gives the client a precise 403 to step up from.
Check your understanding
Read the explanation
A valid token without a tool's scope gets 403 insufficient_scope. The client may step up to the union of its scopes and the challenged one; the authorization server decides whether to grant it.
Build a read-only database MCP server with scoped roles continues with the next part of the MCP application.
References
Checked October 21, 2026, against MCP specification 2026-07-28, the latest version, and mcp 2.3.0.
- MCP specification: authorization: roles, scope selection, resource parameter, token handling, scope challenges and step-up.
- MCP specification: authorization server discovery: Protected Resource Metadata and the 401
resource_metadataparameter. - MCP specification: authorization security considerations: audience validation, token passthrough, HTTPS.
- MCP specification: Streamable HTTP transport
- RFC 9728: OAuth 2.0 Protected Resource Metadata
- RFC 8707: Resource Indicators for OAuth 2.0
- RFC 8414: OAuth 2.0 Authorization Server Metadata
- RFC 6750: Bearer Token Usage
- MCP Python SDK repository and mcp 2.3.0 on PyPI