feat: Add require_confirmation param for MCP tool/toolset

This allows users to require human approval for using MCP tools.

PiperOrigin-RevId: 819800747

contributing/samples/mcp_sse_agent/agent.py

@@ -53,6 +53,7 @@ Allowed directory: {_allowed_path}
                 'get_file_info',
                 'list_allowed_directories',
             ],
+            require_confirmation=True,
         )
     ],
 )

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

@@ -15,8 +15,12 @@
 from __future__ import annotations
 
 import base64
+import inspect
 import logging
+from typing import Any
+from typing import Callable
 from typing import Optional
+from typing import Union
 import warnings
 
 from fastapi.openapi.models import APIKeyIn
@@ -70,6 +74,7 @@ class McpTool(BaseAuthenticatedTool):
       mcp_session_manager: MCPSessionManager,
       auth_scheme: Optional[AuthScheme] = None,
       auth_credential: Optional[AuthCredential] = None,
+      require_confirmation: Union[bool, Callable[..., bool]] = False,
   ):
     """Initializes an MCPTool.
 
@@ -81,6 +86,10 @@ class McpTool(BaseAuthenticatedTool):
         mcp_session_manager: The MCP session manager to use for communication.
         auth_scheme: The authentication scheme to use.
         auth_credential: The authentication credential to use.
+        require_confirmation: Whether this tool requires confirmation. A boolean
+          or a callable that takes the function's arguments and returns a
+          boolean. If the callable returns True, the tool will require
+          confirmation from the user.
 
     Raises:
         ValueError: If mcp_tool or mcp_session_manager is None.
@@ -96,6 +105,7 @@ class McpTool(BaseAuthenticatedTool):
     )
     self._mcp_tool = mcp_tool
     self._mcp_session_manager = mcp_session_manager
+    self._require_confirmation = require_confirmation
 
   @override
   def _get_declaration(self) -> FunctionDeclaration:
@@ -116,6 +126,57 @@ class McpTool(BaseAuthenticatedTool):
     """Returns the raw MCP tool."""
     return self._mcp_tool
 
+  async def _invoke_callable(
+      self, target: Callable[..., Any], args_to_call: dict[str, Any]
+  ) -> Any:
+    """Invokes a callable, handling both sync and async cases."""
+
+    # Functions are callable objects, but not all callable objects are functions
+    # checking coroutine function is not enough. We also need to check whether
+    # Callable's __call__ function is a coroutine funciton
+    is_async = inspect.iscoroutinefunction(target) or (
+        hasattr(target, "__call__")
+        and inspect.iscoroutinefunction(target.__call__)
+    )
+    if is_async:
+      return await target(**args_to_call)
+    else:
+      return target(**args_to_call)
+
+  @override
+  async def run_async(
+      self, *, args: dict[str, Any], tool_context: ToolContext
+  ) -> Any:
+    if isinstance(self._require_confirmation, Callable):
+      require_confirmation = await self._invoke_callable(
+          self._require_confirmation, args
+      )
+    else:
+      require_confirmation = bool(self._require_confirmation)
+
+    if require_confirmation:
+      if not tool_context.tool_confirmation:
+        args_to_show = args.copy()
+        if "tool_context" in args_to_show:
+          args_to_show.pop("tool_context")
+
+        tool_context.request_confirmation(
+            hint=(
+                f"Please approve or reject the tool call {self.name}() by"
+                " responding with a FunctionResponse with an expected"
+                " ToolConfirmation payload."
+            ),
+        )
+        return {
+            "error": (
+                "This tool call requires confirmation, please approve or"
+                " reject."
+            )
+        }
+      elif not tool_context.tool_confirmation.confirmed:
+        return {"error": "This tool call is rejected."}
+    return await super().run_async(args=args, tool_context=tool_context)
+
   @retry_on_closed_resource
   @override
   async def _run_async_impl(

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

@@ -16,6 +16,8 @@ from __future__ import annotations
 
 import logging
 import sys
+from typing import Callable
+from typing import Dict
 from typing import List
 from typing import Optional
 from typing import TextIO
@@ -104,6 +106,7 @@ class McpToolset(BaseToolset):
       errlog: TextIO = sys.stderr,
       auth_scheme: Optional[AuthScheme] = None,
       auth_credential: Optional[AuthCredential] = None,
+      require_confirmation: Union[bool, Callable[..., bool]] = False,
   ):
     """Initializes the MCPToolset.
 
