Fold stored token scope into step-up union (SEP-2350)

On restart only the persisted token is reloaded, not client_metadata.scope, so
the step-up union would re-authorize for less than was previously granted. Seed
the union with the stored token's scope as well so prior grants survive a
restart. Caught in review by Codex.

Author: Kludex

src/mcp/client/auth/oauth2.py

@@ -636,16 +636,18 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
                 if error == "insufficient_scope":  # pragma: no branch
                     try:
                         # Step 2a: Union previously requested scopes with the newly challenged
-                        # scopes (SEP-2350) so escalating one operation keeps the others' grants
+                        # scopes (SEP-2350) so escalating one operation keeps the others' grants.
+                        # Fold in the stored token's scope too: on a restart the token is reloaded
+                        # but client_metadata.scope is not, so it would otherwise be the only basis.
                         challenged_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,
                         )
-                        self.context.client_metadata.scope = union_scopes(
-                            self.context.client_metadata.scope, challenged_scope
-                        )
+                        granted_scope = self.context.current_tokens.scope if self.context.current_tokens else None
+                        prior_scope = union_scopes(self.context.client_metadata.scope, granted_scope)
+                        self.context.client_metadata.scope = union_scopes(prior_scope, challenged_scope)
 
                         # Step 2b: Perform (re-)authorization and token exchange
                         token_response = yield await self._perform_authorization()

tests/client/test_auth.py

@@ -1447,6 +1447,65 @@ async def mock_callback() -> AuthorizationCodeResult:
         except StopAsyncIteration:
             pass  # Expected
 
+    @pytest.mark.anyio
+    async def test_403_step_up_preserves_scope_from_stored_token(
+        self, oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+    ):
+        """SEP-2350: a restart-loaded token's scope is folded into the step-up union.
+
+        On restart only the token is reloaded (not client_metadata.scope), so the stored token's
+        granted scope must seed the union, or the challenge would re-authorize for less.
+        """
+        client_info = OAuthClientInformationFull(
+            client_id="test_client_id",
+            client_secret="test_client_secret",
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        )
+        # Simulate a restart: a token granted "read" is loaded, but client_metadata carries no scope.
+        oauth_provider.context.current_tokens = OAuthToken(access_token="t", scope="read")
+        oauth_provider.context.token_expiry_time = time.time() + 1800
+        oauth_provider.context.client_info = client_info
+        oauth_provider.context.client_metadata.scope = None
+        oauth_provider._initialized = True
+
+        captured_state: str | None = None
+        reauthorize_scope: str | None = None
+
+        async def capture_redirect(url: str) -> None:
+            nonlocal captured_state, reauthorize_scope
+            params = parse_qs(urlparse(url).query)
+            reauthorize_scope = params["scope"][0]
+            captured_state = params.get("state", [None])[0]
+
+        async def mock_callback() -> AuthorizationCodeResult:
+            return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+        oauth_provider.context.redirect_handler = capture_redirect
+        oauth_provider.context.callback_handler = mock_callback
+
+        auth_flow = oauth_provider.async_auth_flow(httpx.Request("GET", "https://api.example.com/mcp"))
+        request = await auth_flow.__anext__()
+        response_403 = httpx.Response(
+            403,
+            headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+            request=request,
+        )
+        token_exchange_request = await auth_flow.asend(response_403)
+
+        assert reauthorize_scope == "read write"
+
+        # Drive the flow to completion so the context lock is released cleanly
+        token_response = httpx.Response(
+            200,
+            json={"access_token": "new", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+            request=token_exchange_request,
+        )
+        final_request = await auth_flow.asend(token_response)
+        try:
+            await auth_flow.asend(httpx.Response(200, request=final_request))
+        except StopAsyncIteration:
+            pass
+
 
 @pytest.mark.parametrize(
     (