modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

SEP-2207: Refresh token guidance (#2039)

Browse source
This commit is contained in:
Wils Dawson 2026-04-14 03:48:07 -07:00 • committed by GitHub
commit 437d15aa71
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 392 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

15
src/mcp/client/auth/oauth2.py
View file

@ -320,7 +320,7 @@ class OAuthClientProvider(httpx.Auth):
raise OAuthFlowError("No callback handler provided for authorization code grant") # pragma: no cover
if self.context.oauth_metadata and self.context.oauth_metadata.authorization_endpoint:
auth_endpoint = str(self.context.oauth_metadata.authorization_endpoint) # pragma: no cover
auth_endpoint = str(self.context.oauth_metadata.authorization_endpoint)
else:
auth_base_url = self.context.get_authorization_base_url(self.context.server_url)
auth_endpoint = urljoin(auth_base_url, "/authorize")
@ -343,11 +343,16 @@ class OAuthClientProvider(httpx.Auth):
# Only include resource param if conditions are met
if self.context.should_include_resource_param(self.context.protocol_version):
auth_params["resource"] = self.context.get_resource_url() # RFC 8707 # pragma: no cover
auth_params["resource"] = self.context.get_resource_url() # RFC 8707
if self.context.client_metadata.scope: # pragma: no branch
auth_params["scope"] = self.context.client_metadata.scope
# OIDC requires prompt=consent when offline_access is requested
# https://openid.net/specs/openid-connect-core-1_0.html#OfflineAccess
if "offline_access" in self.context.client_metadata.scope.split():
auth_params["prompt"] = "consent"
authorization_url = f"{auth_endpoint}?{urlencode(auth_params)}"
await self.context.redirect_handler(authorization_url)
@ -576,6 +581,7 @@ class OAuthClientProvider(httpx.Auth):
extract_scope_from_www_auth(response),
self.context.protected_resource_metadata,
self.context.oauth_metadata,
self.context.client_metadata.grant_types,
)
# Step 4: Register client or use URL-based client ID (CIMD)
@ -622,7 +628,10 @@ class OAuthClientProvider(httpx.Auth):
try:
# Step 2a: Update the required scopes
self.context.client_metadata.scope = get_client_metadata_scopes(
extract_scope_from_www_auth(response), self.context.protected_resource_metadata
extract_scope_from_www_auth(response),
self.context.protected_resource_metadata,
self.context.oauth_metadata,
self.context.client_metadata.grant_types,
)
# Step 2b: Perform (re-)authorization and token exchange

38
src/mcp/client/auth/utils.py
View file

@ -99,24 +99,36 @@ def get_client_metadata_scopes(
www_authenticate_scope: str | None,
protected_resource_metadata: ProtectedResourceMetadata | None,
authorization_server_metadata: OAuthMetadata | None = None,
client_grant_types: list[str] | None = None,
) -> str | None:
"""Select scopes as outlined in the 'Scope Selection Strategy' in the MCP spec."""
# Per MCP spec, scope selection priority order:
# 1. Use scope from WWW-Authenticate header (if provided)
# 2. Use all scopes from PRM scopes_supported (if available)
# 3. Omit scope parameter if neither is available
"""Select effective scopes and augment for refresh token support."""
selected_scope: str | None = None
# MCP spec scope selection priority:
# 1. WWW-Authenticate header scope
# 2. PRM scopes_supported
# 3. AS scopes_supported (SDK fallback)
# 4. Omit scope parameter
if www_authenticate_scope is not None:
# Priority 1: WWW-Authenticate header scope
return www_authenticate_scope
selected_scope = www_authenticate_scope
elif protected_resource_metadata is not None and protected_resource_metadata.scopes_supported is not None:
# Priority 2: PRM scopes_supported
return " ".join(protected_resource_metadata.scopes_supported)
selected_scope = " ".join(protected_resource_metadata.scopes_supported)
elif authorization_server_metadata is not None and authorization_server_metadata.scopes_supported is not None:
return " ".join(authorization_server_metadata.scopes_supported) # pragma: no cover
else:
# Priority 3: Omit scope parameter
return None
selected_scope = " ".join(authorization_server_metadata.scopes_supported)
# SEP-2207: append offline_access when the AS supports it and the client can use refresh tokens
if (
selected_scope is not None
and authorization_server_metadata is not None
and authorization_server_metadata.scopes_supported is not None
and "offline_access" in authorization_server_metadata.scopes_supported
and client_grant_types is not None
and "refresh_token" in client_grant_types
and "offline_access" not in selected_scope.split()
):
selected_scope = f"{selected_scope} offline_access"
return selected_scope
def build_oauth_authorization_server_metadata_discovery_urls(auth_server_url: str | None, server_url: str) -> list[str]:

