refactor: Add diversion logic based on the auth provider resource name

PiperOrigin-RevId: 931487487

Author: google-genai-bot

src/google/adk/integrations/agent_identity/_agent_identity_credentials_provider.py

@@ -0,0 +1,49 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Credentials Provider using the Agent Identity service."""
+
+from __future__ import annotations
+
+from google.adk.agents.callback_context import CallbackContext
+from google.adk.auth.auth_credential import AuthCredential
+
+from .gcp_auth_provider_scheme import GcpAuthProviderScheme
+
+
+class _AgentIdentityCredentialsProvider:
+  """Auth provider implementation using Agent Identity credentials service."""
+
+  async def get_auth_credential(
+      self,
+      auth_scheme: GcpAuthProviderScheme,
+      context: CallbackContext | None = None,
+  ) -> AuthCredential:
+    """Retrieves credentials using the Agent Identity Credentials service.
+
+    Args:
+      auth_scheme: The GcpAuthProviderScheme.
+      context: Optional context for the callback.
+
+    Returns:
+      An AuthCredential instance.
+
+    Raises:
+      NotImplementedError: Auth provider using Agent Identity Credential service
+      is not yet supported.
+    """
+    raise NotImplementedError(
+        "Auth provider using Agent Identity Credential service is not yet"
+        " supported."
+    )

src/google/adk/integrations/agent_identity/gcp_auth_provider.py

@@ -16,21 +16,26 @@
 
 from __future__ import annotations
 
+
+import re
+
 from google.adk.agents.callback_context import CallbackContext
 from google.adk.auth.auth_credential import AuthCredential
 from google.adk.auth.auth_tool import AuthConfig
 from google.adk.auth.base_auth_provider import BaseAuthProvider
 from typing_extensions import override
 
+from ._agent_identity_credentials_provider import _AgentIdentityCredentialsProvider
 from ._iam_connector_credentials_provider import _IamConnectorCredentialsProvider
 from .gcp_auth_provider_scheme import GcpAuthProviderScheme
 
 
 class GcpAuthProvider(BaseAuthProvider):
-  """An auth provider that uses the Agent Identity Credentials service to generate access tokens."""
+  """An auth provider that uses Credentials service to generate access tokens."""
 
   def __init__(self):
     self._iam_connector_provider = _IamConnectorCredentialsProvider()
+    self._agent_identity_provider = _AgentIdentityCredentialsProvider()
 
   @property
   @override
@@ -43,7 +48,7 @@ async def get_auth_credential(
       auth_config: AuthConfig,
       context: CallbackContext | None = None,
   ) -> AuthCredential:
-    """Retrieves credentials using the Agent Identity Credentials service.
+    """Retrieves credentials using the Credentials service.
 
     Args:
       auth_config: The authentication configuration.
@@ -61,6 +66,13 @@ async def get_auth_credential(
           f"Expected GcpAuthProviderScheme, got {type(auth_scheme)}"
       )
 
-    return await self._iam_connector_provider.get_auth_credential(
+    if re.match(
+        r"^projects/[^/]+/locations/[^/]+/connectors/[^/]+$", auth_scheme.name
+    ):
+      return await self._iam_connector_provider.get_auth_credential(
+          auth_scheme=auth_scheme, context=context
+      )
+
+    return await self._agent_identity_provider.get_auth_credential(
         auth_scheme=auth_scheme, context=context
     )

tests/unittests/integrations/agent_identity/test_agent_identity_credentials_provider.py

@@ -0,0 +1,50 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest.mock import Mock
+
+from google.adk.agents.callback_context import CallbackContext
+from google.adk.integrations.agent_identity import GcpAuthProviderScheme
+from google.adk.integrations.agent_identity._agent_identity_credentials_provider import _AgentIdentityCredentialsProvider
+import pytest
+
+
+@pytest.fixture
+def auth_scheme():
+  scheme = GcpAuthProviderScheme(
+      name="projects/test-project/locations/global/connectors/test-connector",
+      scopes=["test-scope"],
+      continue_uri="https://example.com/continue",
+  )
+  return scheme
+
+
+@pytest.fixture
+def context():
+  context = Mock(spec=CallbackContext)
+  context.user_id = "user"
+  return context
+
+
+async def test_get_auth_credential_not_implemented(auth_scheme, context):
+  """Verify that get_auth_credential raises NotImplementedError initially."""
+  provider = _AgentIdentityCredentialsProvider()
+  with pytest.raises(
+      NotImplementedError,
+      match=(
+          "Auth provider using Agent Identity Credential service is not yet"
+          " supported."
+      ),
+  ):
+    await provider.get_auth_credential(auth_scheme, context)

tests/unittests/integrations/agent_identity/test_gcp_auth_provider.py

@@ -11,28 +11,26 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+"""Unit tests for the GcpAuthProvider class."""
 
 from unittest.mock import AsyncMock
 from unittest.mock import Mock
 from unittest.mock import patch
 
+
 from google.adk.agents.callback_context import CallbackContext
 from google.adk.auth.auth_credential import AuthCredential
 from google.adk.auth.auth_tool import AuthConfig
 from google.adk.integrations.agent_identity import GcpAuthProvider
 from google.adk.integrations.agent_identity import GcpAuthProviderScheme
-from google.adk.integrations.agent_identity._iam_connector_credentials_provider import _IamConnectorCredentialsProvider
 import pytest
 
 
 @pytest.fixture
 def auth_config():
-  scheme = GcpAuthProviderScheme(
-      name="projects/test-project/locations/global/connectors/test-connector",
-      scopes=["test-scope"],
-      continue_uri="https://example.com/continue",
-  )
-  return Mock(spec=AuthConfig, auth_scheme=scheme)
+  config = Mock(spec=AuthConfig)
+  config.auth_scheme = Mock(spec=GcpAuthProviderScheme)
+  return config
 
 
 @pytest.fixture
@@ -43,45 +41,72 @@ def context():
 
 
 @pytest.fixture
-def provider():
+def gcp_auth_provider():
   return GcpAuthProvider()
 
 
-def test_supported_auth_schemes(provider):
+def test_supported_auth_schemes(gcp_auth_provider):
   """Verify the provider supports the correct auth scheme."""
-  assert GcpAuthProviderScheme in provider.supported_auth_schemes
+  assert GcpAuthProviderScheme in gcp_auth_provider.supported_auth_schemes
+
+
+async def test_get_auth_credential_raises_error_for_invalid_auth_scheme(
+    context,
+):
+  """Test get_auth_credential raises ValueError for invalid auth scheme."""
+  provider = GcpAuthProvider()
+  invalid_auth_config = Mock(spec=AuthConfig)
+  invalid_auth_config.auth_scheme = Mock()  # Not GcpAuthProviderScheme
+
+  with pytest.raises(ValueError, match="Expected GcpAuthProviderScheme, got"):
+    await provider.get_auth_credential(invalid_auth_config, context)
 
 
 @patch(
     "google.adk.integrations.agent_identity.gcp_auth_provider._IamConnectorCredentialsProvider"
 )
-async def test_gcp_auth_provider_delegates_get_auth_credential(
-    mock_provider_class, auth_config, context
+async def test_get_auth_credential_routes_to_iam_connector_service_provider(
+    mock_iam_cls, auth_config, context
 ):
-  """Test that get_auth_credential delegates to the internal provider."""
+  """Test routing to IAM Connector Credentials service for legacy auth provider resource names."""
+  auth_config.auth_scheme.name = (
+      "projects/test-project/locations/test-location/connectors/test-connector"
+  )
   provider = GcpAuthProvider()
 
   mock_credential = Mock(spec=AuthCredential)
-  mock_provider_instance = mock_provider_class.return_value
-  mock_provider_instance.get_auth_credential = AsyncMock(
+  mock_iam_provider = mock_iam_cls.return_value
+  mock_iam_provider.get_auth_credential = AsyncMock(
       return_value=mock_credential
   )
 
   result = await provider.get_auth_credential(auth_config, context)
 
   assert result == mock_credential
-  mock_provider_instance.get_auth_credential.assert_awaited_once_with(
+  mock_iam_provider.get_auth_credential.assert_awaited_once_with(
       auth_scheme=auth_config.auth_scheme, context=context
   )
 
 
-async def test_get_auth_credential_raises_error_for_invalid_auth_scheme(
-    context,
+@patch(
+    "google.adk.integrations.agent_identity.gcp_auth_provider._AgentIdentityCredentialsProvider"
+)
+async def test_get_auth_credential_routes_to_agent_identity_service_provider(
+    mock_agent_cls, auth_config, context
 ):
-  """Test get_auth_credential raises ValueError for invalid auth scheme."""
+  """Test routing to Agent Identity Credentials service for new auth provider resource names."""
+  auth_config.auth_scheme.name = "projects/test-project/locations/test-location/authProviders/test-provider"
   provider = GcpAuthProvider()
-  invalid_auth_config = Mock(spec=AuthConfig)
-  invalid_auth_config.auth_scheme = Mock()  # Not GcpAuthProviderScheme
 
-  with pytest.raises(ValueError, match="Expected GcpAuthProviderScheme, got"):
-    await provider.get_auth_credential(invalid_auth_config, context)
+  mock_credential = Mock(spec=AuthCredential)
+  mock_agent_provider = mock_agent_cls.return_value
+  mock_agent_provider.get_auth_credential = AsyncMock(
+      return_value=mock_credential
+  )
+
+  result = await provider.get_auth_credential(auth_config, context)
+
+  assert result == mock_credential
+  mock_agent_provider.get_auth_credential.assert_awaited_once_with(
+      auth_scheme=auth_config.auth_scheme, context=context
+  )