OAuth client: refresh before re-authorizing, and discover before refreshing

The 401 branch of OAuthClientProvider discovered metadata but never tried
the stored refresh token, and the pre-request refresh tried the refresh
token but never discovered metadata. After a restart nothing restores an
expiry, so the pre-request path is skipped, the stale bearer draws a 401,
and the flow went straight to interactive authorization with a usable
refresh token in hand; headless clients failed outright. When an
application forced the pre-request path, the refresh was posted to a
path guessed from the server origin, which 404s against an
authorization server under a path, and the refresh token was dropped.

The 401 branch now tries the refresh_token grant after discovery and
registration and runs the full authorization only when there is no
refresh token or the server rejects it. The pre-request refresh runs
only when authorization server metadata is already known, so it never
guesses an endpoint; a cold start takes the 401 and refreshes there. A
fresh dynamic registration clears any held tokens, which belonged to a
previous client.

Closes #3240, #3250, #1318.

Author: maxisbey

docs/client/oauth-clients.md

@@ -81,7 +81,7 @@ The first time `Client` sends a request, the server answers `401`. The provider
 3. **Authorization.** It generates the PKCE pair and a `state`, builds the authorization URL, awaits your `redirect_handler`, then awaits your `callback_handler` for the code.
 4. **Exchange.** It trades the code for an `OAuthToken`, stores it, and replays your original request with `Authorization: Bearer ...`.
 
-After that it is quiet. Tokens come out of storage, an expired access token is refreshed with the refresh token, and only when none of that works does it run the flow again.
+After that it is quiet. Tokens come out of storage, an expired access token is refreshed with the refresh token, and only when none of that works does it run the flow again. That holds across restarts: a new process that finds a refresh token in storage answers the first `401` by rediscovering the authorization server and refreshing, not by sending anyone back to the browser.
 
 You wrote none of it. Two keyword arguments remain (`client_metadata_url` and `validate_resource_url`), and this file needs neither. `client_metadata_url` is the one worth knowing about; it gets its own section below.
 

src/mcp/client/auth/oauth2.py

@@ -586,8 +586,13 @@ 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():
-                # Try to refresh token
+            # Refresh ahead of the request only when the token endpoint is already known; on a cold
+            # start the request goes out and the 401 branch discovers, then refreshes.
+            if (
+                not self.context.is_token_valid()
+                and self.context.can_refresh_token()
+                and self.context.oauth_metadata is not None
+            ):
                 refresh_request = await self._refresh_token()
                 refresh_response = yield refresh_request
 
@@ -741,10 +746,18 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                                 client_information.issuer = discovered_issuer
                             self.context.client_info = client_information
                             await self.context.storage.set_client_info(client_information)
-
-                    # Step 5: Perform authorization and complete token exchange
-                    token_response = yield await self._perform_authorization()
-                    await self._handle_token_response(token_response)
+                            # Held tokens belong to a previous client and cannot be refreshed by this one.
+                            self.context.clear_tokens()
+
+                    # Step 5: Refresh with the stored refresh token first (RFC 6749 §6); run the full
+                    # authorization only when there is none or the server rejects it.
+                    refreshed = False
+                    if self.context.can_refresh_token():
+                        refresh_response = yield await self._refresh_token()
+                        refreshed = await self._handle_refresh_response(refresh_response)
+                    if not refreshed:
+                        token_response = yield await self._perform_authorization()
+                        await self._handle_token_response(token_response)
                 except Exception:
                     logger.exception("OAuth flow error")
                     raise

tests/client/test_auth.py

@@ -3253,3 +3253,29 @@ async def echo_callback() -> AuthorizationCodeResult:
         await auth_flow.asend(httpx2.Response(200, request=final_req))
     except StopAsyncIteration:
         pass