354
tests/client/test_auth.py
View file

@ -2264,3 +2264,357 @@ class TestCIMD:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
class TestSEP2207OfflineAccessScope:
"""Test SEP-2207: offline_access scope augmentation for OIDC-flavored refresh tokens."""
def _make_as_metadata(self, scopes_supported: list[str] | None = None) -> OAuthMetadata:
return OAuthMetadata(
issuer=AnyHttpUrl("https://auth.example.com"),
authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
scopes_supported=scopes_supported,
)
def _make_prm(self, scopes_supported: list[str] | None = None) -> ProtectedResourceMetadata:
return ProtectedResourceMetadata(
resource=AnyHttpUrl("https://api.example.com/v1/mcp"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
scopes_supported=scopes_supported,
)
def test_offline_access_added_when_as_supports_and_client_has_refresh_token(self):
"""offline_access is appended when AS advertises it and client supports refresh_token grant."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write offline_access"
def test_offline_access_added_with_www_authenticate_scope(self):
"""offline_access is appended even when scopes come from WWW-Authenticate header."""
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope="read write",
protected_resource_metadata=None,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write offline_access"
def test_offline_access_not_added_when_as_does_not_support(self):
"""offline_access is not added when AS does not advertise it in scopes_supported."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write"
def test_offline_access_not_added_when_client_has_no_refresh_token_grant(self):
"""offline_access is not added when client does not support refresh_token grant."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code"],
)
assert scopes == "read write"
def test_offline_access_not_duplicated_when_already_present(self):
"""offline_access is not added again if it already appears in the selected scopes."""
prm = self._make_prm(scopes_supported=["read", "offline_access", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read offline_access write"
def test_offline_access_not_added_when_no_scopes_selected(self):
"""offline_access is not added when no base scopes are available (None)."""
asm = self._make_as_metadata(scopes_supported=["offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=None,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
# When AS scopes are the only source and include offline_access,
# the base scope is "offline_access" and no duplication happens
assert scopes == "offline_access"
def test_offline_access_not_added_when_as_scopes_supported_is_none(self):
"""offline_access is not added when AS scopes_supported is None."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=None)
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write"
def test_offline_access_not_added_when_no_as_metadata(self):
"""offline_access is not added when AS metadata is not available."""
prm = self._make_prm(scopes_supported=["read", "write"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=None,
client_grant_types=["authorization_code", "refresh_token"],
)
assert scopes == "read write"
def test_offline_access_not_added_when_no_grant_types_provided(self):
"""offline_access is not added when client_grant_types is None."""
prm = self._make_prm(scopes_supported=["read", "write"])
asm = self._make_as_metadata(scopes_supported=["read", "write", "offline_access"])
scopes = get_client_metadata_scopes(
www_authenticate_scope=None,
protected_resource_metadata=prm,
authorization_server_metadata=asm,
client_grant_types=None,
)
assert scopes == "read write"
def test_default_client_metadata_includes_refresh_token_grant(self):
"""Default OAuthClientMetadata includes refresh_token in grant_types (SEP-2207 guidance)."""
metadata = OAuthClientMetadata(redirect_uris=[AnyUrl("http://localhost:3030/callback")])
assert "refresh_token" in metadata.grant_types
@pytest.mark.anyio
async def test_auth_flow_adds_offline_access_when_as_advertises(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""E2E: auth flow includes offline_access in authorization request when AS advertises it."""
captured_auth_url: str | None = None
captured_state: str | None = None
async def redirect_handler(url: str) -> None:
nonlocal captured_auth_url, captured_state
captured_auth_url = url
parsed = urlparse(url)
params = parse_qs(parsed.query)
captured_state = params.get("state", [None])[0]
async def callback_handler() -> tuple[str, str | None]:
return "test_auth_code", captured_state
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
# Pre-set client info to skip DCR
provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
test_request = httpx.Request("GET", "https://api.example.com/v1/mcp")
auth_flow = provider.async_auth_flow(test_request)
# First request
request = await auth_flow.__anext__()
assert "Authorization" not in request.headers
# Send 401
response = httpx.Response(401, headers={}, request=test_request)
# PRM discovery
prm_request = await auth_flow.asend(response)
prm_response = httpx.Response(
200,
content=(
b'{"resource": "https://api.example.com/v1/mcp",'
b' "authorization_servers": ["https://auth.example.com"],'
b' "scopes_supported": ["read", "write"]}'
),
request=prm_request,
)
# OAuth metadata discovery - AS advertises offline_access
oauth_request = await auth_flow.asend(prm_response)
oauth_response = httpx.Response(
200,
content=(
b'{"issuer": "https://auth.example.com",'
b' "authorization_endpoint": "https://auth.example.com/authorize",'
b' "token_endpoint": "https://auth.example.com/token",'
b' "scopes_supported": ["read", "write", "offline_access"]}'
),
request=oauth_request,
)
# This triggers authorization, which calls redirect_handler
token_request = await auth_flow.asend(oauth_response)
# Verify the authorization URL included offline_access in the scope
assert captured_auth_url is not None
parsed = urlparse(captured_auth_url)
params = parse_qs(parsed.query)
scope_value = params["scope"][0]
scope_parts = scope_value.split()
assert "offline_access" in scope_parts
assert "read" in scope_parts
assert "write" in scope_parts
# OIDC requires prompt=consent when offline_access is requested
assert params["prompt"][0] == "consent"
# Complete the token exchange
token_response = httpx.Response(
200,
content=(
b'{"access_token": "new_access_token", "token_type": "Bearer",'
b' "expires_in": 3600, "refresh_token": "new_refresh_token"}'
),
request=token_request,
)
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer new_access_token"
# Close the generator
final_response = httpx.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass
@pytest.mark.anyio
async def test_auth_flow_no_offline_access_when_as_does_not_advertise(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
):
"""E2E: auth flow does NOT include offline_access when AS doesn't advertise it."""
captured_auth_url: str | None = None
captured_state: str | None = None
async def redirect_handler(url: str) -> None:
nonlocal captured_auth_url, captured_state
captured_auth_url = url
parsed = urlparse(url)
params = parse_qs(parsed.query)
captured_state = params.get("state", [None])[0]
async def callback_handler() -> tuple[str, str | None]:
return "test_auth_code", captured_state
provider = OAuthClientProvider(
server_url="https://api.example.com/v1/mcp",
client_metadata=client_metadata,
storage=mock_storage,
redirect_handler=redirect_handler,
callback_handler=callback_handler,
)
provider.context.current_tokens = None
provider.context.token_expiry_time = None
provider._initialized = True
# Pre-set client info to skip DCR
provider.context.client_info = OAuthClientInformationFull(
client_id="test_client",
client_secret="test_secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
test_request = httpx.Request("GET", "https://api.example.com/v1/mcp")
auth_flow = provider.async_auth_flow(test_request)
# First request
await auth_flow.__anext__()
# Send 401
response = httpx.Response(401, headers={}, request=test_request)
# PRM discovery
prm_request = await auth_flow.asend(response)
prm_response = httpx.Response(
200,
content=(
b'{"resource": "https://api.example.com/v1/mcp",'
b' "authorization_servers": ["https://auth.example.com"],'
b' "scopes_supported": ["read", "write"]}'
),
request=prm_request,
)
# OAuth metadata discovery - AS does NOT advertise offline_access
oauth_request = await auth_flow.asend(prm_response)
oauth_response = httpx.Response(
200,
content=(
b'{"issuer": "https://auth.example.com",'
b' "authorization_endpoint": "https://auth.example.com/authorize",'
b' "token_endpoint": "https://auth.example.com/token",'
b' "scopes_supported": ["read", "write"]}'
),
request=oauth_request,
)
# This triggers authorization, which calls redirect_handler
token_request = await auth_flow.asend(oauth_response)
# Verify the authorization URL does NOT include offline_access
assert captured_auth_url is not None
parsed = urlparse(captured_auth_url)
params = parse_qs(parsed.query)
scope_value = params["scope"][0]
scope_parts = scope_value.split()
assert "offline_access" not in scope_parts
assert "read" in scope_parts
assert "write" in scope_parts
# prompt=consent should NOT be present without offline_access
assert "prompt" not in params
# Complete the token exchange
token_response = httpx.Response(
200,
content=b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600}',
request=token_request,
)
final_request = await auth_flow.asend(token_response)
assert final_request.headers["Authorization"] == "Bearer new_access_token"
# Close the generator
final_response = httpx.Response(200, request=final_request)
try:
await auth_flow.asend(final_response)
except StopAsyncIteration:
pass