@@ -416,6 +416,22 @@ async def _refresh_credential(
     if await refresher.is_refresh_needed(
         credential, self._auth_config.auth_scheme
     ):
+      raw_auth_credential = self._auth_config.raw_auth_credential
+      if (
+          raw_auth_credential is not None
+          and self._is_client_credentials_flow()
+          and credential.oauth2
+          and not credential.oauth2.refresh_token
+      ):
+        # A client credentials token has no refresh token; get a new one from
+        # the client credentials instead. Copy them so the exchange does not
+        # mutate the shared tool config.
+        new_credential, was_exchanged = await self._exchange_credential(
+            raw_auth_credential.model_copy(deep=True)
+        )
+        if was_exchanged:
+          return new_credential, True
+        return credential, False
       refreshed_credential = await refresher.refresh(
           credential, self._auth_config.auth_scheme
       )
 
@@ -103,6 +103,14 @@ async def refresh(
           "expires_at": auth_credential.oauth2.expires_at,
           "expires_in": auth_credential.oauth2.expires_in,
       }).is_expired():
+        if not auth_credential.oauth2.refresh_token:
+          # Client credentials tokens, and some providers' user tokens, come
+          # without a refresh token. Keep the stale token, as for a failed
+          # refresh below.
+          logger.warning(
+              "OAuth2 token expired and has no refresh token; cannot refresh"
+          )
+          return auth_credential
         client, token_endpoint = create_oauth2_session(
             auth_scheme, auth_credential
         )
 
