fix(oauth): preserve existing refresh_token when refresh response omits it (#2270)

Author: oedokumaci

src/mcp/client/auth/oauth2.py

@@ -447,7 +447,7 @@ async def _refresh_token(self) -> httpx.Request:
 
         return httpx.Request("POST", token_url, data=refresh_data, headers=headers)
 
-    async def _handle_refresh_response(self, response: httpx.Response) -> bool:  # pragma: no cover
+    async def _handle_refresh_response(self, response: httpx.Response) -> bool:
         """Handle token refresh response. Returns True if successful."""
         if response.status_code != 200:
             logger.warning(f"Token refresh failed: {response.status_code}")
@@ -458,6 +458,18 @@ async def _handle_refresh_response(self, response: httpx.Response) -> bool:  # p
             content = await response.aread()
             token_response = OAuthToken.model_validate_json(content)
 
+            # Per RFC 6749 Section 6, the authorization server MAY issue a new
+            # refresh token. If the response omits one, preserve the existing
+            # refresh token so subsequent refresh attempts remain possible.
+            if (
+                not token_response.refresh_token
+                and self.context.current_tokens
+                and self.context.current_tokens.refresh_token
+            ):
+                token_response = token_response.model_copy(
+                    update={"refresh_token": self.context.current_tokens.refresh_token}
+                )
+
             self.context.current_tokens = token_response
             self.context.update_token_expiry(token_response)
             await self.context.storage.set_tokens(token_response)

tests/client/test_auth.py

@@ -711,6 +711,109 @@ async def test_basic_auth_refresh_token(self, oauth_provider: OAuthClientProvide
         content = request.content.decode()
         assert "client_secret=" not in content
 
+    @pytest.mark.anyio
+    async def test_handle_refresh_response_preserves_existing_refresh_token(
+        self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
+    ):
+        """Test that the existing refresh_token is preserved when the server omits it.
+
+        Per RFC 6749 Section 6, the authorization server MAY issue a new refresh
+        token in the refresh response. If it doesn't, the client should continue
+        using the existing one.
+        """
+        oauth_provider.context.current_tokens = valid_tokens
+
+        # Server response without refresh_token
+        refresh_response = httpx.Response(
+            200,
+            content=b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600}',
+            request=httpx.Request("POST", "https://auth.example.com/token"),
+        )
+
+        result = await oauth_provider._handle_refresh_response(refresh_response)
+
+        assert result is True
+        assert oauth_provider.context.current_tokens is not None
+        assert oauth_provider.context.current_tokens.access_token == "new_access_token"
+        # Old refresh_token should be preserved
+        assert oauth_provider.context.current_tokens.refresh_token == "test_refresh_token"
+
+    @pytest.mark.anyio
+    async def test_handle_refresh_response_uses_new_refresh_token(
+        self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
+    ):
+        """Test that a new refresh_token from the server replaces the old one."""
+        oauth_provider.context.current_tokens = valid_tokens
+
+        # Server response with a new refresh_token (token rotation)
+        refresh_response = httpx.Response(
+            200,
+            content=(
+                b'{"access_token": "new_access_token", "token_type": "Bearer",'
+                b' "expires_in": 3600, "refresh_token": "rotated_refresh_token"}'
+            ),
+            request=httpx.Request("POST", "https://auth.example.com/token"),
+        )
+
+        result = await oauth_provider._handle_refresh_response(refresh_response)
+
+        assert result is True
+        assert oauth_provider.context.current_tokens is not None
+        assert oauth_provider.context.current_tokens.access_token == "new_access_token"
+        assert oauth_provider.context.current_tokens.refresh_token == "rotated_refresh_token"
+
+    @pytest.mark.anyio
+    async def test_handle_refresh_response_no_prior_tokens(self, oauth_provider: OAuthClientProvider):
+        """Test refresh response when there are no prior tokens stored."""
+        oauth_provider.context.current_tokens = None
+
+        refresh_response = httpx.Response(
+            200,
+            content=b'{"access_token": "new_access_token", "token_type": "Bearer", "expires_in": 3600}',
+            request=httpx.Request("POST", "https://auth.example.com/token"),
+        )
+
+        result = await oauth_provider._handle_refresh_response(refresh_response)
+
+        assert result is True
+        assert oauth_provider.context.current_tokens is not None
+        assert oauth_provider.context.current_tokens.access_token == "new_access_token"
+        assert oauth_provider.context.current_tokens.refresh_token is None
+
+    @pytest.mark.anyio
+    async def test_handle_refresh_response_failure(self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken):
+        """Test that a non-200 refresh response clears tokens."""
+        oauth_provider.context.current_tokens = valid_tokens
+
+        refresh_response = httpx.Response(
+            401,
+            content=b"Unauthorized",
+            request=httpx.Request("POST", "https://auth.example.com/token"),
+        )
+
+        result = await oauth_provider._handle_refresh_response(refresh_response)
+
+        assert result is False
+        assert oauth_provider.context.current_tokens is None
+
+    @pytest.mark.anyio
+    async def test_handle_refresh_response_invalid_json(
+        self, oauth_provider: OAuthClientProvider, valid_tokens: OAuthToken
+    ):
+        """Test that an invalid response body clears tokens."""
+        oauth_provider.context.current_tokens = valid_tokens
+
+        refresh_response = httpx.Response(
+            200,
+            content=b"not valid json",
+            request=httpx.Request("POST", "https://auth.example.com/token"),
+        )
+
+        result = await oauth_provider._handle_refresh_response(refresh_response)
+
+        assert result is False
+        assert oauth_provider.context.current_tokens is None
+
     @pytest.mark.anyio
     async def test_none_auth_method(self, oauth_provider: OAuthClientProvider):
         """Test 'none' authentication method (public client)."""