fix(mcp): reject config-supplied stdio MCP servers by default

src/google/adk/tools/mcp_tool/mcp_toolset.py

@@ -61,6 +61,22 @@
 
 logger = logging.getLogger("google_adk." + __name__)
 
+# Whether an agent config may declare a stdio MCP server. Stdio connection
+# params carry a `command` and `args` that are launched as a local process, and
+# agent configs are untrusted input (CVE-2026-4810), so they are rejected in
+# `from_config()` unless the embedding application explicitly opts in.
+_ALLOW_CONFIG_STDIO_SERVERS = False
+
+
+def _set_allow_config_stdio_servers(value: bool) -> None:
+  """Sets whether `McpToolset.from_config()` may build stdio MCP servers.
+
+  Args:
+    value: True to allow agent configs to launch local MCP servers.
+  """
+  global _ALLOW_CONFIG_STDIO_SERVERS
+  _ALLOW_CONFIG_STDIO_SERVERS = value
+
 
 T = TypeVar("T")
 
@@ -515,6 +531,17 @@ def from_config(
     """Creates an McpToolset from a configuration object."""
     mcp_toolset_config = McpToolsetConfig.model_validate(config.model_dump())
 
+    if (
+        mcp_toolset_config.stdio_server_params
+        or mcp_toolset_config.stdio_connection_params
+    ) and not _ALLOW_CONFIG_STDIO_SERVERS:
+      raise ValueError(
+          "Stdio MCP servers are not allowed in agent configs: they launch a"
+          " local process from a config-supplied 'command'. Build the"
+          " McpToolset in code, or call _set_allow_config_stdio_servers(True)"
+          " if this application only loads trusted configs."
+      )
+
     if mcp_toolset_config.stdio_server_params:
       connection_params = mcp_toolset_config.stdio_server_params
     elif mcp_toolset_config.stdio_connection_params:

tests/unittests/tools/mcp_tool/test_mcp_toolset.py

@@ -41,6 +41,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_tool import MCPTool
+from google.adk.tools.mcp_tool.mcp_toolset import _set_allow_config_stdio_servers
 from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
 from google.adk.tools.tool_configs import ToolArgsConfig
 from mcp import StdioServerParameters
@@ -254,11 +255,51 @@ def test_from_config_with_credential_key(self):
         auth_scheme=auth_scheme,
         credential_key="my_custom_key",
     )
-    toolset = McpToolset.from_config(config, "")
+    _set_allow_config_stdio_servers(True)
+    try:
+      toolset = McpToolset.from_config(config, "")
+    finally:
+      _set_allow_config_stdio_servers(False)
 
     assert isinstance(toolset._auth_scheme, OAuth2)
     assert toolset._auth_config.credential_key == "my_custom_key"
 
+  def test_from_config_rejects_stdio_server_params(self):
+    """Config-supplied stdio servers are rejected by default."""
+    config = ToolArgsConfig(stdio_server_params=self.mock_stdio_params)
+    with pytest.raises(ValueError, match="not allowed in agent configs"):
+      McpToolset.from_config(config, "")
+
+  def test_from_config_rejects_stdio_connection_params(self):
+    """The stdio_connection_params spelling is rejected the same way."""
+    config = ToolArgsConfig(
+        stdio_connection_params=StdioConnectionParams(
+            server_params=self.mock_stdio_params
+        )
+    )
+    with pytest.raises(ValueError, match="not allowed in agent configs"):
+      McpToolset.from_config(config, "")
+
+  def test_from_config_allows_stdio_when_opted_in(self):
+    """The opt-in restores the previous behavior."""
+    config = ToolArgsConfig(stdio_server_params=self.mock_stdio_params)
+    _set_allow_config_stdio_servers(True)
+    try:
+      toolset = McpToolset.from_config(config, "")
+    finally:
+      _set_allow_config_stdio_servers(False)
+    assert isinstance(toolset, McpToolset)
+
+  def test_from_config_allows_remote_connection_params(self):
+    """Remote MCP servers are unaffected: no local process is launched."""
+    config = ToolArgsConfig(
+        sse_connection_params=SseConnectionParams(
+            url="https://example.com/sse"
+        )
+    )
+    toolset = McpToolset.from_config(config, "")
+    assert isinstance(toolset, McpToolset)
+
   def test_init_missing_connection_params(self):
     """Test initialization with missing connection params raises error."""
     with pytest.raises(ValueError, match="Missing connection params"):