feat: Update agent_registry to handle mTLS endpoints internally

Transforms connection URIs conditionally based on mtls.should_use_mtls_endpoint() and client_cert checks, addressing the mTLS implementation guidelines safely.

PiperOrigin-RevId: 954825139

Author: google-genai-bot

src/google/adk/integrations/agent_registry/agent_registry.py

@@ -39,6 +39,7 @@
 from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
 from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
 from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+from google.adk.utils import _mtls_utils
 import google.auth
 from google.auth.transport import mtls
 from google.auth.transport import requests as requests_auth
@@ -222,7 +223,12 @@ def __init__(
           else None
       )
       self._session.configure_mtls_channel(client_cert_source)
-    self._base_url = _get_agent_registry_base_url(client_cert_source)
+    self._use_mtls = _should_use_mtls_endpoint(client_cert_source)
+    self._base_url = (
+        AGENT_REGISTRY_MTLS_BASE_URL
+        if self._use_mtls
+        else AGENT_REGISTRY_BASE_URL
+    )
 
   def _get_auth_headers(self) -> Dict[str, str]:
     """Refreshes credentials and returns authorization headers."""
@@ -325,6 +331,8 @@ def _get_connection_uri(
         if protocol_binding and mapped_binding != protocol_binding:
           continue
         if url := i.get("url"):
+          if self._use_mtls:
+            url = _mtls_utils.effective_googleapis_endpoint(url)
           return url, protocol_version, mapped_binding
 
     return None, None, None
@@ -637,17 +645,19 @@ def _use_client_cert_effective() -> bool:
     return use_client_cert_str == "true"
 
 
-def _get_agent_registry_base_url(client_cert_source: Any | None = None) -> str:
-  """Returns the base URL based on mTLS configuration and cert availability."""
+def _should_use_mtls_endpoint(client_cert_source: Any | None = None) -> bool:
+  """Returns whether the mTLS endpoint should be used."""
+  try:
+    return bool(mtls.should_use_mtls_endpoint())
+  except (ImportError, AttributeError):
+    pass
   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 is _MtlsEndpoint.ALWAYS) or (
+  return (use_mtls_endpoint is _MtlsEndpoint.ALWAYS) or (
       use_mtls_endpoint is _MtlsEndpoint.AUTO and client_cert_source is not None
-  ):
-    return AGENT_REGISTRY_MTLS_BASE_URL
-  return AGENT_REGISTRY_BASE_URL
+  )

tests/unittests/integrations/agent_registry/test_agent_registry.py

@@ -888,23 +888,23 @@ def test_use_client_cert_effective(
         assert _use_client_cert_effective() == expected
 
   @pytest.mark.parametrize(
-      "use_mtls_env, client_cert_source, expected_domain",
+      "use_mtls_env, client_cert_source, expected",
       [
           # Auto mode (default)
-          (None, None, "agentregistry.googleapis.com"),
-          (None, lambda: True, "agentregistry.mtls.googleapis.com"),
+          (None, None, False),
+          (None, lambda: True, True),
           # Always mode
-          ("always", None, "agentregistry.mtls.googleapis.com"),
-          ("always", lambda: True, "agentregistry.mtls.googleapis.com"),
+          ("always", None, True),
+          ("always", lambda: True, True),
           # Never mode
-          ("never", None, "agentregistry.googleapis.com"),
-          ("never", lambda: True, "agentregistry.googleapis.com"),
+          ("never", None, False),
+          ("never", lambda: True, False),
       ],
   )
-  def test_get_agent_registry_base_url(
-      self, use_mtls_env, client_cert_source, expected_domain, registry
+  def test_should_use_mtls_endpoint(
+      self, use_mtls_env, client_cert_source, expected, registry
   ):
-    from google.adk.integrations.agent_registry.agent_registry import _get_agent_registry_base_url
+    from google.adk.integrations.agent_registry.agent_registry import _should_use_mtls_endpoint
 
     env_patch = {}
     if use_mtls_env is not None:
@@ -914,7 +914,7 @@ def test_get_agent_registry_base_url(
       env_patch = {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"}
 
     with patch.dict(os.environ, env_patch):
-      assert expected_domain in _get_agent_registry_base_url(client_cert_source)
+      assert expected == _should_use_mtls_endpoint(client_cert_source)
 
   def test_make_request_error_handling(self, registry):
     mock_session = registry._session