fix: implement dynamic mtls endpoint resolution for parameter manager

Replace hardcoded regional URLs with a helper function that prioritizes
.mtls. endpoints when client certificates are present and enabled,
fulfilling mTLS/CAA requirements.

PiperOrigin-RevId: 937310286

Author: google-genai-bot

src/google/adk/integrations/parameter_manager/parameter_client.py

@@ -26,9 +26,17 @@
 from google.oauth2 import service_account
 
 from ... import version
+from ...utils import mtls_utils
 
 USER_AGENT = f"google-adk/{version.__version__}"
 
+_DEFAULT_REGIONAL_ENDPOINT_TEMPLATE = (
+    "parametermanager.{location}.rep.googleapis.com"
+)
+_DEFAULT_MTLS_REGIONAL_ENDPOINT_TEMPLATE = (
+    "parametermanager.{location}.rep.mtls.googleapis.com"
+)
+
 
 class ParameterManagerClient:
   """A client for interacting with Google Cloud Parameter Manager.
@@ -113,7 +121,11 @@ def __init__(
     client_options = None
     if location:
       client_options = {
-          "api_endpoint": f"parametermanager.{location}.rep.googleapis.com"
+          "api_endpoint": mtls_utils.get_api_endpoint(
+              location,
+              _DEFAULT_REGIONAL_ENDPOINT_TEMPLATE,
+              _DEFAULT_MTLS_REGIONAL_ENDPOINT_TEMPLATE,
+          )
       }
 
     self._client = parametermanager_v1.ParameterManagerClient(

src/google/adk/utils/mtls_utils.py

@@ -0,0 +1,68 @@
+# 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.
+
+"""Utilities for mTLS regional endpoint resolution."""
+
+from __future__ import annotations
+
+import enum
+import os
+
+from google.auth.transport import mtls
+
+
+class MtlsEndpoint(enum.Enum):
+  """Enum for the mTLS endpoint setting."""
+
+  AUTO = "auto"
+  ALWAYS = "always"
+  NEVER = "never"
+
+
+def use_client_cert_effective() -> bool:
+  """Returns whether client certificate should be used for mTLS."""
+  try:
+    return mtls.should_use_client_cert()
+  except (ImportError, AttributeError):
+    return (
+        os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false").lower()
+        == "true"
+    )
+
+
+def get_api_endpoint(
+    location: str, default_template: str, mtls_template: str
+) -> str:
+  """Returns API endpoint based on mTLS configuration and cert availability.
+
+  Args:
+      location: The region location.
+      default_template: Template for default regional endpoint (e.g.
+        "secretmanager.{location}.rep.googleapis.com").
+      mtls_template: Template for mTLS regional endpoint (e.g.
+        "secretmanager.{location}.rep.mtls.googleapis.com").
+  """
+  use_mtls_endpoint_str = os.getenv(
+      "GOOGLE_API_USE_MTLS_ENDPOINT", MtlsEndpoint.AUTO.value
+  ).lower()
+  try:
+    use_mtls_endpoint = MtlsEndpoint(use_mtls_endpoint_str)
+  except ValueError:
+    use_mtls_endpoint = MtlsEndpoint.AUTO
+
+  if (use_mtls_endpoint == MtlsEndpoint.ALWAYS) or (
+      use_mtls_endpoint == MtlsEndpoint.AUTO and use_client_cert_effective()
+  ):
+    return mtls_template.format(location=location)
+  return default_template.format(location=location)

tests/unittests/integrations/parameter_manager/test_parameter_client.py

@@ -122,8 +122,14 @@ def test_init_with_auth_token(self, mock_pm_client_class):
   @patch(
       "google.adk.integrations.parameter_manager.parameter_client.default_service_credential"
   )
+  @patch(
+      "google.adk.integrations.parameter_manager.parameter_client.mtls_utils.get_api_endpoint"
+  )
   def test_init_with_location(
-      self, mock_default_service_credential, mock_pm_client_class
+      self,
+      mock_get_api_endpoint,
+      mock_default_service_credential,
+      mock_pm_client_class,
   ):
     """Test initialization with a specific location."""
     # Setup
@@ -133,6 +139,7 @@ def test_init_with_location(
         "test-project",
     )
     location = "us-central1"
+    mock_get_api_endpoint.return_value = "resolved-endpoint"
 
     # Execute
     ParameterManagerClient(location=location)
@@ -142,9 +149,14 @@ def test_init_with_location(
     call_kwargs = mock_pm_client_class.call_args.kwargs
     assert call_kwargs["credentials"] == mock_credentials
     assert call_kwargs["client_options"] == {
-        "api_endpoint": f"parametermanager.{location}.rep.googleapis.com"
+        "api_endpoint": "resolved-endpoint"
     }
     assert call_kwargs["client_info"].user_agent == USER_AGENT
+    mock_get_api_endpoint.assert_called_once_with(
+        location,
+        "parametermanager.{location}.rep.googleapis.com",
+        "parametermanager.{location}.rep.mtls.googleapis.com",
+    )
 
   @patch(
       "google.adk.integrations.parameter_manager.parameter_client.default_service_credential"

tests/unittests/utils/test_mtls_utils.py

@@ -0,0 +1,120 @@
+# 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.
+
+"""Unit tests for mtls_utils."""
+
+import os
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+from google.adk.utils import mtls_utils
+import pytest
+
+_DEFAULT_TEMPLATE = "service.{location}.rep.googleapis.com"
+_MTLS_TEMPLATE = "service.{location}.rep.mtls.googleapis.com"
+_LOCATION = "us-central1"
+
+
+class TestMtlsUtils:
+  """Tests for mtls_utils functions."""
+
+  @patch("google.auth.transport.mtls.should_use_client_cert")
+  def test_use_client_cert_effective_with_mtls_cert_true(
+      self, mock_should_use_client_cert
+  ):
+    mock_should_use_client_cert.return_value = True
+    assert mtls_utils.use_client_cert_effective() is True
+    mock_should_use_client_cert.assert_called_once()
+
+  @patch("google.auth.transport.mtls.should_use_client_cert")
+  def test_use_client_cert_effective_with_mtls_cert_false(
+      self, mock_should_use_client_cert
+  ):
+    mock_should_use_client_cert.return_value = False
+    assert mtls_utils.use_client_cert_effective() is False
+    mock_should_use_client_cert.assert_called_once()
+
+  @patch("google.auth.transport.mtls.should_use_client_cert")
+  @patch.dict("os.environ", {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"})
+  def test_use_client_cert_effective_fallback_true(
+      self, mock_should_use_client_cert
+  ):
+    mock_should_use_client_cert.side_effect = AttributeError
+    assert mtls_utils.use_client_cert_effective() is True
+
+  @patch("google.auth.transport.mtls.should_use_client_cert")
+  @patch.dict("os.environ", {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"})
+  def test_use_client_cert_effective_fallback_false(
+      self, mock_should_use_client_cert
+  ):
+    mock_should_use_client_cert.side_effect = AttributeError
+    assert mtls_utils.use_client_cert_effective() is False
+
+  @patch("google.auth.transport.mtls.should_use_client_cert")
+  @patch.dict("os.environ", {}, clear=True)
+  def test_use_client_cert_effective_fallback_default_false(
+      self, mock_should_use_client_cert
+  ):
+    mock_should_use_client_cert.side_effect = AttributeError
+    assert mtls_utils.use_client_cert_effective() is False
+
+  @patch("google.adk.utils.mtls_utils.use_client_cert_effective")
+  @patch.dict("os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"})
+  def test_get_api_endpoint_always(self, mock_use_client_cert):
+    endpoint = mtls_utils.get_api_endpoint(
+        _LOCATION, _DEFAULT_TEMPLATE, _MTLS_TEMPLATE
+    )
+    assert endpoint == _MTLS_TEMPLATE.format(location=_LOCATION)
+    mock_use_client_cert.assert_not_called()
+
+  @patch("google.adk.utils.mtls_utils.use_client_cert_effective")
+  @patch.dict("os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": "never"})
+  def test_get_api_endpoint_never(self, mock_use_client_cert):
+    endpoint = mtls_utils.get_api_endpoint(
+        _LOCATION, _DEFAULT_TEMPLATE, _MTLS_TEMPLATE
+    )
+    assert endpoint == _DEFAULT_TEMPLATE.format(location=_LOCATION)
+    mock_use_client_cert.assert_not_called()
+
+  @patch("google.adk.utils.mtls_utils.use_client_cert_effective")
+  @patch.dict("os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"})
+  def test_get_api_endpoint_auto_with_cert(self, mock_use_client_cert):
+    mock_use_client_cert.return_value = True
+    endpoint = mtls_utils.get_api_endpoint(
+        _LOCATION, _DEFAULT_TEMPLATE, _MTLS_TEMPLATE
+    )
+    assert endpoint == _MTLS_TEMPLATE.format(location=_LOCATION)
+    mock_use_client_cert.assert_called_once()
+
+  @patch("google.adk.utils.mtls_utils.use_client_cert_effective")
+  @patch.dict("os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"})
+  def test_get_api_endpoint_auto_without_cert(self, mock_use_client_cert):
+    mock_use_client_cert.return_value = False
+    endpoint = mtls_utils.get_api_endpoint(
+        _LOCATION, _DEFAULT_TEMPLATE, _MTLS_TEMPLATE
+    )
+    assert endpoint == _DEFAULT_TEMPLATE.format(location=_LOCATION)
+    mock_use_client_cert.assert_called_once()
+
+  @patch("google.adk.utils.mtls_utils.use_client_cert_effective")
+  @patch.dict("os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": "invalid_value"})
+  def test_get_api_endpoint_invalid_fallback_to_auto(
+      self, mock_use_client_cert
+  ):
+    mock_use_client_cert.return_value = True
+    endpoint = mtls_utils.get_api_endpoint(
+        _LOCATION, _DEFAULT_TEMPLATE, _MTLS_TEMPLATE
+    )
+    assert endpoint == _MTLS_TEMPLATE.format(location=_LOCATION)
+    mock_use_client_cert.assert_called_once()