feat: refresh credentials if token is missing in the common code and samples

PiperOrigin-RevId: 894473156

Author: google-genai-bot

contributing/samples/bigquery/agent.py

@@ -21,6 +21,7 @@
 from google.adk.tools.bigquery.config import BigQueryToolConfig
 from google.adk.tools.bigquery.config import WriteMode
 import google.auth
+import google.auth.transport.requests
 
 # Define the desired credential type.
 # By default use Application Default Credentials (ADC) from the local
@@ -57,6 +58,8 @@
   # service account key file
   # https://cloud.google.com/iam/docs/service-account-creds#user-managed-keys
   creds, _ = google.auth.load_credentials_from_file("service_account_key.json")
+  if not creds.valid:
+    creds.refresh(google.auth.transport.requests.Request())
   credentials_config = BigQueryCredentialsConfig(credentials=creds)
 elif CREDENTIALS_TYPE == AuthCredentialTypes.HTTP:
   # Initialize the tools to use the externally provided access token. One such
@@ -73,6 +76,10 @@
   # Initialize the tools to use the application default credentials.
   # https://cloud.google.com/docs/authentication/provide-credentials-adc
   application_default_credentials, _ = google.auth.default()
+  if not application_default_credentials.valid:
+    application_default_credentials.refresh(
+        google.auth.transport.requests.Request()
+    )
   credentials_config = BigQueryCredentialsConfig(
       credentials=application_default_credentials
   )

contributing/samples/data_agent/agent.py

@@ -51,6 +51,10 @@
   # Initialize the tools to use the application default credentials.
   # https://cloud.google.com/docs/authentication/provide-credentials-adc
   application_default_credentials, _ = google.auth.default()
+  if not application_default_credentials.valid:
+    application_default_credentials.refresh(
+        google.auth.transport.requests.Request()
+    )
   credentials_config = DataAgentCredentialsConfig(
       credentials=application_default_credentials
   )

src/google/adk/tools/_google_credentials.py

@@ -191,6 +191,13 @@ async def get_valid_credentials(
     # If non-oauth credentials are provided then use them as is. This helps
     # in flows such as service account keys
     if creds and not isinstance(creds, google.oauth2.credentials.Credentials):
+      if not creds.valid:
+        try:
+          creds.refresh(Request())
+        except Exception:  # pylint: disable=broad-except
+          # If refresh fails, we still return the creds as they might work
+          # for some libraries that handle refresh internally.
+          pass
       return creds
 
     # Check if we have valid credentials

tests/unittests/tools/test_base_google_credentials_manager.py

@@ -105,26 +105,56 @@ async def test_get_valid_credentials_with_valid_existing_creds(
       ],
   )
   @pytest.mark.asyncio
+  @patch.object(_google_credentials, "Request", autospec=True)
   async def test_get_valid_credentials_with_existing_non_oauth_creds(
-      self, manager, mock_tool_context, valid
+      self, mock_request_class, manager, mock_tool_context, valid
   ):
-    """Test that existing non-oauth credentials are returned immediately.
+    """Test that existing non-oauth credentials handle refresh logic correctly.
 
-    When credentials are of non-oauth type, no refresh or OAuth flow
-    is triggered irrespective of whether or not it is valid.
+    When credentials are of non-oauth type, a refresh is triggered if they
+    are invalid. No OAuth flow is ever triggered.
     """
-    # Create mock credentials that are already valid
+    # Create mock credentials with the specified validity
     mock_creds = create_autospec(AuthCredentials, instance=True)
     mock_creds.valid = valid
     manager.credentials_config.credentials = mock_creds
 
     result = await manager.get_valid_credentials(mock_tool_context)
 
     assert result == mock_creds
+    # Verify refresh behavior
+    if valid:
+      mock_creds.refresh.assert_not_called()
+    else:
+      mock_creds.refresh.assert_called_once_with(
+          mock_request_class.return_value
+      )
+
     # Verify no OAuth flow was triggered
     mock_tool_context.get_auth_response.assert_not_called()
     mock_tool_context.request_credential.assert_not_called()
 
+  @pytest.mark.asyncio
+  @patch.object(_google_credentials, "Request", autospec=True)
+  async def test_get_valid_credentials_with_non_oauth_refresh_failure(
+      self, mock_request_class, manager, mock_tool_context
+  ):
+    """Test that non-oauth refresh failures are caught gracefully.
+
+    Even if refresh fails, we should still return the credentials as they
+    might work for some downstream libraries.
+    """
+    mock_creds = create_autospec(AuthCredentials, instance=True)
+    mock_creds.valid = False
+    mock_creds.refresh.side_effect = Exception("Refresh failed")
+    manager.credentials_config.credentials = mock_creds
+
+    result = await manager.get_valid_credentials(mock_tool_context)
+
+    # Credentials should still be returned
+    assert result == mock_creds
+    mock_creds.refresh.assert_called_once_with(mock_request_class.return_value)
+
   @pytest.mark.asyncio
   async def test_get_credentials_from_cache_when_none_in_manager(
       self, manager, mock_tool_context