From 4fac3c7b015ccc3ac6c806c528bc8c618448c1e3 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 7 Aug 2026 05:32:40 +0000
Subject: [PATCH 1/9] fix(client/auth): discard stored client registrations
 with an expired secret

The client persists client_secret_expires_at (RFC 7591) through
TokenStorage but never reads it back, and registration only happens
when stored client info is absent. Once a dynamically registered
secret lapses, every token-endpoint interaction fails with
invalid_client - including the exchange after a fresh interactive
authorization - so the client is permanently stuck until the
application manually deletes the persisted client info (#3256).

Treat a stored registration whose secret-authenticating record carries
a non-zero, past client_secret_expires_at as absent when loading from
storage. The next 401 flow then re-registers (or resolves CIMD) and
overwrites the dead record via the existing set_client_info call - no
change to the TokenStorage contract. Stored tokens are kept: a live
access token continues to work without client authentication, and with
no client info the refresh path that would present the lapsed secret
is skipped.

Fixes #3256

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CjbXueCDdFNJK6imejCXgM
---
 src/mcp/client/auth/oauth2.py |  34 +++++++++++-
 tests/client/test_auth.py     | 101 ++++++++++++++++++++++++++++++++++
 2 files changed, 133 insertions(+), 2 deletions(-)

diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index 7dc62b52b9..46cb33e9c4 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -109,6 +109,22 @@ def check_registration_usable(client_info: OAuthClientInformationFull) -> None:
         )
 
 
+def stored_registration_expired(client_info: OAuthClientInformationFull) -> bool:
+    """Whether a stored registration's minted secret has lapsed and can no longer authenticate.
+
+    RFC 7591 requires `client_secret_expires_at` whenever a secret is issued, with ``0``
+    meaning the secret never expires. Once a non-zero expiry passes, every token-endpoint
+    interaction authenticating with that secret fails with ``invalid_client`` — and with no
+    RFC 7592 rotation endpoint, re-registration is the only standard recovery. The lapse
+    only matters for registrations that authenticate with the minted secret: ``none`` (or
+    an absent method) sends no secret, and `private_key_jwt` signs an assertion instead.
+    """
+    if client_info.token_endpoint_auth_method not in _SECRET_TOKEN_ENDPOINT_AUTH_METHODS:
+        return False
+    expires_at = client_info.client_secret_expires_at
+    return expires_at is not None and expires_at != 0 and expires_at < int(time.time())
+
+
 class PKCEParameters(BaseModel):
     """PKCE (Proof Key for Code Exchange) parameters."""
 
@@ -548,9 +564,23 @@ async def _handle_refresh_response(self, response: httpx2.Response) -> bool:
             return False
 
     async def _initialize(self) -> None:
-        """Load stored tokens and client info."""
+        """Load stored tokens and client info.
+
+        Stored client information whose minted secret has expired (RFC 7591
+        `client_secret_expires_at`) is treated as absent: reusing it can only produce
+        `invalid_client` at the token endpoint — even interactive re-authorization ends in
+        the same failure, permanently — so it is discarded here and the next 401 flow
+        re-registers (or resolves CIMD), overwriting the dead record in storage. Any still
+        stored tokens are kept: a live access token keeps working without client
+        authentication, and with no client info the refresh path (which would present the
+        lapsed secret) is skipped.
+        """
         self.context.current_tokens = await self.context.storage.get_tokens()
-        self.context.client_info = await self.context.storage.get_client_info()
+        client_info = await self.context.storage.get_client_info()
+        if client_info is not None and stored_registration_expired(client_info):
+            logger.debug("Stored client registration secret has expired; discarding so the next flow re-registers")
+            client_info = None
+        self.context.client_info = client_info
         self._initialized = True
 
     def _add_auth_header(self, request: httpx2.Request) -> None:
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index be96cc8eec..d9056dd42a 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -13,6 +13,7 @@
 
 from mcp.client.auth import OAuthClientProvider, PKCEParameters
 from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError, OAuthTokenError
+from mcp.client.auth.oauth2 import stored_registration_expired
 from mcp.client.auth.utils import (
     build_oauth_authorization_server_metadata_discovery_urls,
     build_protected_resource_metadata_discovery_urls,
@@ -3253,3 +3254,103 @@ async def echo_callback() -> AuthorizationCodeResult:
         await auth_flow.asend(httpx2.Response(200, request=final_req))
     except StopAsyncIteration:
         pass
+
+
+def test_stored_registration_expired_only_for_lapsed_secret_backed_registrations():
+    """RFC 7591: only a non-zero, past `client_secret_expires_at` on a secret-authenticating
+    registration marks the stored record as expired; `0` means the secret never expires, and
+    methods that send no secret (`none`) are unaffected by the lapse.
+    """
+    base: dict[str, object] = {
+        "client_id": "c",
+        "client_secret": "s",
+        "redirect_uris": [AnyUrl("http://localhost:3030/callback")],
+    }
+    lapsed = int(time.time()) - 3600
+    live = int(time.time()) + 3600
+
+    expired = OAuthClientInformationFull.model_validate(
+        {**base, "token_endpoint_auth_method": "client_secret_post", "client_secret_expires_at": lapsed}
+    )
+    assert stored_registration_expired(expired)
+    assert stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "client_secret_basic", "client_secret_expires_at": lapsed}
+        )
+    )
+
+    # 0 means "never expires" (RFC 7591); absent means no expiry was declared.
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "client_secret_post", "client_secret_expires_at": 0}
+        )
+    )
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate({**base, "token_endpoint_auth_method": "client_secret_post"})
+    )
+
+    # Still-live secret, and methods that never present the secret.
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "client_secret_post", "client_secret_expires_at": live}
+        )
+    )
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "none", "client_secret_expires_at": lapsed}
+        )
+    )
+
+
+@pytest.mark.anyio
+async def test_expired_stored_registration_is_discarded_and_the_flow_re_registers(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, valid_tokens: OAuthToken
+):
+    """Regression for #3256: a stored DCR registration whose secret has lapsed is not reused.
+
+    Reusing it makes every token-endpoint interaction fail with ``invalid_client`` — even a
+    fresh interactive authorization ends in the same failure, so the client is permanently
+    stuck (\"I re-authenticated and nothing changed\"). The lapsed record must be treated as
+    absent on load, so the next 401 flow re-registers instead of presenting the dead secret;
+    stored tokens are kept (a live access token still works without client authentication).
+    """
+    await mock_storage.set_client_info(
+        OAuthClientInformationFull(
+            client_id="dead-client",
+            client_secret="expired-secret",
+            client_secret_expires_at=int(time.time()) - 3600,
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_post",
+        )
+    )
+    await mock_storage.set_tokens(valid_tokens)
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+
+    # The lapsed registration is treated as absent; the stored access token is kept and used.
+    request = await auth_flow.__anext__()
+    assert oauth_provider.context.client_info is None
+    assert oauth_provider.context.current_tokens is not None
+    assert request.headers["Authorization"] == f"Bearer {valid_tokens.access_token}"
+
+    # Server rejects the stale token: the 401 flow re-registers instead of reusing the record.
+    response_401 = httpx2.Response(401, request=request)
+    prm_req = await auth_flow.asend(response_401)
+    prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
+    asm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"issuer": "https://api.example.com", '
+            b'"authorization_endpoint": "https://api.example.com/authorize", '
+            b'"token_endpoint": "https://api.example.com/token", '
+            b'"registration_endpoint": "https://api.example.com/register"}'
+        ),
+        request=asm_req,
+    )
+
+    register_req = await auth_flow.asend(asm_response)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://api.example.com/register"
+    await auth_flow.aclose()

From 50d6b0b2ff10fb44bd2dad2ef35ee8aa766a5a4e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 7 Aug 2026 05:56:10 +0000
Subject: [PATCH 2/9] fix(client/auth): re-check secret expiry in the 401 flow;
 document the discard

Address review findings: _initialize runs once per provider instance,
so a registration whose secret lapses mid-session (long-lived process)
was still reused - Step 4 skipped re-registration and the interactive
authorization burned a user consent only to fail invalid_client at the
token exchange. Re-check stored_registration_expired at the start of
the 401 handler and discard the dead record so Step 4 re-registers.

Also update docs/client/oauth-clients.md: the storage tip and the
'stored client_info still wins' statement now mention the
expired-secret discard.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CjbXueCDdFNJK6imejCXgM
---
 docs/client/oauth-clients.md  |  7 +++++-
 src/mcp/client/auth/oauth2.py | 12 ++++++++++
 tests/client/test_auth.py     | 45 +++++++++++++++++++++++++++++++++++
 3 files changed, 63 insertions(+), 1 deletion(-)

diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md
index cd7de35626..131ebe3e85 100644
--- a/docs/client/oauth-clients.md
+++ b/docs/client/oauth-clients.md
@@ -52,6 +52,11 @@ The in-memory version above works. It also forgets everything when the process e
     Store `client_info`, not only the tokens. The provider registers dynamically the first time it
     finds no stored `client_info`. Throw it away and you mint a fresh registration on every run.
 
+    One exception: a stored registration whose dynamically issued secret has expired (a non-zero
+    `client_secret_expires_at` in the past) is treated as absent — the expired secret could never
+    authenticate again, so the provider discards the record and re-registers on the next flow,
+    overwriting it in your storage with the fresh registration.
+
 ### The two handlers
 
 The authorization code flow needs a human exactly once: someone has to sign in and click "allow".