@@ -308,3 +308,35 @@ async def test_refresh_unexpected_error_propagates(
     refresher = OAuth2CredentialRefresher()
     with pytest.raises(ValueError, match="unexpected bug"):
       await refresher.refresh(credential, scheme)
+
+  @patch("google.adk.auth.refresher.oauth2_credential_refresher.logger")
+  @pytest.mark.asyncio
+  async def test_refresh_without_refresh_token_returns_original_and_logs(
+      self, mock_logger
+  ):
+    """An expired token with no refresh token is returned as is, not a crash."""
+    scheme = OpenIdConnectWithConfig(
+        type_="openIdConnect",
+        openId_connect_url=(
+            "https://example.com/.well-known/openid_configuration"
+        ),
+        authorization_endpoint="https://example.com/auth",
+        token_endpoint="https://example.com/token",
+        scopes=["openid"],
+    )
+    credential = AuthCredential(
+        auth_type=AuthCredentialTypes.OPEN_ID_CONNECT,
+        oauth2=OAuth2Auth(
+            client_id="test_client_id",
+            client_secret="test_client_secret",
+            access_token="old_token",
+            expires_at=int(time.time()) - 3600,  # Expired
+        ),
+    )
+
+    refresher = OAuth2CredentialRefresher()
+    result = await refresher.refresh(credential, scheme)
+
+    assert result is credential
+    assert result.oauth2.access_token == "old_token"
+    mock_logger.warning.assert_called_once()
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import asyncio
+import time
 from unittest.mock import ANY
 from unittest.mock import AsyncMock
 from unittest.mock import Mock
@@ -1180,6 +1181,175 @@ async def test_refresh_credential_no_refresher(self):
     assert result == mock_credential
     assert was_refreshed is False
 
+  @pytest.mark.asyncio
+  @pytest.mark.parametrize(
+      "auth_scheme, auth_type",
+      [
+          (
+              OAuth2(
+                  flows=OAuthFlows(
+                      clientCredentials=OAuthFlowClientCredentials(
+                          tokenUrl="https://example.com/token",
+                          scopes={},
+                      )
+                  )
+              ),
+              AuthCredentialTypes.OAUTH2,
+          ),
+          (
+              OpenIdConnectWithConfig(
+                  authorization_endpoint="https://example.com/auth",
+                  token_endpoint="https://example.com/token",
+                  grant_types_supported=["client_credentials"],
+              ),
+              AuthCredentialTypes.OPEN_ID_CONNECT,
+          ),
+      ],
+      ids=["oauth2", "oidc"],
+  )
+  async def test_get_auth_credential_refetches_expired_client_credentials_token(
+      self, auth_scheme, auth_type
+  ):
+    """Client credentials tokens have no refresh token, so fetch a new one."""
+    raw_credential = AuthCredential(
+        auth_type=auth_type,
+        oauth2=OAuth2Auth(
+            client_id="mock_client_id",
+            client_secret="mock_client_secret",
+        ),
+    )
+    stored_credential = raw_credential.model_copy(deep=True)
+    stored_credential.oauth2.access_token = "expired_token"
+    stored_credential.oauth2.expires_at = int(time.time()) - 60
+    manager = CredentialManager(
+        AuthConfig(auth_scheme=auth_scheme, raw_auth_credential=raw_credential)
+    )
+    manager._load_existing_credential = AsyncMock(
+        return_value=stored_credential
+    )
+    manager._save_credential = AsyncMock()
+    context = Mock()
+
+    with patch.object(
+        oauth2_credential_util, "OAuth2Session", autospec=True
+    ) as mock_oauth2_session:
+      mock_client = mock_oauth2_session.return_value
+      mock_client.fetch_token.return_value = {
+          "access_token": "new_token",
+          "expires_at": int(time.time()) + 3600,
+      }
+      # There is no refresh token to send.
+      mock_client.refresh_token.side_effect = AssertionError(
+          "refresh_token() called without a refresh token"
+      )
+      result = await manager.get_auth_credential(context)
+
+    assert result.oauth2.access_token == "new_token"
+    mock_client.fetch_token.assert_called_once()
+    manager._save_credential.assert_awaited_once_with(context, result)
+    # The shared tool config is not mutated.
+    assert raw_credential.oauth2.access_token is None
+
+  @pytest.mark.asyncio
+  async def test_get_auth_credential_refreshes_token_that_has_refresh_token(
+      self,
+  ):
+    """A client credentials token that has a refresh token is refreshed."""
+    auth_scheme = OAuth2(
+        flows=OAuthFlows(
+            clientCredentials=OAuthFlowClientCredentials(
+                tokenUrl="https://example.com/token",
+                scopes={},
+            )
+        )
+    )
+    raw_credential = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(
+            client_id="mock_client_id",
+            client_secret="mock_client_secret",
+        ),
+    )
+    stored_credential = raw_credential.model_copy(deep=True)
+    stored_credential.oauth2.access_token = "expired_token"
+    stored_credential.oauth2.refresh_token = "refresh_token"
+    stored_credential.oauth2.expires_at = int(time.time()) - 60
+    manager = CredentialManager(
+        AuthConfig(auth_scheme=auth_scheme, raw_auth_credential=raw_credential)
+    )
+    manager._load_existing_credential = AsyncMock(
+        return_value=stored_credential
+    )
+    manager._save_credential = AsyncMock()
+    context = Mock()
+
+    with patch.object(
+        oauth2_credential_util, "OAuth2Session", autospec=True
+    ) as mock_oauth2_session:
+      mock_client = mock_oauth2_session.return_value
+      mock_client.refresh_token.return_value = {
+          "access_token": "refreshed_token",
+          "refresh_token": "refresh_token",
+          "expires_at": int(time.time()) + 3600,
+      }
+      result = await manager.get_auth_credential(context)
+
+    assert result.oauth2.access_token == "refreshed_token"
+    mock_client.refresh_token.assert_called_once()
+    assert (
+        mock_client.refresh_token.call_args.kwargs["refresh_token"]
+        == "refresh_token"
+    )
+    mock_client.fetch_token.assert_not_called()
+
+  @pytest.mark.asyncio
+  async def test_get_auth_credential_keeps_stored_token_when_refetch_fails(
+      self,
+  ):
+    """A failed client credentials refetch returns the stored token unsaved."""
+    auth_scheme = OAuth2(
+        flows=OAuthFlows(
+            clientCredentials=OAuthFlowClientCredentials(
+                tokenUrl="https://example.com/token",
+                scopes={},
+            )
+        )
+    )
+    raw_credential = AuthCredential(
+        auth_type=AuthCredentialTypes.OAUTH2,
+        oauth2=OAuth2Auth(
+            client_id="mock_client_id",
+            client_secret="mock_client_secret",
+        ),
+    )
+    stored_credential = raw_credential.model_copy(deep=True)
+    stored_credential.oauth2.access_token = "expired_token"
+    stored_credential.oauth2.expires_at = int(time.time()) - 60
+    manager = CredentialManager(
+        AuthConfig(auth_scheme=auth_scheme, raw_auth_credential=raw_credential)
+    )
+    manager._load_existing_credential = AsyncMock(
+        return_value=stored_credential
+    )
+    manager._save_credential = AsyncMock()
+    context = Mock()
+
+    with patch.object(
+        oauth2_credential_util, "OAuth2Session", autospec=True
+    ) as mock_oauth2_session:
+      mock_client = mock_oauth2_session.return_value
+      mock_client.fetch_token.side_effect = OAuth2Error(error="invalid_client")
+      mock_client.refresh_token.side_effect = AssertionError(
+          "refresh_token() called without a refresh token"
+      )
+      result = await manager.get_auth_credential(context)
+
+    assert result is stored_credential
+    assert result.oauth2.access_token == "expired_token"
+    mock_client.fetch_token.assert_called_once()
+    manager._save_credential.assert_not_awaited()
+    assert raw_credential.oauth2.access_token is None
+
   @pytest.mark.asyncio
   async def test_is_credential_ready_api_key(self):
     """Test _is_credential_ready with API key credential."""