@@ -124,6 +127,9 @@ class McpToolset(BaseToolset):
       errlog: TextIO stream for error logging.
       auth_scheme: The auth scheme of the tool for tool calling
       auth_credential: The auth credential of the tool for tool calling
+      require_confirmation: Whether tools in this toolset require
+        confirmation. Can be a single boolean or a callable to apply to all
+        tools.
     """
     super().__init__(tool_filter=tool_filter, tool_name_prefix=tool_name_prefix)
 
@@ -140,6 +146,7 @@ class McpToolset(BaseToolset):
     )
     self._auth_scheme = auth_scheme
     self._auth_credential = auth_credential
+    self._require_confirmation = require_confirmation
 
   @retry_on_closed_resource
   async def get_tools(
@@ -169,6 +176,7 @@ class McpToolset(BaseToolset):
           mcp_session_manager=self._mcp_session_manager,
           auth_scheme=self._auth_scheme,
           auth_credential=self._auth_credential,
+          require_confirmation=self._require_confirmation,
       )
 
       if self._is_tool_selected(mcp_tool, readonly_context):

tests/unittests/tools/mcp_tool/test_mcp_tool.py

@@ -549,6 +549,89 @@ class TestMCPTool:
           in logged_message
       )
 
+  @pytest.mark.asyncio
+  async def test_run_async_require_confirmation_true_no_confirmation(self):
+    """Test require_confirmation=True with no confirmation in context."""
+    tool = MCPTool(
+        mcp_tool=self.mock_mcp_tool,
+        mcp_session_manager=self.mock_session_manager,
+        require_confirmation=True,
+    )
+    tool_context = Mock(spec=ToolContext)
+    tool_context.tool_confirmation = None
+    tool_context.request_confirmation = Mock()
+    args = {"param1": "test_value"}
+
+    result = await tool.run_async(args=args, tool_context=tool_context)
+
+    assert result == {
+        "error": (
+            "This tool call requires confirmation, please approve or reject."
+        )
+    }
+    tool_context.request_confirmation.assert_called_once()
+
+  @pytest.mark.asyncio
+  async def test_run_async_require_confirmation_true_rejected(self):
+    """Test require_confirmation=True with rejection in context."""
+    tool = MCPTool(
+        mcp_tool=self.mock_mcp_tool,
+        mcp_session_manager=self.mock_session_manager,
+        require_confirmation=True,
+    )
+    tool_context = Mock(spec=ToolContext)
+    tool_context.tool_confirmation = Mock(confirmed=False)
+    args = {"param1": "test_value"}
+
+    result = await tool.run_async(args=args, tool_context=tool_context)
+
+    assert result == {"error": "This tool call is rejected."}
+
+  @pytest.mark.asyncio
+  async def test_run_async_require_confirmation_true_confirmed(self):
+    """Test require_confirmation=True with confirmation in context."""
+    tool = MCPTool(
+        mcp_tool=self.mock_mcp_tool,
+        mcp_session_manager=self.mock_session_manager,
+        require_confirmation=True,
+    )
+    tool_context = Mock(spec=ToolContext)
+    tool_context.tool_confirmation = Mock(confirmed=True)
+    args = {"param1": "test_value"}
+
+    with patch(
+        "google.adk.tools.base_authenticated_tool.BaseAuthenticatedTool.run_async",
+        new_callable=AsyncMock,
+    ) as mock_super_run_async:
+      await tool.run_async(args=args, tool_context=tool_context)
+      mock_super_run_async.assert_called_once_with(
+          args=args, tool_context=tool_context
+      )
+
+  @pytest.mark.asyncio
+  async def test_run_async_require_confirmation_callable_true_no_confirmation(
+      self,
+  ):
+    """Test require_confirmation=callable with no confirmation in context."""
+    tool = MCPTool(
+        mcp_tool=self.mock_mcp_tool,
+        mcp_session_manager=self.mock_session_manager,
+        require_confirmation=lambda **kwargs: True,
+    )
+    tool_context = Mock(spec=ToolContext)
+    tool_context.tool_confirmation = None
+    tool_context.request_confirmation = Mock()
+    args = {"param1": "test_value"}
+
+    result = await tool.run_async(args=args, tool_context=tool_context)
+
+    assert result == {
+        "error": (
+            "This tool call requires confirmation, please approve or reject."
+        )
+    }
+    tool_context.request_confirmation.assert_called_once()
+
   def test_init_validation(self):
     """Test that initialization validates required parameters."""
     # This test ensures that the MCPTool properly handles its dependencies