@@ -95,7 +100,7 @@ The repository ships the live version. `examples/servers/simple-auth/` runs a st
 
 The 2026-07-28 revision of the spec deprecates dynamic client registration in favor of **Client ID Metadata Documents** (CIMD). Instead of POSTing a fresh registration to every authorization server it meets, your client publishes one JSON document about itself at a stable HTTPS URL, and that URL *is* its `client_id`. The authorization server fetches the document; the provider never touches it.
 
-The SDK already speaks it: pass the URL as `client_metadata_url=` when you construct the provider. When the authorization server's metadata advertises `client_id_metadata_document_supported: true`, the provider skips the `/register` request entirely: the URL goes into the flow as the `client_id`, and there is no `client_secret`. When the server doesn't advertise it (most don't yet), or you never pass a URL, the provider falls back to dynamic registration **silently**, and everything above works exactly as described. Stored `client_info` still wins over both.
+The SDK already speaks it: pass the URL as `client_metadata_url=` when you construct the provider. When the authorization server's metadata advertises `client_id_metadata_document_supported: true`, the provider skips the `/register` request entirely: the URL goes into the flow as the `client_id`, and there is no `client_secret`. When the server doesn't advertise it (most don't yet), or you never pass a URL, the provider falls back to dynamic registration **silently**, and everything above works exactly as described. Stored `client_info` still wins over both, as long as its registration is usable — a record whose dynamically issued secret has expired is discarded and the provider registers (or resolves the CIMD URL) afresh.
 
 The URL must be HTTPS with a non-root path; anything else is a `ValueError` at construction, before any network happens. The shipped `examples/clients/simple-auth-client/` takes it as the `MCP_CLIENT_METADATA_URL` environment variable.
 
diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index 46cb33e9c4..a35c8a10ac 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -634,6 +634,18 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                 # Perform full OAuth flow
                 try:
                     # OAuth flow must be inline due to generator constraints
+
+                    # A registration whose minted secret lapsed mid-session (after
+                    # _initialize already loaded it) can no longer authenticate either —
+                    # discard it here too, so Step 4 re-registers instead of running an
+                    # interactive authorization doomed to fail `invalid_client` at the
+                    # token endpoint.
+                    if self.context.client_info is not None and stored_registration_expired(self.context.client_info):
+                        logger.debug(
+                            "Stored client registration secret has expired; discarding so this flow re-registers"
+                        )
+                        self.context.client_info = None
+
                     www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response)
 
                     # Step 1: Discover protected resource metadata (SEP-985 with fallback support)
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index d9056dd42a..64f7a40323 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -3354,3 +3354,48 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
     assert register_req.method == "POST"
     assert str(register_req.url) == "https://api.example.com/register"
     await auth_flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_registration_that_expires_mid_session_is_discarded_by_the_401_flow(
+    oauth_provider: OAuthClientProvider,
+):
+    """A secret that lapses after load (long-lived process) is also discarded, in-flow.
+
+    ``_initialize`` runs once per provider instance, so a registration that expires while
+    the process is running would otherwise be reused by the 401 flow: Step 4 would skip
+    re-registration and the interactive authorization would burn a user consent only to
+    fail ``invalid_client`` at the token exchange. The 401 handler re-checks the expiry
+    and discards the dead record so Step 4 re-registers instead.
+    """
+    oauth_provider._initialized = True  # already initialized while the secret was live
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,  # lapsed after load
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # 401 → discovery; the lapsed registration is discarded, so the flow re-registers.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"issuer": "https://api.example.com", '
+            b'"authorization_endpoint": "https://api.example.com/authorize", '
+            b'"token_endpoint": "https://api.example.com/token", '
+            b'"registration_endpoint": "https://api.example.com/register"}'
+        ),
+        request=asm_req,
+    )
+
+    register_req = await auth_flow.asend(asm_response)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://api.example.com/register"
+    await auth_flow.aclose()

From 6b7ab0d87fdcc00b4ce8191cf4bbe1f020f6b356 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 7 Aug 2026 05:59:44 +0000
Subject: [PATCH 3/9] chore: retrigger CI after PyPI download timeout

The 3.13/locked/ubuntu job failed fetching opentelemetry-api from
files.pythonhosted.org (operation timed out) during uv sync - an
infrastructure flake unrelated to the change.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CjbXueCDdFNJK6imejCXgM

From 5a246e67730261e92fa4adf2f1a348df6e2b5d8f Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 10 Aug 2026 05:33:50 +0000
Subject: [PATCH 4/9] fix(client/auth): run the expiry discard after the
 SEP-2352 issuer checks; cover the refresh and 403 step-up paths
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The expired-registration discard previously nulled client_info before the
SEP-2352 issuer-binding checks, so a record that was both expired and
issuer-mismatched skipped the cross-issuer cleanup (old tokens and cached AS
metadata were kept). The discard now runs just before Step 4, after both
issuer checks; _initialize loads the record as-is and defers the discard to
the flow.

The two other paths that present the minted secret are covered as well: the
refresh branch skips a refresh whose registration secret has lapsed (falling
through to the 401 flow's re-registration instead of failing invalid_client),
and the 403 insufficient_scope step-up re-registers — mirroring Step 4,
reusing any discovered AS metadata, extracted into
_prepare_client_registration/_complete_client_registration — before running
the interactive authorization, instead of burning a consent doomed to fail
invalid_client while the live access token keeps the 401 discard from running.
---
 src/mcp/client/auth/oauth2.py | 181 +++++++++++++--------
 tests/client/test_auth.py     | 298 +++++++++++++++++++++++++++++++++-
 2 files changed, 404 insertions(+), 75 deletions(-)

diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index a35c8a10ac..a98c5511a3 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -566,21 +566,16 @@ async def _handle_refresh_response(self, response: httpx2.Response) -> bool:
     async def _initialize(self) -> None:
         """Load stored tokens and client info.
 
-        Stored client information whose minted secret has expired (RFC 7591
-        `client_secret_expires_at`) is treated as absent: reusing it can only produce
-        `invalid_client` at the token endpoint — even interactive re-authorization ends in
-        the same failure, permanently — so it is discarded here and the next 401 flow
-        re-registers (or resolves CIMD), overwriting the dead record in storage. Any still
-        stored tokens are kept: a live access token keeps working without client
-        authentication, and with no client info the refresh path (which would present the
-        lapsed secret) is skipped.
+        A stored registration whose minted secret has expired (RFC 7591
+        `client_secret_expires_at`) is loaded as-is rather than discarded here: the auth
+        flow discards it right before re-registering, *after* the SEP-2352 issuer checks,
+        which need the record's issuer stamp — an expired record that is also bound to a
+        different issuer must still get its cross-issuer cleanup (dropping the old
+        issuer's tokens and cached metadata). Until then the dead secret is never
+        presented: the refresh branch and the 403 step-up skip it explicitly.
         """
         self.context.current_tokens = await self.context.storage.get_tokens()
-        client_info = await self.context.storage.get_client_info()
-        if client_info is not None and stored_registration_expired(client_info):
-            logger.debug("Stored client registration secret has expired; discarding so the next flow re-registers")
-            client_info = None
-        self.context.client_info = client_info
+        self.context.client_info = await self.context.storage.get_client_info()
         self._initialized = True
 
     def _add_auth_header(self, request: httpx2.Request) -> None:
@@ -607,6 +602,63 @@ async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None
         if not check_resource_allowed(requested_resource=default_resource, configured_resource=prm_resource):
             raise OAuthFlowError(f"Protected resource {prm_resource} does not match expected {default_resource}")
 
+    def _registration_issuer(self) -> str | None:
+        """SEP-2352: the issuer to bind newly minted credentials to, when known."""
+        if self.context.oauth_metadata is not None:
+            return self.context.auth_server_url or str(self.context.oauth_metadata.issuer)
+        return None
+
+    async def _prepare_client_registration(self) -> httpx2.Request | None:
+        """Resolve a URL-based client ID (CIMD) or build a Dynamic Client Registration request.
+
+        When the server supports CIMD the client information is created (and persisted)
+        immediately and ``None`` is returned — no network round trip is needed. Otherwise
+        the returned registration request must be sent and its response passed to
+        `_complete_client_registration`.
+        """
+        if should_use_client_metadata_url(self.context.oauth_metadata, self.context.client_metadata_url):
+            # Use URL-based client ID (CIMD). CIMD records are portable across
+            # authorization servers, so the issuer stamp is informational.
+            logger.debug(f"Using URL-based client ID (CIMD): {self.context.client_metadata_url}")
+            client_information = create_client_info_from_metadata_url(
+                self.context.client_metadata_url,  # type: ignore[arg-type]
+                redirect_uris=self.context.client_metadata.redirect_uris,
+            )
+            client_information.issuer = self._registration_issuer()
+            self.context.client_info = client_information
+            await self.context.storage.set_client_info(client_information)
+            return None
+
+        # Fallback to Dynamic Client Registration
+        fallback_base = self.context.get_authorization_base_url(self.context.server_url)
+        return create_client_registration_request(
+            self.context.oauth_metadata, self.context.client_metadata, fallback_base
+        )
+
+    async def _complete_client_registration(self, response: httpx2.Response) -> None:
+        """Handle a Dynamic Client Registration response and persist the minted record."""
+        client_information = await handle_registration_response(response)
+        check_registration_usable(client_information)
+        discovered_issuer = self._registration_issuer()
+        fallback_base = self.context.get_authorization_base_url(self.context.server_url)
+        # Only record the issuer when the registration actually targeted the discovered
+        # AS — either via its published registration_endpoint, or because the
+        # resource-origin /register fallback is on the issuer's own host (legacy
+        # same-origin embedded AS). Otherwise the fallback hit a different server and
+        # recording a binding to the PRM-advertised AS would persist a binding that was
+        # never established.
+        if (
+            self.context.oauth_metadata is not None
+            and discovered_issuer is not None
+            and (
+                self.context.oauth_metadata.registration_endpoint is not None
+                or self.context.get_authorization_base_url(discovered_issuer) == fallback_base
+            )
+        ):
+            client_information.issuer = discovered_issuer
+        self.context.client_info = client_information
+        await self.context.storage.set_client_info(client_information)
+
     async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
         """httpx2 auth flow integration."""
         async with self.context.lock:
@@ -616,7 +668,16 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
             # Capture protocol version from request headers
             self.context.protocol_version = request.headers.get(MCP_PROTOCOL_VERSION_HEADER)
 
-            if not self.context.is_token_valid() and self.context.can_refresh_token():
+            # A refresh request authenticates with the minted secret, so a registration
+            # whose secret has lapsed (RFC 7591 `client_secret_expires_at`) can only fail
+            # `invalid_client`. Skip the doomed refresh and fall through to the 401 flow,
+            # which re-registers; the record itself is kept for now so the flow's SEP-2352
+            # issuer checks can still read its issuer stamp before the expiry discard runs.
+            registration_expired = self.context.client_info is not None and stored_registration_expired(
+                self.context.client_info
+            )
+
+            if not self.context.is_token_valid() and self.context.can_refresh_token() and not registration_expired:
                 # Try to refresh token
                 refresh_request = await self._refresh_token()
                 refresh_response = yield refresh_request
@@ -635,17 +696,6 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                 try:
                     # OAuth flow must be inline due to generator constraints
 
-                    # A registration whose minted secret lapsed mid-session (after
-                    # _initialize already loaded it) can no longer authenticate either —
-                    # discard it here too, so Step 4 re-registers instead of running an
-                    # interactive authorization doomed to fail `invalid_client` at the
-                    # token endpoint.
-                    if self.context.client_info is not None and stored_registration_expired(self.context.client_info):
-                        logger.debug(
-                            "Stored client registration secret has expired; discarding so this flow re-registers"
-                        )
-                        self.context.client_info = None
-
                     www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response)
 
                     # Step 1: Discover protected resource metadata (SEP-985 with fallback support)
