Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion src/mcp/client/auth/oauth2.py
Original file line number Diff line number Diff line change
Expand Up @@ -687,12 +687,15 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
self.context.client_info = None
self.context.clear_tokens()

# Step 3: Apply scope selection strategy
# Step 3: Apply scope selection strategy, preserving an explicitly
# configured scope. An explicit empty string intentionally requests no
# scopes, so only ``None`` opts into discovery-based selection.
self.context.client_metadata.scope = get_client_metadata_scopes(
extract_scope_from_www_auth(response),
self.context.protected_resource_metadata,
self.context.oauth_metadata,
self.context.client_metadata.grant_types,
configured_scope=self.context.client_metadata.scope,
)

# Step 4: Register client or use URL-based client ID (CIMD)
Expand Down
23 changes: 15 additions & 8 deletions src/mcp/client/auth/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,21 +102,27 @@ def get_client_metadata_scopes(
protected_resource_metadata: ProtectedResourceMetadata | None,
authorization_server_metadata: OAuthMetadata | None = None,
client_grant_types: list[str] | None = None,
configured_scope: str | None = None,
) -> str | None:
"""Select effective scopes and augment for refresh token support."""
selected_scope: str | None = None
"""Select effective scopes and augment for refresh token support.

An explicitly configured scope is preferred over discovered scopes. ``None``
continues to opt into the discovery-based selection strategy.
"""
selected_scope: str | None = configured_scope

# 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:
selected_scope = www_authenticate_scope
elif protected_resource_metadata is not None and protected_resource_metadata.scopes_supported is not None:
selected_scope = " ".join(protected_resource_metadata.scopes_supported)
elif authorization_server_metadata is not None and authorization_server_metadata.scopes_supported is not None:
selected_scope = " ".join(authorization_server_metadata.scopes_supported)
if selected_scope is None:
if www_authenticate_scope is not None:
selected_scope = www_authenticate_scope
elif protected_resource_metadata is not None and protected_resource_metadata.scopes_supported is not None:
selected_scope = " ".join(protected_resource_metadata.scopes_supported)
elif authorization_server_metadata is not None and authorization_server_metadata.scopes_supported is not 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 (
Expand All @@ -126,6 +132,7 @@ def get_client_metadata_scopes(
and "offline_access" in authorization_server_metadata.scopes_supported
and client_grant_types is not None
and "refresh_token" in client_grant_types
and selected_scope
and "offline_access" not in selected_scope.split()
):
selected_scope = f"{selected_scope} offline_access"
Expand Down
49 changes: 49 additions & 0 deletions tests/client/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,55 @@ def test_clear_tokens(self, oauth_provider: OAuthClientProvider, valid_tokens: O
class TestOAuthFlow:
"""Test OAuth flow methods."""

@pytest.mark.anyio
async def test_explicit_client_scope_is_not_overwritten_by_discovery(self, oauth_provider: OAuthClientProvider):
oauth_provider.context.client_metadata.scope = "explicit:read"
oauth_provider.context.client_info = OAuthClientInformationFull(
client_id="existing-client",
client_secret="client-secret",
redirect_uris=[AnyUrl("http://localhost:3030/callback")],
)
oauth_provider._initialized = True
oauth_provider._perform_authorization = mock.AsyncMock(
return_value=httpx2.Request("POST", "https://auth.example.com/token")
)

auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
request = await auth_flow.__anext__()
prm_request = await auth_flow.asend(
httpx2.Response(
401,
headers={"WWW-Authenticate": 'Bearer resource_metadata="https://api.example.com/.well-known/prm"'},
request=request,
)
)
prm_response = httpx2.Response(
200,
json={
"resource": "https://api.example.com/v1/mcp",
"authorization_servers": ["https://auth.example.com"],
"scopes_supported": ["discovered:read", "discovered:write"],
},
request=prm_request,
)
asm_request = await auth_flow.asend(prm_response)
asm_response = httpx2.Response(
200,
json={
"issuer": "https://auth.example.com",
"authorization_endpoint": "https://auth.example.com/authorize",
"token_endpoint": "https://auth.example.com/token",
"scopes_supported": ["discovered:read", "discovered:write"],
},
request=asm_request,
)

token_request = await auth_flow.asend(asm_response)

assert oauth_provider.context.client_metadata.scope == "explicit:read"
assert token_request.method == "POST"
await auth_flow.aclose()

@pytest.mark.anyio
async def test_build_protected_resource_discovery_urls(
self, client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
Expand Down
Loading