+
+
+@pytest.mark.anyio
+async def test_expired_token_is_not_refreshed_ahead_of_the_request_before_metadata_is_discovered(
+    oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
+) -> None:
+    """With no authorization-server metadata yet, an expired token is not refreshed at a guessed endpoint.
+
+    The request goes out unauthenticated instead, so the 401 branch discovers the real token
+    endpoint before the refresh token is presented anywhere (#3240).
+    """
+    oauth_provider.context.current_tokens = valid_tokens
+    oauth_provider.context.token_expiry_time = time.time() - 60
+    oauth_provider.context.client_info = OAuthClientInformationFull(client_id="c", redirect_uris=None)
+    oauth_provider.context.oauth_metadata = None
+    oauth_provider._initialized = True
+
+    request = httpx2.Request("POST", "https://api.example.com/v1/mcp")
+    auth_flow = oauth_provider.async_auth_flow(request)
+    first = await auth_flow.__anext__()
+
+    assert first is request
+    assert "Authorization" not in first.headers
+
+    with pytest.raises(StopAsyncIteration):
+        await auth_flow.asend(httpx2.Response(200, request=request))

tests/interaction/_requirements.py

@@ -3883,6 +3883,26 @@ def __post_init__(self) -> None:
         transports=("streamable-http",),
         note="OAuth is HTTP-only.",
     ),