@@ -737,52 +787,27 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         self.context.client_metadata.grant_types,
                     )
 
+                    # A registration whose minted secret lapsed (RFC 7591
+                    # `client_secret_expires_at`) — whether loaded from storage or expired
+                    # mid-session — can no longer authenticate: reusing it would burn an
+                    # interactive authorization doomed to fail `invalid_client` at the
+                    # token endpoint. Discard it only now, after the SEP-2352 issuer
+                    # checks above, so an expired record bound to a different issuer
+                    # still got its cross-issuer cleanup; Step 4 then re-registers,
+                    # overwriting the dead record in storage. Any stored tokens are kept:
+                    # a live access token keeps working without client authentication.
+                    if self.context.client_info is not None and stored_registration_expired(self.context.client_info):
+                        logger.debug(
+                            "Stored client registration secret has expired; discarding so this flow re-registers"
+                        )
+                        self.context.client_info = None
+
                     # Step 4: Register client or use URL-based client ID (CIMD)
                     if not self.context.client_info:
-                        # SEP-2352: the issuer to bind these credentials to, when known.
-                        discovered_issuer: str | None = None
-                        if self.context.oauth_metadata is not None:
-                            discovered_issuer = self.context.auth_server_url or str(self.context.oauth_metadata.issuer)
-
-                        if should_use_client_metadata_url(
-                            self.context.oauth_metadata, self.context.client_metadata_url
-                        ):
-                            # Use URL-based client ID (CIMD). CIMD records are portable across
-                            # authorization servers, so the issuer stamp is informational.
-                            logger.debug(f"Using URL-based client ID (CIMD): {self.context.client_metadata_url}")
-                            client_information = create_client_info_from_metadata_url(
-                                self.context.client_metadata_url,  # type: ignore[arg-type]
-                                redirect_uris=self.context.client_metadata.redirect_uris,
-                            )
-                            client_information.issuer = discovered_issuer
-                            self.context.client_info = client_information
-                            await self.context.storage.set_client_info(client_information)
-                        else:
-                            # Fallback to Dynamic Client Registration
-                            fallback_base = self.context.get_authorization_base_url(self.context.server_url)
-                            registration_request = create_client_registration_request(
-                                self.context.oauth_metadata, self.context.client_metadata, fallback_base
-                            )
+                        registration_request = await self._prepare_client_registration()
+                        if registration_request is not None:
                             registration_response = yield registration_request
-                            client_information = await handle_registration_response(registration_response)
-                            check_registration_usable(client_information)
-                            # Only record the issuer when the registration above actually targeted
-                            # the discovered AS — either via its published registration_endpoint,
-                            # or because the resource-origin /register fallback is on the issuer's
-                            # own host (legacy same-origin embedded AS). Otherwise the fallback hit
-                            # a different server and recording a binding to the PRM-advertised AS
-                            # would persist a binding that was never established.
-                            if (
-                                self.context.oauth_metadata is not None
-                                and discovered_issuer is not None
-                                and (
-                                    self.context.oauth_metadata.registration_endpoint is not None
-                                    or self.context.get_authorization_base_url(discovered_issuer) == fallback_base
-                                )
-                            ):
-                                client_information.issuer = discovered_issuer
-                            self.context.client_info = client_information
-                            await self.context.storage.set_client_info(client_information)
+                            await self._complete_client_registration(registration_response)
 
                     # Step 5: Perform authorization and complete token exchange
                     token_response = yield await self._perform_authorization()
@@ -815,6 +840,26 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         prior_scope = union_scopes(self.context.client_metadata.scope, granted_scope)
                         self.context.client_metadata.scope = union_scopes(prior_scope, challenged_scope)
 
+                        # A registration whose minted secret lapsed (RFC 7591
+                        # `client_secret_expires_at`) cannot complete the step-up: the
+                        # token exchange would fail `invalid_client` after burning a full
+                        # interactive consent — and the still-live access token keeps the
+                        # 401 flow's discard from ever running. Discard it and mint fresh
+                        # credentials first (mirroring the 401 flow's Step 4, reusing any
+                        # AS metadata already discovered).
+                        if self.context.client_info is not None and stored_registration_expired(
+                            self.context.client_info
+                        ):
+                            logger.debug(
+                                "Stored client registration secret has expired; re-registering before the step-up"
+                            )
+                            self.context.client_info = None
+                        if not self.context.client_info:
+                            registration_request = await self._prepare_client_registration()
+                            if registration_request is not None:
+                                registration_response = yield registration_request
+                                await self._complete_client_registration(registration_response)
+
                         # Step 2b: Perform (re-)authorization and token exchange
                         token_response = yield await self._perform_authorization()
                         await self._handle_token_response(token_response)
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index 64f7a40323..847485f64f 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -3310,9 +3310,9 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
 
     Reusing it makes every token-endpoint interaction fail with ``invalid_client`` — even a
     fresh interactive authorization ends in the same failure, so the client is permanently
