fix: route MCP calls to the .mtls.googleapis.com endpoint for Agent Identity

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._mtls_utils import effective_googleapis_endpoint
 import google.auth
 from google.auth.transport import mtls
 from google.auth.transport import requests as requests_auth
@@ -222,6 +223,9 @@ def __init__(
           else None
       )
       self._session.configure_mtls_channel(client_cert_source)
+    # Retained so endpoint selection for MCP/A2A targets can mirror the base-URL
+    # policy: `auto` uses the mTLS endpoint only when a cert is actually present.
+    self._client_cert_source = client_cert_source
     self._base_url = _get_agent_registry_base_url(client_cert_source)
 
   def _get_auth_headers(self) -> Dict[str, str]:
@@ -422,6 +426,17 @@ def get_mcp_toolset(
           f"MCP Server endpoint URI not found for: {mcp_server_name}"
       )
 
+    # Route Google-hosted MCP servers to the mutual-TLS endpoint following the
+    # same policy as the registry base URL. A channel-bound (Agent Identity)
+    # access token is only accepted on *.mtls.googleapis.com; presenting it to
+    # the standard endpoint returns 401 even though mTLS was negotiated. Gating
+    # on the resolved client cert (not just the cert-use policy) avoids pointing
+    # a plain client at the mTLS host when `auto` is set but no cert is present.
+    # effective_googleapis_endpoint is itself a no-op for non-Google hosts and
+    # already-mTLS hosts.
+    if _should_use_mtls_endpoint(self._client_cert_source):
+      endpoint_uri = effective_googleapis_endpoint(endpoint_uri)
+
     if mcp_server_id and not auth_scheme:
       try:
         bindings_data = self._make_request("bindings")
@@ -637,17 +652,29 @@ 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) -> bool:
+  """Whether Google endpoints should resolve to their .mtls.googleapis.com host.
+
+  Follows the GOOGLE_API_USE_MTLS_ENDPOINT policy (AIP-4114): ``always`` selects
+  the mTLS endpoint unconditionally, ``auto`` only when a client certificate is
+  actually available (``client_cert_source is not None``), and ``never`` never.
+  Using cert *availability* rather than cert-use *policy* avoids pointing a
+  plain client (mTLS fell back) at an mTLS-only host.
+  """
   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
-  ):
+  )
+
+
+def _get_agent_registry_base_url(client_cert_source: Any | None = None) -> str:
+  """Returns the base URL based on mTLS configuration and cert availability."""
+  if _should_use_mtls_endpoint(client_cert_source):
     return AGENT_REGISTRY_MTLS_BASE_URL
   return AGENT_REGISTRY_BASE_URL

tests/unittests/integrations/agent_registry/test_agent_registry.py

@@ -196,6 +196,67 @@ async def test_get_mcp_toolset_handles_missing_destination_id(
       # The custom_metadata shouldn't have been added
       assert tool.custom_metadata is None
 
+  _GOOGLE_MCP_URL = "https://us-central1-aiplatform.googleapis.com/mcp"
+  _GOOGLE_MCP_MTLS_URL = (
+      "https://us-central1-aiplatform.mtls.googleapis.com/mcp"
+  )
+
+  def _stub_mcp_server(self, registry, url):
+    """Configures the registry to return an MCP server whose interface is `url`."""
+    mock_response = MagicMock()
+    mock_response.json.return_value = {
+        "displayName": "TestPrefix",
+        "interfaces": [{"url": url, "protocolBinding": "JSONRPC"}],
+    }
+    registry._session.get.return_value = mock_response
+    registry._credentials.token = "token"
+    registry._credentials.refresh = MagicMock()
+
+  def test_get_mcp_toolset_auto_with_cert_uses_mtls_endpoint(self, registry):
+    # auto + client certificate available -> mTLS endpoint.
+    self._stub_mcp_server(registry, self._GOOGLE_MCP_URL)
+    registry._client_cert_source = object()
+    with patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"}):
+      toolset = registry.get_mcp_toolset("test-mcp-server")
+    assert toolset.connection_params.url == self._GOOGLE_MCP_MTLS_URL
+
+  def test_get_mcp_toolset_auto_without_cert_keeps_plain_endpoint(
+      self, registry
+  ):
+    # auto + no certificate available -> plain endpoint (avoids pointing a
+    # plain client, after mTLS fallback, at the mTLS host).
+    self._stub_mcp_server(registry, self._GOOGLE_MCP_URL)
+    registry._client_cert_source = None
+    with patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"}):
+      toolset = registry.get_mcp_toolset("test-mcp-server")
+    assert toolset.connection_params.url == self._GOOGLE_MCP_URL
+
+  def test_get_mcp_toolset_always_without_cert_uses_mtls_endpoint(
+      self, registry
+  ):
+    # always -> mTLS endpoint unconditionally, even without a cert.
+    self._stub_mcp_server(registry, self._GOOGLE_MCP_URL)
+    registry._client_cert_source = None
+    with patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}):
+      toolset = registry.get_mcp_toolset("test-mcp-server")
+    assert toolset.connection_params.url == self._GOOGLE_MCP_MTLS_URL
+
+  def test_get_mcp_toolset_never_keeps_plain_endpoint_with_cert(self, registry):
+    # never -> plain endpoint even when a cert is available.
+    self._stub_mcp_server(registry, self._GOOGLE_MCP_URL)
+    registry._client_cert_source = object()
+    with patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}):
+      toolset = registry.get_mcp_toolset("test-mcp-server")
+    assert toolset.connection_params.url == self._GOOGLE_MCP_URL
+
+  def test_get_mcp_toolset_keeps_non_google_endpoint(self, registry):
+    # Non-Google hosts are never rewritten, even under always + cert.
+    self._stub_mcp_server(registry, "https://mcp.example.com/mcp")
+    registry._client_cert_source = object()
+    with patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}):
+      toolset = registry.get_mcp_toolset("test-mcp-server")
+    assert toolset.connection_params.url == "https://mcp.example.com/mcp"
+
   def test_init_raises_value_error_if_params_missing(self):
     with pytest.raises(
         ValueError, match="project_id and location must be provided"