+    "client-auth:refresh:on-401": Requirement(
+        source="issue:#3250",
+        behavior=(
+            "A 401 received while a refresh token is held is answered, after rediscovery, with a "
+            "refresh_token grant before any interactive authorization, so a client constructed over "
+            "persisted tokens and client registration recovers from an expired access token headlessly."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only. RFC 6749 §1.5 (E)-(H); matches the TypeScript, C# and Rust SDKs.",
+    ),
+    "client-auth:refresh:discovered-endpoint": Requirement(
+        source="issue:#3240",
+        behavior=(
+            "A refresh in a process that has not yet discovered the authorization server happens only after "
+            "protected-resource and authorization-server metadata discovery and posts to the advertised "
+            "token endpoint, never to a path guessed from the server origin."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
     "client-auth:resource-parameter": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#resource-parameter-implementation",
         behavior=(

tests/interaction/auth/_harness.py

@@ -26,7 +26,14 @@
 from mcp.server import Server
 from mcp.server.auth.provider import AccessToken, ProviderTokenVerifier
 from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions
-from mcp.shared.auth import AuthorizationCodeResult, OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
+from mcp.shared.auth import (
+    AuthorizationCodeResult,
+    OAuthClientInformationFull,
+    OAuthClientMetadata,
+    OAuthMetadata,
+    OAuthToken,
+    ProtectedResourceMetadata,
+)
 from tests.interaction._connect import BASE_URL, NO_DNS_REBINDING_PROTECTION
 from tests.interaction.auth._provider import InMemoryAuthorizationServerProvider
 from tests.interaction.transports._bridge import StreamingASGITransport
@@ -273,6 +280,53 @@ def shim(
     return lambda app: shimmed_app(app, not_found=not_found, serve=serve)
 
 
+def path_prefixed_as_shim(prefix: str) -> AppShim:
+    """Build an `app_shim` that presents the co-hosted authorization server as living under `prefix`.
+
+    The SDK server mounts `/authorize`, `/token` and `/register` at the origin root whatever the
+    issuer, so an AS whose endpoints sit under a path cannot be configured natively. This serves
+    PRM naming `{BASE_URL}{prefix}` as the AS, serves that issuer's metadata at the RFC 8414
+    path-inserted well-known URL with every endpoint under the prefix, forwards `{prefix}/x` to the
+    real `/x`, and 404s the bare root endpoints and root metadata so a client guessing origin-root
+    paths fails as it would against such a server. Pair with
+    `InMemoryAuthorizationServerProvider(issuer=f"{BASE_URL}{prefix}")` so the redirect `iss` matches.
+    """
+    issuer = f"{BASE_URL}{prefix}"
+    prm = ProtectedResourceMetadata(resource=AnyHttpUrl(f"{BASE_URL}/mcp"), authorization_servers=[AnyHttpUrl(issuer)])
+    asm = OAuthMetadata(
+        issuer=AnyHttpUrl(issuer),
+        authorization_endpoint=AnyHttpUrl(f"{issuer}/authorize"),
+        token_endpoint=AnyHttpUrl(f"{issuer}/token"),
+        registration_endpoint=AnyHttpUrl(f"{issuer}/register"),
+        scopes_supported=["mcp"],
+        response_types_supported=["code"],
+        grant_types_supported=["authorization_code", "refresh_token"],
+        token_endpoint_auth_methods_supported=["client_secret_post", "client_secret_basic", "none"],
+        code_challenge_methods_supported=["S256"],
+    )
+
+    def factory(app: ASGIApp) -> ASGIApp:
+        inner = shimmed_app(
+            app,
+            not_found=frozenset({"/token", "/authorize", "/register", "/.well-known/oauth-authorization-server"}),
+            serve={
+                "/.well-known/oauth-protected-resource/mcp": metadata_body(prm),
+                f"/.well-known/oauth-authorization-server{prefix}": metadata_body(asm),
+            },
+        )
+
+        async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
+            if scope["type"] == "http" and scope["path"].startswith(f"{prefix}/"):
+                path = scope["path"][len(prefix) :]
+                await app({**scope, "path": path, "raw_path": path.encode()}, receive, send)
+                return
+            await inner(scope, receive, send)
+
+        return wrapped
+
+    return factory
+
+
 @dataclass
 class _FirstChallenge:
     """ASGI shim that answers the first request to a path with 401 + a given WWW-Authenticate.

tests/interaction/auth/_provider.py

@@ -103,6 +103,10 @@ def mint_access_token(self, *, client_id: str, scopes: list[str], resource: str
         )
         return access
 
+    def expire_access_token(self, token: str) -> None:
+        """Move an issued access token's server-side expiry into the past so the bearer middleware 401s it."""
+        self.access_tokens[token] = self.access_tokens[token].model_copy(update={"expires_at": int(time.time()) - 1})
+
     async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
         return self.clients.get(client_id)
 

tests/interaction/auth/test_lifecycle.py

@@ -18,19 +18,23 @@
 from pydantic import AnyHttpUrl, AnyUrl
 
 from mcp import MCPError
+from mcp.client.auth import OAuthClientProvider
 from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider
 from mcp.server import Server, ServerRequestContext
 from mcp.shared.auth import OAuthClientInformationFull, OAuthMetadata
 from tests.interaction._connect import BASE_URL
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import (
     REDIRECT_URI,
+    AppShim,
     InMemoryTokenStorage,
     RecordedRequest,
     auth_settings,
     connect_with_oauth,
     m2m_token_shim,
     metadata_body,
+    oauth_client_metadata,
+    path_prefixed_as_shim,
     record_requests,
     shim,
     step_up_shim,
@@ -98,6 +102,29 @@ def seeded_client(provider: InMemoryAuthorizationServerProvider, **kwargs: objec
     return info
 
 
+async def first_process_login(
+    provider: InMemoryAuthorizationServerProvider, storage: InMemoryTokenStorage, *, app_shim: AppShim | None = None
+) -> str:
+    """Run one interactive connect so `storage` holds what a first process leaves behind; return its access token.
+
+    The restart tests then build a fresh `OAuthClientProvider` over the same storage, as a second
+    process would, so the registration and tokens carry exactly what the SDK persists.
+    """
+    server = Server("guarded", on_list_tools=list_tools)
+    async with connect_with_oauth(server, provider=provider, storage=storage, app_shim=app_shim) as (client, _):
+        await client.list_tools()
+    assert storage.tokens is not None and storage.tokens.refresh_token is not None
+    return storage.tokens.access_token
+
+
+def restarted_headless_provider(storage: InMemoryTokenStorage) -> OAuthClientProvider:
+    """A provider as a second, headless process constructs it: same storage, fresh state, no handlers.
+
+    Reaching the interactive step raises rather than opening a browser the scenario says is absent.
+    """
+    return OAuthClientProvider(server_url=f"{BASE_URL}/mcp", client_metadata=oauth_client_metadata(), storage=storage)
+
+
 @requirement("client-auth:refresh:transparent")
 async def test_an_expired_access_token_is_transparently_refreshed_before_the_next_request() -> None:
     """An access token the client considers expired is refreshed and the new bearer is used.
@@ -354,6 +381,91 @@ async def test_a_failed_refresh_clears_stored_tokens_and_restarts_the_full_flow(
     assert storage.tokens.access_token in provider.access_tokens
 
 
+@requirement("client-auth:refresh:on-401")
+async def test_a_restarted_client_answers_a_401_with_its_stored_refresh_token() -> None:
+    """A second process holding only persisted tokens and registration refreshes on 401 instead of re-authorizing.
+
+    Steps: (1) a first process logs in and its storage keeps the registration and a refresh token;
+    (2) the server-side access token lapses; (3) a fresh provider over the same storage, with no
+    browser, connects. The recording proves the stale bearer drew a 401, discovery ran, one
+    `refresh_token` grant followed, and neither `/authorize` nor `/register` was touched.
+    SDK behaviour per RFC 6749 §1.5; regression bar for #3250 / #1318.
+    """
+    provider = InMemoryAuthorizationServerProvider()
+    storage = InMemoryTokenStorage()
+    with anyio.fail_after(5):
+        stale_access_token = await first_process_login(provider, storage)
+    provider.expire_access_token(stale_access_token)
+
+    recorded, on_request = record_requests()
+    server = Server("guarded", on_list_tools=list_tools)
+    with anyio.fail_after(5):
+        async with connect_with_oauth(
+            server, provider=provider, auth=restarted_headless_provider(storage), on_request=on_request
+        ) as (client, _):
+            result = await client.list_tools()
+
+    assert result.tools[0].name == "echo"
+    assert [(r.method, r.path) for r in recorded[:5]] == snapshot(
+        [
+            ("POST", "/mcp"),
+            ("GET", "/.well-known/oauth-protected-resource/mcp"),
+            ("GET", "/.well-known/oauth-authorization-server"),
+            ("POST", "/token"),
+            ("POST", "/mcp"),
+        ]
+    )
+    assert recorded[0].headers["authorization"] == f"Bearer {stale_access_token}"
+    assert [form_body(r)["grant_type"] for r in find(recorded, "POST", "/token")] == ["refresh_token"]
+    assert find(recorded, "GET", "/authorize") == [] and find(recorded, "POST", "/register") == []
+    assert storage.tokens is not None and storage.tokens.access_token != stale_access_token
+    assert storage.tokens.access_token in provider.access_tokens
+
+
+@requirement("client-auth:refresh:discovered-endpoint")
+async def test_a_restarted_client_refreshes_at_the_token_endpoint_advertised_under_a_path() -> None:
+    """Against an authorization server under `/oauth2/v1`, a second process refreshes at `/oauth2/v1/token`.
+
+    The bare `/token` 404s here. Nothing is discovered yet in the second process, so no refresh is
+    attempted before the request; the 401 drives discovery and the single refresh POST goes to the
+    advertised endpoint. Regression bar for #3240, where the guessed `{origin}/token` 404ed and the
+    refresh token was discarded.
+    """
+    prefix = "/oauth2/v1"
+    provider = InMemoryAuthorizationServerProvider(issuer=f"{BASE_URL}{prefix}")
+    storage = InMemoryTokenStorage()
+    app_shim = path_prefixed_as_shim(prefix)
+    with anyio.fail_after(5):
+        stale_access_token = await first_process_login(provider, storage, app_shim=app_shim)
+    provider.expire_access_token(stale_access_token)
+
+    recorded, on_request = record_requests()
+    server = Server("guarded", on_list_tools=list_tools)
+    with anyio.fail_after(5):
+        async with connect_with_oauth(
+            server,
+            provider=provider,
+            auth=restarted_headless_provider(storage),
+            app_shim=app_shim,
+            on_request=on_request,
+        ) as (client, _):
+            result = await client.list_tools()
+
+    assert result.tools[0].name == "echo"
+    assert [(r.method, r.path) for r in recorded[:5]] == snapshot(
+        [
+            ("POST", "/mcp"),
+            ("GET", "/.well-known/oauth-protected-resource/mcp"),
+            ("GET", "/.well-known/oauth-authorization-server/oauth2/v1"),
+            ("POST", "/oauth2/v1/token"),
+            ("POST", "/mcp"),
+        ]
+    )
+    token_posts = [r for r in recorded if r.method == "POST" and r.path.endswith("/token")]
+    assert [(r.path, form_body(r)["grant_type"]) for r in token_posts] == [("/oauth2/v1/token", "refresh_token")]
+    assert not any(r.path.endswith("/authorize") for r in recorded)
+
+
 @requirement("client-auth:client-credentials")
 async def test_client_credentials_provider_obtains_a_token_without_an_authorize_step() -> None:
     """The client-credentials provider connects with no authorize step and a `client_credentials` grant.