-    stuck (\"I re-authenticated and nothing changed\"). The lapsed record must be treated as
-    absent on load, so the next 401 flow re-registers instead of presenting the dead secret;
-    stored tokens are kept (a live access token still works without client authentication).
+    stuck (\"I re-authenticated and nothing changed\"). The 401 flow must discard the lapsed
+    record and re-register instead of presenting the dead secret; stored tokens are kept
+    (a live access token still works without client authentication).
     """
     await mock_storage.set_client_info(
         OAuthClientInformationFull(
@@ -3327,9 +3327,10 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
 
     auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
 
-    # The lapsed registration is treated as absent; the stored access token is kept and used.
+    # The record is loaded as-is (the flow discards it later, after the SEP-2352 issuer
+    # checks); the stored access token is kept and used.
     request = await auth_flow.__anext__()
-    assert oauth_provider.context.client_info is None
+    assert oauth_provider.context.client_info is not None
     assert oauth_provider.context.current_tokens is not None
     assert request.headers["Authorization"] == f"Bearer {valid_tokens.access_token}"
 
@@ -3351,6 +3352,7 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
     )
 
     register_req = await auth_flow.asend(asm_response)
+    assert oauth_provider.context.client_info is None  # discarded in-flow, just before Step 4
     assert register_req.method == "POST"
     assert str(register_req.url) == "https://api.example.com/register"
     await auth_flow.aclose()
@@ -3365,8 +3367,9 @@ async def test_registration_that_expires_mid_session_is_discarded_by_the_401_flo
     ``_initialize`` runs once per provider instance, so a registration that expires while
     the process is running would otherwise be reused by the 401 flow: Step 4 would skip
     re-registration and the interactive authorization would burn a user consent only to
-    fail ``invalid_client`` at the token exchange. The 401 handler re-checks the expiry
-    and discards the dead record so Step 4 re-registers instead.
+    fail ``invalid_client`` at the token exchange. The 401 flow re-checks the expiry
+    (after the SEP-2352 issuer checks) and discards the dead record so Step 4
+    re-registers instead.
     """
     oauth_provider._initialized = True  # already initialized while the secret was live
     oauth_provider.context.client_info = OAuthClientInformationFull(
@@ -3399,3 +3402,284 @@ async def test_registration_that_expires_mid_session_is_discarded_by_the_401_flo
     assert register_req.method == "POST"
     assert str(register_req.url) == "https://api.example.com/register"
     await auth_flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_expired_and_issuer_mismatched_registration_still_gets_cross_issuer_cleanup(
+    oauth_provider: OAuthClientProvider,
+):
+    """SEP-2352: the expiry discard must not shadow the issuer-binding cleanup.
+
+    A record that is both expired and bound to a different issuer (the server migrated
+    authorization servers while the secret lapsed) still needs the cross-issuer cleanup:
+    the old issuer's tokens are dropped so they can never be presented to the new AS, and
+    the old AS's cached metadata is dropped so a failed rediscovery cannot leak its
+    registration/token endpoints into Step 4. The expiry discard therefore runs only
+    after the issuer checks, which read the issuer stamp off the record.
+    """
+    oauth_provider._initialized = True
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+        issuer="https://old-as.example.com",
+    )
+    oauth_provider.context.current_tokens = OAuthToken(access_token="old-as-access", refresh_token="old-as-refresh")
+    oauth_provider.context.token_expiry_time = time.time() + 1800
+    oauth_provider.context.oauth_metadata = OAuthMetadata(
+        issuer=AnyHttpUrl("https://old-as.example.com"),
+        authorization_endpoint=AnyHttpUrl("https://old-as.example.com/authorize"),
+        token_endpoint=AnyHttpUrl("https://old-as.example.com/token"),
+        registration_endpoint=AnyHttpUrl("https://old-as.example.com/register"),
+    )
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # 401 → PRM discovery now advertises a different authorization server.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    prm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+
+    # ASM rediscovery for the new AS fails at both well-known URLs.
+    asm_req = await auth_flow.asend(prm_response)
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=asm_req))
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/openid-configuration"
+    register_req = await auth_flow.asend(httpx2.Response(404, request=asm_req))
+
+    # The issuer-binding cleanup fired despite the lapsed secret: the old issuer's tokens
+    # and cached metadata are gone, so registration goes to the resource-origin fallback,
+    # not the old AS's registration endpoint.
+    assert oauth_provider.context.current_tokens is None
+    assert oauth_provider.context.oauth_metadata is None
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://api.example.com/register"
+    await auth_flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_refresh_is_skipped_when_registration_secret_expired_mid_session(
+    oauth_provider: OAuthClientProvider,
+):
+    """The refresh path never presents a lapsed secret.
+
+    ``can_refresh_token()`` ignores expiry, so without the extra check the refresh would
+    authenticate with the dead secret and fail ``invalid_client`` before recovering via
+    the subsequent 401. Instead the doomed refresh is skipped outright: the request goes
+    out unauthenticated and the 401 flow re-registers.
+    """
+    oauth_provider._initialized = True
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,  # lapsed after load
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+    oauth_provider.context.current_tokens = OAuthToken(access_token="stale", refresh_token="still-there")
+    oauth_provider.context.token_expiry_time = time.time() - 60  # access token expired → refresh territory
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # No refresh request was yielded: the first request out is the original one, unauthenticated.
+    assert request.method == "GET"
+    assert str(request.url) == "https://api.example.com/v1/mcp"
+    assert "Authorization" not in request.headers
+
+    # The 401 flow then recovers by re-registering.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"issuer": "https://api.example.com", '
+            b'"authorization_endpoint": "https://api.example.com/authorize", '
+            b'"token_endpoint": "https://api.example.com/token", '
+            b'"registration_endpoint": "https://api.example.com/register"}'
+        ),
+        request=asm_req,
+    )
+    register_req = await auth_flow.asend(asm_response)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://api.example.com/register"
+    await auth_flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_403_step_up_re_registers_when_registration_secret_expired(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+):
+    """A 403 scope step-up with a lapsed secret re-registers instead of burning a consent.
+
+    On this path the access token is still live, so the 401 flow's discard can never run;
+    without its own re-check every step-up would complete a full interactive authorization
+    only to fail ``invalid_client`` at the token exchange, until the access token itself
+    expires. The step-up mints a fresh registration first — against the AS metadata already
+    discovered — and completes the exchange with the new credentials.
+    """
+    oauth_provider._initialized = True
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,  # lapsed after load
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+    oauth_provider.context.current_tokens = OAuthToken(access_token="live-token", scope="read")
+    oauth_provider.context.token_expiry_time = time.time() + 1800  # access token still live
+    oauth_provider.context.oauth_metadata = OAuthMetadata(
+        issuer=AnyHttpUrl("https://auth.example.com"),
+        authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+        token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        registration_endpoint=AnyHttpUrl("https://auth.example.com/register"),
+    )
+
+    captured_state: str | None = None
+    authorize_client_id: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state, authorize_client_id
+        params = parse_qs(urlparse(url).query)
+        authorize_client_id = params["client_id"][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(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # The step-up re-registers before authorizing, using the cached registration endpoint.
+    register_req = await auth_flow.asend(response_403)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://auth.example.com/register"
+    register_response = httpx2.Response(
+        201,
+        json={
+            "client_id": "fresh-client",
+            "client_secret": "fresh-secret",
+            "redirect_uris": ["http://localhost:3030/callback"],
+            "token_endpoint_auth_method": "client_secret_post",
+        },
+        request=register_req,
+    )
+
+    # The interactive authorization and token exchange use the fresh credentials only.
+    token_exchange_request = await auth_flow.asend(register_response)
+    assert authorize_client_id == "fresh-client"
+    content = token_exchange_request.content.decode()
+    assert "client_id=fresh-client" in content
+    assert "client_secret=fresh-secret" in content
+    assert "expired-secret" not in content
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "fresh-client"  # dead record overwritten in storage
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_403_step_up_resolves_cimd_when_registration_secret_expired(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+):
+    """A 403 step-up with a lapsed secret resolves the CIMD URL when the AS supports it.
+
+    Mirrors Step 4 of the 401 flow: with `client_id_metadata_document_supported` in the
+    cached AS metadata and a configured `client_metadata_url`, the dead registration is
+    replaced locally — no ``/register`` round trip — and the step-up proceeds straight to
+    authorization with the URL-based client ID (which has no secret to expire).
+    """
+    captured_state: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state
+        captured_state = parse_qs(urlparse(url).query).get("state", [None])[0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+        redirect_handler=capture_redirect,
+        callback_handler=mock_callback,
+        client_metadata_url="https://example.com/client",
+    )
+    provider._initialized = True
+    provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,  # lapsed after load
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+    provider.context.current_tokens = OAuthToken(access_token="live-token", scope="read")
+    provider.context.token_expiry_time = time.time() + 1800  # access token still live
+    provider.context.oauth_metadata = OAuthMetadata(
+        issuer=AnyHttpUrl("https://auth.example.com"),
+        authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+        token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        client_id_metadata_document_supported=True,
+    )
+
+    auth_flow = provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # No /register round trip: the next request is already the token exchange, and the
+    # dead record was replaced by the URL-based client ID.
+    token_exchange_request = await auth_flow.asend(response_403)
+    assert str(token_exchange_request.url) == "https://auth.example.com/token"
+    content = token_exchange_request.content.decode()
+    assert "client_id=https%3A%2F%2Fexample.com%2Fclient" in content
+    assert "expired-secret" not in content
+    assert provider.context.client_info is not None
+    assert provider.context.client_info.token_endpoint_auth_method == "none"
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "https://example.com/client"  # dead record overwritten in storage
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass

From e53e5263bb9e10e71bfff1c28275d4a9dfe572ec Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 10 Aug 2026 06:55:24 +0000
Subject: [PATCH 5/9] fix(client/auth): drop the orphaned refresh token on
 expiry discard; centralize the expiry predicate
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A refresh token is bound to the client it was issued to (RFC 6749 §6), so
once the expiry discard replaces the registration, the kept refresh token
could only be presented as the wrong client and fail invalid_grant — which
happened whenever a flow failed between re-registration and token exchange.
Both discard sites (401 pre-Step-4 and 403 step-up) now drop the refresh
token with the record and persist the trimmed tokens, so an interrupted flow
or a restart cannot resurrect the orphan; the live access token is a bearer
credential and is kept.

The null-guarded expiry check, previously inlined verbatim at three flow
sites, moves into OAuthContext.registration_secret_expired() beside its
sibling predicates, and the discard mechanics into
OAuthContext.discard_expired_registration(), giving the rule a single home.
---
 docs/client/oauth-clients.md  |   4 +-
 src/mcp/client/auth/oauth2.py |  38 +++++++---
 tests/client/test_auth.py     | 132 ++++++++++++++++++++++++++++++++--
 3 files changed, 158 insertions(+), 16 deletions(-)

diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md
index 131ebe3e85..c3e81c2510 100644
--- a/docs/client/oauth-clients.md
+++ b/docs/client/oauth-clients.md
@@ -55,7 +55,9 @@ The in-memory version above works. It also forgets everything when the process e
     One exception: a stored registration whose dynamically issued secret has expired (a non-zero
     `client_secret_expires_at` in the past) is treated as absent — the expired secret could never
     authenticate again, so the provider discards the record and re-registers on the next flow,
-    overwriting it in your storage with the fresh registration.
+    overwriting it in your storage with the fresh registration. Any refresh token goes with the
+    discarded record (it was issued to that `client_id` and no other client can redeem it), so the
+    stored tokens are rewritten without it; a still-live access token is kept and keeps working.
 
 ### The two handlers
 
diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index a98c5511a3..2dba6bc321 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -208,6 +208,25 @@ def can_refresh_token(self) -> bool:
         """Check if token can be refreshed."""
         return bool(self.current_tokens and self.current_tokens.refresh_token and self.client_info)
 
+    def registration_secret_expired(self) -> bool:
+        """Whether the loaded registration's minted secret has lapsed (RFC 7591)."""
+        return self.client_info is not None and stored_registration_expired(self.client_info)
+
+    async def discard_expired_registration(self) -> None:
+        """Discard a registration whose minted secret lapsed so the flow re-registers.
+
+        The refresh token goes with it: RFC 6749 §6 binds a refresh token to the client
+        it was issued to, so once the flow re-registers under a fresh `client_id` the
+        orphaned token could only fail `invalid_grant`. The trimmed tokens are persisted
+        so an interrupted flow (or a restart) cannot resurrect the orphan. The live
+        access token is a bearer credential that keeps working without client
+        authentication, so it is kept.
+        """
+        self.client_info = None
+        if self.current_tokens is not None and self.current_tokens.refresh_token is not None:
+            self.current_tokens.refresh_token = None
+            await self.storage.set_tokens(self.current_tokens)
+
     def clear_tokens(self) -> None:
         """Clear current tokens."""
         self.current_tokens = None
@@ -673,9 +692,7 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
             # `invalid_client`. Skip the doomed refresh and fall through to the 401 flow,
             # which re-registers; the record itself is kept for now so the flow's SEP-2352
             # issuer checks can still read its issuer stamp before the expiry discard runs.
-            registration_expired = self.context.client_info is not None and stored_registration_expired(
-                self.context.client_info
-            )
+            registration_expired = self.context.registration_secret_expired()
 
             if not self.context.is_token_valid() and self.context.can_refresh_token() and not registration_expired:
                 # Try to refresh token
@@ -794,13 +811,14 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                     # token endpoint. Discard it only now, after the SEP-2352 issuer
                     # checks above, so an expired record bound to a different issuer
                     # still got its cross-issuer cleanup; Step 4 then re-registers,
-                    # overwriting the dead record in storage. Any stored tokens are kept:
-                    # a live access token keeps working without client authentication.
-                    if self.context.client_info is not None and stored_registration_expired(self.context.client_info):
+                    # overwriting the dead record in storage. The refresh token is
+                    # dropped with the record it was issued to; the live access token
+                    # is kept — it works without client authentication.
+                    if self.context.registration_secret_expired():
                         logger.debug(
                             "Stored client registration secret has expired; discarding so this flow re-registers"
                         )
-                        self.context.client_info = None
+                        await self.context.discard_expired_registration()
 
                     # Step 4: Register client or use URL-based client ID (CIMD)
                     if not self.context.client_info:
@@ -847,13 +865,11 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         # 401 flow's discard from ever running. Discard it and mint fresh
                         # credentials first (mirroring the 401 flow's Step 4, reusing any
                         # AS metadata already discovered).
-                        if self.context.client_info is not None and stored_registration_expired(
-                            self.context.client_info
-                        ):
+                        if self.context.registration_secret_expired():
                             logger.debug(
                                 "Stored client registration secret has expired; re-registering before the step-up"
                             )
-                            self.context.client_info = None
+                            await self.context.discard_expired_registration()
                         if not self.context.client_info:
                             registration_request = await self._prepare_client_registration()
                             if registration_request is not None:
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index 847485f64f..c5dc7c5fd9 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -3302,6 +3302,24 @@ def test_stored_registration_expired_only_for_lapsed_secret_backed_registrations
     )
 
 
+def test_registration_secret_expired_is_false_until_a_lapsed_registration_is_loaded(
+    oauth_provider: OAuthClientProvider,
+):
+    """`OAuthContext.registration_secret_expired` owns the null check for the flow's call
+    sites (refresh gate, 401 discard, 403 step-up): no registration loaded means "not
+    expired", and a loaded record defers to `stored_registration_expired`.
+    """
+    assert not oauth_provider.context.registration_secret_expired()
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+    assert oauth_provider.context.registration_secret_expired()
+
+
 @pytest.mark.anyio
 async def test_expired_stored_registration_is_discarded_and_the_flow_re_registers(
     oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, valid_tokens: OAuthToken
@@ -3311,8 +3329,9 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
     Reusing it makes every token-endpoint interaction fail with ``invalid_client`` — even a
     fresh interactive authorization ends in the same failure, so the client is permanently
     stuck (\"I re-authenticated and nothing changed\"). The 401 flow must discard the lapsed
-    record and re-register instead of presenting the dead secret; stored tokens are kept
-    (a live access token still works without client authentication).
+    record and re-register instead of presenting the dead secret; the access token is kept
+    (it still works without client authentication) while the refresh token — issued to the
+    discarded `client_id` — goes with the record.
     """
     await mock_storage.set_client_info(
         OAuthClientInformationFull(
@@ -3355,6 +3374,13 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
     assert oauth_provider.context.client_info is None  # discarded in-flow, just before Step 4
     assert register_req.method == "POST"
     assert str(register_req.url) == "https://api.example.com/register"
+    # The discard also dropped the dead client's refresh token, in memory and in storage.
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+    stored_tokens = await mock_storage.get_tokens()
+    assert stored_tokens is not None
+    assert stored_tokens.refresh_token is None
+    assert stored_tokens.access_token == valid_tokens.access_token
     await auth_flow.aclose()
 
 
@@ -3515,6 +3541,93 @@ async def test_refresh_is_skipped_when_registration_secret_expired_mid_session(
     await auth_flow.aclose()
 
 
+@pytest.mark.anyio
+async def test_interrupted_flow_cannot_pair_the_old_refresh_token_with_the_fresh_registration(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+):
+    """RFC 6749 §6 binds a refresh token to the client it was issued to, so the expiry
+    discard drops the token alongside the dead record — otherwise a flow failing between
+    re-registration and token exchange leaves fresh `client_info` paired with the old
+    client's refresh token, and the next request presents that token as the new client
+    (a guaranteed `invalid_grant` round trip).
+
+    Steps:
+    1. A registration whose secret lapsed mid-session sits next to an expired access
+       token and a refresh token issued to it.
+    2. The 401 flow discards the record — dropping the refresh token in memory and in
+       storage — and re-registers; the interactive authorization then fails (the user
+       abandons the consent) after the fresh registration was already persisted.
+    3. The next request attempts no refresh: with the orphan gone, `can_refresh_token()`
+       is false and the request goes out unauthenticated.
+    """
+    oauth_provider._initialized = True
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,  # lapsed after load
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+    oauth_provider.context.current_tokens = OAuthToken(access_token="stale", refresh_token="issued-to-dead-client")
+    oauth_provider.context.token_expiry_time = time.time() - 60  # access token expired
+
+    async def abandoned_consent(url: str) -> None:
+        raise RuntimeError("user closed the browser")
+
+    oauth_provider.context.redirect_handler = abandoned_consent
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # 401 → discovery; the lapsed record is discarded and the flow re-registers.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"issuer": "https://api.example.com", '
+            b'"authorization_endpoint": "https://api.example.com/authorize", '
+            b'"token_endpoint": "https://api.example.com/token", '
+            b'"registration_endpoint": "https://api.example.com/register"}'
+        ),
+        request=asm_req,
+    )
+    register_req = await auth_flow.asend(asm_response)
+    register_response = httpx2.Response(
+        201,
+        json={
+            "client_id": "fresh-client",
+            "client_secret": "fresh-secret",
+            "redirect_uris": ["http://localhost:3030/callback"],
+            "token_endpoint_auth_method": "client_secret_post",
+        },
+        request=register_req,
+    )
+
+    # The fresh registration is persisted, then the interactive authorization fails.
+    with pytest.raises(RuntimeError, match="user closed the browser"):
+        await auth_flow.asend(register_response)
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "fresh-client"
+
+    # The orphaned refresh token is gone from memory and storage; the access token survives.
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+    stored_tokens = await mock_storage.get_tokens()
+    assert stored_tokens is not None
+    assert stored_tokens.refresh_token is None
+    assert stored_tokens.access_token == "stale"
+
+    # The next request attempts no refresh: it goes out unauthenticated straight away.
+    retry_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    retry_request = await retry_flow.__anext__()
+    assert str(retry_request.url) == "https://api.example.com/v1/mcp"
+    assert "Authorization" not in retry_request.headers
+    await retry_flow.aclose()
+
+
 @pytest.mark.anyio
 async def test_403_step_up_re_registers_when_registration_secret_expired(
     oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
@@ -3525,7 +3638,8 @@ async def test_403_step_up_re_registers_when_registration_secret_expired(
     without its own re-check every step-up would complete a full interactive authorization
     only to fail ``invalid_client`` at the token exchange, until the access token itself
     expires. The step-up mints a fresh registration first — against the AS metadata already
-    discovered — and completes the exchange with the new credentials.
+    discovered — and completes the exchange with the new credentials; the dead client's
+    refresh token is dropped with its record.
     """
     oauth_provider._initialized = True
     oauth_provider.context.client_info = OAuthClientInformationFull(
@@ -3535,7 +3649,9 @@ async def test_403_step_up_re_registers_when_registration_secret_expired(
         redirect_uris=[AnyUrl("http://localhost:3030/callback")],
         token_endpoint_auth_method="client_secret_post",
     )
-    oauth_provider.context.current_tokens = OAuthToken(access_token="live-token", scope="read")
+    oauth_provider.context.current_tokens = OAuthToken(
+        access_token="live-token", refresh_token="issued-to-dead-client", scope="read"
+    )
     oauth_provider.context.token_expiry_time = time.time() + 1800  # access token still live
     oauth_provider.context.oauth_metadata = OAuthMetadata(
         issuer=AnyHttpUrl("https://auth.example.com"),
@@ -3571,6 +3687,14 @@ async def mock_callback() -> AuthorizationCodeResult:
     register_req = await auth_flow.asend(response_403)
     assert register_req.method == "POST"
     assert str(register_req.url) == "https://auth.example.com/register"
+    # The discard dropped the dead client's refresh token (in memory and in storage) while
+    # keeping the live access token.
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+    stored_tokens = await mock_storage.get_tokens()
+    assert stored_tokens is not None
+    assert stored_tokens.refresh_token is None
+    assert stored_tokens.access_token == "live-token"
     register_response = httpx2.Response(
         201,
         json={

From 32a056d4cfd51480f78f7c4b0b455ecea1993edf Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 10 Aug 2026 07:40:26 +0000
Subject: [PATCH 6/9] test(client/auth): make the expiry-discard test's storage
 assertions non-vacuous

MockTokenStorage stores tokens by reference and _initialize loads that
same object into context.current_tokens, so the discard's in-place
refresh-token trim reached the stored object even without the
set_tokens persistence call - the storage-side assertions in
test_expired_stored_registration_is_discarded_and_the_flow_re_registers
passed vacuously. Re-seed storage with a distinct copy after the load
so those assertions pass only if the discard actually persists the
trimmed tokens (verified by mutation: deleting the set_tokens call now
fails the test).
---
 tests/client/test_auth.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index c5dc7c5fd9..4887d950f4 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -3353,6 +3353,12 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
     assert oauth_provider.context.current_tokens is not None
     assert request.headers["Authorization"] == f"Bearer {valid_tokens.access_token}"
 
+    # `MockTokenStorage` stores by reference and `_initialize` loaded that same object into
+    # `context.current_tokens`, so the discard's in-place trim would reach the stored object
+    # even if it never persisted. Re-seed storage with a distinct copy so the storage-side
+    # assertions below pass only if the discard actually calls `set_tokens`.
+    await mock_storage.set_tokens(valid_tokens.model_copy())
+
     # Server rejects the stale token: the 401 flow re-registers instead of reusing the record.
     response_401 = httpx2.Response(401, request=request)
     prm_req = await auth_flow.asend(response_401)

From 43d9599d84353b4d9443c082e990ae573d5f3356 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 10 Aug 2026 08:06:02 +0000
Subject: [PATCH 7/9] fix(client/auth): run discovery before the 403 step-up
 re-registers when no AS metadata is cached
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

After a restart _initialize() restores tokens and client info but never AS
metadata, so a 403 insufficient_scope step-up that needs to re-register (lapsed
secret, or no stored record) previously registered blind: it bypassed a
configured CIMD URL (should_use_client_metadata_url is False without metadata)
and POSTed the registration document to the resource origin's /register
fallback — the wrong server in the standard separate-AS topology, failing every
step-up until the access token expired.

The 401 flow's discovery sequence (PRM + SEP-2352 issuer checks + ASM) now
lives in a _discover_authorization_server_metadata sub-generator shared by both
branches; the 403 step-up runs it before the expiry discard whenever
oauth_metadata is None and re-registration is coming, so registration targets
the discovered endpoint (or resolves CIMD). extract_resource_metadata_from_
www_auth now honors the resource_metadata parameter on 403 challenges too
(RFC 9728 attaches it to any WWW-Authenticate challenge), so the step-up's
discovery is seeded from the challenge header exactly like the 401 flow's.

Covered by two restart-shaped step-up tests: DCR against the discovered
registration endpoint, and CIMD resolution once discovery restores the
advertised capability.
---
 src/mcp/client/auth/oauth2.py | 213 ++++++++++++++++++++-------------
 src/mcp/client/auth/utils.py  |   5 +-
 tests/client/test_auth.py     | 218 ++++++++++++++++++++++++++++++++++
 3 files changed, 353 insertions(+), 83 deletions(-)

diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index 2dba6bc321..6d6e2529d9 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -678,6 +678,101 @@ async def _complete_client_registration(self, response: httpx2.Response) -> None
         self.context.client_info = client_information
         await self.context.storage.set_client_info(client_information)
 
+    async def _discover_authorization_server_metadata(
+        self, response: httpx2.Response
+    ) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+        """Discover protected resource and authorization server metadata.
+
+        Runs the discovery sequence shared by the 401 flow and the 403 step-up:
+        protected resource metadata (SEP-985, seeded from the challenge's
+        `resource_metadata` parameter when present), the SEP-2352 issuer checks on
+        stored credentials, and OAuth authorization server metadata. Yields each
+        discovery request; the caller must send the response back into the generator
+        (the httpx auth-flow protocol).
+        """
+        www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response)
+
+        # Step 1: Discover protected resource metadata (SEP-985 with fallback support)
+        prm_discovery_urls = build_protected_resource_metadata_discovery_urls(
+            www_auth_resource_metadata_url, self.context.server_url
+        )
+
+        for url in prm_discovery_urls:  # pragma: no branch
+            discovery_request = create_oauth_metadata_request(url)
+
+            discovery_response = yield discovery_request  # sending request
+
+            prm = await handle_protected_resource_response(discovery_response)
+            if prm:
+                # Validate PRM resource matches server URL (RFC 8707)
+                await self._validate_resource_match(prm)
+                self.context.protected_resource_metadata = prm
+
+                # todo: try all authorization_servers to find the OASM
+                assert (
+                    len(prm.authorization_servers) > 0
+                )  # this is always true as authorization_servers has a min length of 1
+
+                self.context.auth_server_url = str(prm.authorization_servers[0])
+                break
+            else:
+                logger.debug(f"Protected resource metadata discovery failed: {url}")
+
+        # SEP-2352: stored credentials are bound to the issuer that registered them.
+        # If the authorization server changed, drop them (and the old tokens) so the
+        # flow re-registers instead of presenting another server's credentials.
+        if (
+            self.context.client_info is not None
+            and self.context.auth_server_url is not None
+            and not credentials_match_issuer(
+                self.context.client_info, self.context.auth_server_url, self.context.client_metadata_url
+            )
+        ):
+            logger.debug("Authorization server changed; discarding bound credentials and re-registering")
+            self.context.client_info = None
+            self.context.clear_tokens()
+            # Any cached AS metadata is for the old server; drop it so a failed
+            # rediscovery cannot leak the old registration/token endpoints into Step 4.
+            self.context.oauth_metadata = None
+
+        asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
+            self.context.auth_server_url, self.context.server_url
+        )
+
+        # Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers)
+        for url in asm_discovery_urls:  # pragma: no branch
+            oauth_metadata_request = create_oauth_metadata_request(url)
+            oauth_metadata_response = yield oauth_metadata_request
+
+            ok, asm = await handle_auth_metadata_response(oauth_metadata_response)
+            if not ok:
+                break
+            if ok and asm:
+                # SEP-2468: metadata issuer must match the discovery issuer
+                if self.context.auth_server_url is not None:
+                    validate_metadata_issuer(asm, self.context.auth_server_url)
+                self.context.oauth_metadata = asm
+                break
+            else:
+                logger.debug(f"OAuth metadata discovery failed: {url}")
+
+        # SEP-2352: on the legacy no-PRM path the issuer is only known after ASM
+        # discovery, so re-evaluate the binding here using the discovered metadata
+        # issuer (mirroring the bound_issuer fallback in Step 4).
+        if (
+            self.context.client_info is not None
+            and self.context.auth_server_url is None
+            and self.context.oauth_metadata is not None
+            and not credentials_match_issuer(
+                self.context.client_info,
+                str(self.context.oauth_metadata.issuer),
+                self.context.client_metadata_url,
+            )
+        ):
+            logger.debug("Authorization server changed; discarding bound credentials and re-registering")
+            self.context.client_info = None
+            self.context.clear_tokens()
+
     async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
         """httpx2 auth flow integration."""
         async with self.context.lock:
@@ -713,88 +808,19 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                 try:
                     # OAuth flow must be inline due to generator constraints
 
-                    www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response)
-
-                    # Step 1: Discover protected resource metadata (SEP-985 with fallback support)
-                    prm_discovery_urls = build_protected_resource_metadata_discovery_urls(
-                        www_auth_resource_metadata_url, self.context.server_url
-                    )
-
-                    for url in prm_discovery_urls:  # pragma: no branch
-                        discovery_request = create_oauth_metadata_request(url)
-
-                        discovery_response = yield discovery_request  # sending request
-
-                        prm = await handle_protected_resource_response(discovery_response)
-                        if prm:
-                            # Validate PRM resource matches server URL (RFC 8707)
-                            await self._validate_resource_match(prm)
-                            self.context.protected_resource_metadata = prm
-
-                            # todo: try all authorization_servers to find the OASM
-                            assert (
-                                len(prm.authorization_servers) > 0
-                            )  # this is always true as authorization_servers has a min length of 1
-
-                            self.context.auth_server_url = str(prm.authorization_servers[0])
-                            break
-                        else:
-                            logger.debug(f"Protected resource metadata discovery failed: {url}")
-
-                    # SEP-2352: stored credentials are bound to the issuer that registered them.
-                    # If the authorization server changed, drop them (and the old tokens) so the
-                    # flow re-registers instead of presenting another server's credentials.
-                    if (
-                        self.context.client_info is not None
-                        and self.context.auth_server_url is not None
-                        and not credentials_match_issuer(
-                            self.context.client_info, self.context.auth_server_url, self.context.client_metadata_url
-                        )
-                    ):
-                        logger.debug("Authorization server changed; discarding bound credentials and re-registering")
-                        self.context.client_info = None
-                        self.context.clear_tokens()
-                        # Any cached AS metadata is for the old server; drop it so a failed
-                        # rediscovery cannot leak the old registration/token endpoints into Step 4.
-                        self.context.oauth_metadata = None
-
-                    asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
-                        self.context.auth_server_url, self.context.server_url
-                    )
-
-                    # Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers)
-                    for url in asm_discovery_urls:  # pragma: no branch
-                        oauth_metadata_request = create_oauth_metadata_request(url)
-                        oauth_metadata_response = yield oauth_metadata_request
-
-                        ok, asm = await handle_auth_metadata_response(oauth_metadata_response)
-                        if not ok:
-                            break
-                        if ok and asm:
-                            # SEP-2468: metadata issuer must match the discovery issuer
-                            if self.context.auth_server_url is not None:
-                                validate_metadata_issuer(asm, self.context.auth_server_url)
-                            self.context.oauth_metadata = asm
-                            break
-                        else:
-                            logger.debug(f"OAuth metadata discovery failed: {url}")
-
-                    # SEP-2352: on the legacy no-PRM path the issuer is only known after ASM
-                    # discovery, so re-evaluate the binding here using the discovered metadata
-                    # issuer (mirroring the bound_issuer fallback in Step 4).
-                    if (
-                        self.context.client_info is not None
-                        and self.context.auth_server_url is None
-                        and self.context.oauth_metadata is not None
-                        and not credentials_match_issuer(
-                            self.context.client_info,
-                            str(self.context.oauth_metadata.issuer),
-                            self.context.client_metadata_url,
-                        )
-                    ):
-                        logger.debug("Authorization server changed; discarding bound credentials and re-registering")
-                        self.context.client_info = None
-                        self.context.clear_tokens()
+                    # Steps 1-2: Discover protected resource and authorization server
+                    # metadata, applying the SEP-2352 issuer checks along the way. The
+                    # sequence lives in a sub-generator shared with the 403 step-up;
+                    # its requests are relayed by hand (`yield from` cannot cross an
+                    # async generator).
+                    discovery = self._discover_authorization_server_metadata(response)
+                    try:
+                        discovery_request = await anext(discovery)
+                        while True:
+                            discovery_response = yield discovery_request
+                            discovery_request = await discovery.asend(discovery_response)
+                    except StopAsyncIteration:
+                        pass
 
                     # Step 3: Apply scope selection strategy
                     self.context.client_metadata.scope = get_client_metadata_scopes(
@@ -844,6 +870,29 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                 # Step 2: Check if we need to step-up authorization
                 if error == "insufficient_scope":  # pragma: no branch
                     try:
+                        # After a restart the 403 step-up can be the first auth event:
+                        # `_initialize` restores tokens and client info but never AS
+                        # metadata, and the still-live access token keeps the 401 flow
+                        # (and its discovery) from running. When this step-up will need
+                        # to re-register — the stored secret lapsed, or no record is
+                        # stored at all — discover the metadata first instead of
+                        # registering blind: the blind fallback would bypass a
+                        # configured CIMD URL and POST the registration document to the
+                        # resource origin's `/register`, the wrong server in the
+                        # separate-AS topology. Running it before the expiry discard
+                        # lets the SEP-2352 issuer checks still see the record's stamp.
+                        if self.context.oauth_metadata is None and (
+                            self.context.registration_secret_expired() or self.context.client_info is None
+                        ):
+                            discovery = self._discover_authorization_server_metadata(response)
+                            try:
+                                discovery_request = await anext(discovery)
+                                while True:
+                                    discovery_response = yield discovery_request
+                                    discovery_request = await discovery.asend(discovery_response)
+                            except StopAsyncIteration:
+                                pass
+
                         # Step 2a: Union previously requested scopes with the newly challenged
                         # 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
diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py
index 31e2e5cade..5210e821a3 100644
--- a/src/mcp/client/auth/utils.py
+++ b/src/mcp/client/auth/utils.py
@@ -51,10 +51,13 @@ def extract_scope_from_www_auth(response: Response) -> str | None:
 def extract_resource_metadata_from_www_auth(response: Response) -> str | None:
     """Extract protected resource metadata URL from WWW-Authenticate header as per RFC 9728.
 
+    RFC 9728 attaches the `resource_metadata` parameter to any WWW-Authenticate
+    challenge, so both 401 and 403 (scope step-up) responses are honored.
+
     Returns:
         Resource metadata URL if found in WWW-Authenticate header, None otherwise
     """
-    if not response or response.status_code != 401:
+    if not response or response.status_code not in (401, 403):
         return None  # pragma: no cover
 
     return extract_field_from_www_auth(response, "resource_metadata")
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index 4887d950f4..d19773fa8d 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -3736,6 +3736,224 @@ async def mock_callback() -> AuthorizationCodeResult:
         pass
 
 
+@pytest.mark.anyio
+async def test_403_step_up_discovers_the_as_before_re_registering_after_a_restart(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+):
+    """A 403 step-up that must re-register with no AS metadata cached discovers it first.
+
+    After a restart `_initialize` restores tokens and client info but no AS metadata, and
+    the still-live access token keeps the 401 flow's discovery from ever running. Without
+    its own discovery the step-up would POST the registration document to the resource
+    origin's `/register` fallback — the wrong server in the separate-AS topology — so it
+    runs the 401 flow's discovery sequence first, honoring the 403's `resource_metadata`
+    pointer, and registers at the discovered endpoint.
+    """
+    await mock_storage.set_client_info(
+        OAuthClientInformationFull(
+            client_id="dead-client",
+            client_secret="expired-secret",
+            client_secret_expires_at=int(time.time()) - 3600,  # lapsed while stored
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_post",
+            issuer="https://auth.example.com",
+        )
+    )
+    await mock_storage.set_tokens(
+        OAuthToken(access_token="live-token", refresh_token="issued-to-dead-client", scope="read")
+    )
+
+    captured_state: str | None = None
+    authorize_client_id: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state, authorize_client_id
+        params = parse_qs(urlparse(url).query)
+        authorize_client_id = params["client_id"][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
+
+    # The restarted provider initializes from storage: the loaded token records no expiry,
+    # so it is presented as-is and the 401 flow never runs.
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    assert request.headers["Authorization"] == "Bearer live-token"
+
+    response_403 = httpx2.Response(
+        403,
+        headers={
+            "WWW-Authenticate": (
+                'Bearer error="insufficient_scope", scope="write", '
+                'resource_metadata="https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"'
+            )
+        },
+        request=request,
+    )
+
+    # The step-up discovers the PRM first, seeded from the 403's resource_metadata pointer.
+    prm_req = await auth_flow.asend(response_403)
+    assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
+    prm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+
+    # Then the AS metadata.
+    asm_req = await auth_flow.asend(prm_response)
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
+    asm_response = httpx2.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'"registration_endpoint": "https://auth.example.com/register"}'
+        ),
+        request=asm_req,
+    )
+
+    # The registration goes to the discovered endpoint, not the resource origin's /register.
+    register_req = await auth_flow.asend(asm_response)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://auth.example.com/register"
+    register_response = httpx2.Response(
+        201,
+        json={
+            "client_id": "fresh-client",
+            "client_secret": "fresh-secret",
+            "redirect_uris": ["http://localhost:3030/callback"],
+            "token_endpoint_auth_method": "client_secret_post",
+        },
+        request=register_req,
+    )
+
+    # The step-up completes with the fresh credentials; the dead client's refresh token
+    # was dropped with its record.
+    token_exchange_request = await auth_flow.asend(register_response)
+    assert authorize_client_id == "fresh-client"
+    content = token_exchange_request.content.decode()
+    assert "client_id=fresh-client" in content
+    assert "expired-secret" not in content
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_403_step_up_after_a_restart_resolves_cimd_once_metadata_is_discovered(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+):
+    """A 403 step-up with a configured CIMD URL and no cached AS metadata still uses CIMD.
+
+    Registering blind would silently bypass the CIMD capability the (undiscovered) AS
+    metadata advertises; the step-up's discovery restores it, so the dead record is
+    replaced locally by the URL-based client ID — no `/register` round trip at all.
+    """
+    captured_state: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state
+        captured_state = parse_qs(urlparse(url).query).get("state", [None])[0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+        redirect_handler=capture_redirect,
+        callback_handler=mock_callback,
+        client_metadata_url="https://example.com/client",
+    )
+    await mock_storage.set_client_info(
+        OAuthClientInformationFull(
+            client_id="dead-client",
+            client_secret="expired-secret",
+            client_secret_expires_at=int(time.time()) - 3600,  # lapsed while stored
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_post",
+            issuer="https://auth.example.com",
+        )
+    )
+    await mock_storage.set_tokens(OAuthToken(access_token="live-token", scope="read"))
+
+    auth_flow = provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    assert request.headers["Authorization"] == "Bearer live-token"
+
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # No resource_metadata pointer this time: discovery falls back to the well-known PRM URL.
+    prm_req = await auth_flow.asend(response_403)
+    assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
+    prm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+    asm_req = await auth_flow.asend(prm_response)
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
+    asm_response = httpx2.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'"client_id_metadata_document_supported": true}'
+        ),
+        request=asm_req,
+    )
+
+    # No /register round trip: the next request is already the token exchange, and the
+    # dead record was replaced by the URL-based client ID.
+    token_exchange_request = await auth_flow.asend(asm_response)
+    assert str(token_exchange_request.url) == "https://auth.example.com/token"
+    content = token_exchange_request.content.decode()
+    assert "client_id=https%3A%2F%2Fexample.com%2Fclient" in content
+    assert "expired-secret" not in content
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "https://example.com/client"  # dead record overwritten in storage
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass
+
+
 @pytest.mark.anyio
 async def test_403_step_up_resolves_cimd_when_registration_secret_expired(
     client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage

From aedc88c780e043e3dede817bcd2525cfd456e5dd Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 10 Aug 2026 08:33:08 +0000
Subject: [PATCH 8/9] fix(client/auth): finalize the discovery sub-generator
 when the flow is closed mid-discovery
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

httpx2's _send_handling_auth acloses the auth flow on any transport error or
cancellation, throwing GeneratorExit at the relay's yield — caught by neither
except StopAsyncIteration nor except Exception — so both hand-relay sites (the
401 branch and the 403 step-up) abandoned the suspended
_discover_authorization_server_metadata sub-generator to GC (a ResourceWarning
on trio, which filterwarnings=["error"] turns into a test failure for any
future abort-mid-discovery test).

Both relays now drive the sub-generator under contextlib.aclosing, matching the
eager-refresh relay in #3263, so closing the outer flow finalizes it in the
same unwind. Covered by two regression tests that close the flow mid-discovery
and assert the captured sub-generator reports exhaustion instead of a live
suspended frame.
---
 src/mcp/client/auth/oauth2.py | 39 +++++++++-------
 tests/client/test_auth.py     | 83 +++++++++++++++++++++++++++++++++++
 2 files changed, 105 insertions(+), 17 deletions(-)

diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index 6d6e2529d9..05f5b07534 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -10,6 +10,7 @@
 import string
 import time
 from collections.abc import AsyncGenerator, Awaitable, Callable
+from contextlib import aclosing
 from dataclasses import dataclass, field
 from typing import Any, Protocol, get_args
 from urllib.parse import quote, urlencode, urljoin, urlparse
@@ -812,15 +813,17 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                     # metadata, applying the SEP-2352 issuer checks along the way. The
                     # sequence lives in a sub-generator shared with the 403 step-up;
                     # its requests are relayed by hand (`yield from` cannot cross an
-                    # async generator).
-                    discovery = self._discover_authorization_server_metadata(response)
-                    try:
-                        discovery_request = await anext(discovery)
-                        while True:
-                            discovery_response = yield discovery_request
-                            discovery_request = await discovery.asend(discovery_response)
-                    except StopAsyncIteration:
-                        pass
+                    # async generator). `aclosing` finalizes the sub-generator when
+                    # httpx2 closes this flow mid-discovery (transport error or
+                    # cancellation throws `GeneratorExit` at the relay's `yield`).
+                    async with aclosing(self._discover_authorization_server_metadata(response)) as discovery:
+                        try:
+                            discovery_request = await anext(discovery)
+                            while True:
+                                discovery_response = yield discovery_request
+                                discovery_request = await discovery.asend(discovery_response)
+                        except StopAsyncIteration:
+                            pass
 
                     # Step 3: Apply scope selection strategy
                     self.context.client_metadata.scope = get_client_metadata_scopes(
@@ -884,14 +887,16 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         if self.context.oauth_metadata is None and (
                             self.context.registration_secret_expired() or self.context.client_info is None
                         ):
-                            discovery = self._discover_authorization_server_metadata(response)
-                            try:
-                                discovery_request = await anext(discovery)
-                                while True:
-                                    discovery_response = yield discovery_request
-                                    discovery_request = await discovery.asend(discovery_response)
-                            except StopAsyncIteration:
-                                pass
+                            # `aclosing` mirrors the 401 relay above: it finalizes the
+                            # sub-generator when httpx2 closes this flow mid-discovery.
+                            async with aclosing(self._discover_authorization_server_metadata(response)) as discovery:
+                                try:
+                                    discovery_request = await anext(discovery)
+                                    while True:
+                                        discovery_response = yield discovery_request
+                                        discovery_request = await discovery.asend(discovery_response)
+                                except StopAsyncIteration:
+                                    pass
 
                         # Step 2a: Union previously requested scopes with the newly challenged
                         # scopes (SEP-2350) so escalating one operation keeps the others' grants.
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index d19773fa8d..69cec9c3f4 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -3,6 +3,7 @@
 import base64
 import json
 import time
+from collections.abc import AsyncGenerator
 from unittest import mock
 from urllib.parse import parse_qs, quote, unquote, urlparse
 
@@ -4031,3 +4032,85 @@ async def mock_callback() -> AuthorizationCodeResult:
         await auth_flow.asend(httpx2.Response(200, request=final_request))
     except StopAsyncIteration:
         pass
+
+
+@pytest.mark.anyio
+async def test_closing_the_flow_mid_discovery_finalizes_the_401_discovery_sub_generator(
+    oauth_provider: OAuthClientProvider,
+):
+    """httpx2's `_send_handling_auth` acloses the auth flow when a discovery request
+    fails at the transport level (or the request is cancelled), throwing `GeneratorExit`
+    at the 401 relay's `yield`; the relay must finalize the discovery sub-generator with
+    the flow rather than abandon it suspended to GC (a `ResourceWarning` under trio).
+    The sub-generator is captured via a wrapper because its finalization is not
+    observable through the auth-flow protocol itself.
+    """
+    captured: list[AsyncGenerator[httpx2.Request, httpx2.Response]] = []
+    original = oauth_provider._discover_authorization_server_metadata
+
+    def capturing(response: httpx2.Response) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+        discovery = original(response)
+        captured.append(discovery)
+        return discovery
+
+    oauth_provider._discover_authorization_server_metadata = capturing
+    oauth_provider._initialized = True
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # 401 → the flow yields the first discovery request, suspending both generators mid-relay.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    assert "oauth-protected-resource" in str(prm_req.url)
+    assert len(captured) == 1
+
+    # Closing the flow here mirrors httpx2 aborting it on a transport error.
+    await auth_flow.aclose()
+
+    # The sub-generator was closed with the flow: probing it reports exhaustion instead
+    # of resuming a suspended frame.
+    with pytest.raises(StopAsyncIteration):
+        await captured[0].__anext__()
+
+
+@pytest.mark.anyio
+async def test_closing_the_flow_mid_discovery_finalizes_the_403_step_up_discovery_sub_generator(
+    oauth_provider: OAuthClientProvider,
+):
+    """The 403 step-up's discovery relay must finalize its sub-generator when httpx2
+    acloses the flow mid-discovery, exactly like the 401 relay (same abandonment
+    hazard, same fix); the sub-generator is captured via a wrapper because its
+    finalization is not observable through the auth-flow protocol itself.
+    """
+    captured: list[AsyncGenerator[httpx2.Request, httpx2.Response]] = []
+    original = oauth_provider._discover_authorization_server_metadata
+
+    def capturing(response: httpx2.Response) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+        discovery = original(response)
+        captured.append(discovery)
+        return discovery
+
+    oauth_provider._discover_authorization_server_metadata = capturing
+    # Restart shape: a live token and no stored registration, so the 403 step-up must
+    # discover before registering (`oauth_metadata` is never restored by `_initialize`).
+    oauth_provider.context.current_tokens = OAuthToken(access_token="live-token")
+    oauth_provider._initialized = True
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # 403 → the step-up yields the first discovery request, suspending both generators mid-relay.
+    prm_req = await auth_flow.asend(response_403)
+    assert "oauth-protected-resource" in str(prm_req.url)
+    assert len(captured) == 1
+
+    await auth_flow.aclose()
+
+    with pytest.raises(StopAsyncIteration):
+        await captured[0].__anext__()

From 928b7bf262dce322ff2003a5fd02ba135758646f Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 10 Aug 2026 08:56:38 +0000
Subject: [PATCH 9/9] test(client/auth): cover the 403 step-up's exception
 relay; drop its stale no-cover pragma

The mid-discovery close test traverses the 403 branch's except clause
(GeneratorExit evaluates the match), so strict-no-cover now flags the
'pragma: no cover' on that line as wrongly marked. Cover the handler
properly instead: a new test drives an OAuthFlowError (RFC 8707 resource
mismatch during step-up discovery) through the branch, asserting it
propagates to the caller like the 401 branch's error contract, and the
pragma is removed.
---
 src/mcp/client/auth/oauth2.py |  2 +-
 tests/client/test_auth.py     | 34 ++++++++++++++++++++++++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index 05f5b07534..d5718ac4b9 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -933,7 +933,7 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         # Step 2b: Perform (re-)authorization and token exchange
                         token_response = yield await self._perform_authorization()
                         await self._handle_token_response(token_response)
-                    except Exception:  # pragma: no cover
+                    except Exception:
                         logger.exception("OAuth flow error")
                         raise
 
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index 69cec9c3f4..a209020902 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -4114,3 +4114,37 @@ def capturing(response: httpx2.Response) -> AsyncGenerator[httpx2.Request, httpx
 
     with pytest.raises(StopAsyncIteration):
         await captured[0].__anext__()
+
+
+@pytest.mark.anyio
+async def test_403_step_up_surfaces_oauth_flow_errors_to_the_caller(
+    oauth_provider: OAuthClientProvider,
+):
+    """An `OAuthFlowError` raised inside the 403 step-up — here the SEP-985/RFC 8707
+    resource-identity check failing during the step-up's discovery — propagates to the
+    caller instead of being swallowed, matching the 401 branch's error contract.
+    """
+    # Restart shape: a live token and no stored registration, so the step-up discovers.
+    oauth_provider.context.current_tokens = OAuthToken(access_token="live-token")
+    oauth_provider._initialized = True
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+    prm_req = await auth_flow.asend(response_403)
+
+    # The PRM advertises a different resource, so the RFC 8707 identity check must fail.
+    mismatched_prm = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://other.example.com/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+    with pytest.raises(OAuthFlowError):
+        await auth_flow.asend(mismatched_prm)