`require_confirmation` callable returning a non-bool skips the confirmation gate: `cast(bool, ...)` is not a runtime check

Required Information

Describe the Bug:

check_require_confirmation returns the confirmation predicate's value through cast(bool, ...). typing.cast is an assertion to the type checker and does nothing at runtime, so whatever the predicate returns arrives at the gate unchanged. A predicate that falls off the end of a branch returns None, if require_confirmation: is false, and the tool runs with no confirmation requested.

Two sites, both on main at c7ffcfa8:

# src/google/adk/tools/function_tool.py:202-205
      return cast(
          bool,
          await self._invoke_callable(self._require_confirmation, args_to_call),
      )

# src/google/adk/tools/mcp_tool/mcp_tool.py:470-473
      return cast(
          bool,
          await self._invoke_callable(self._require_confirmation, args_to_call),
      )

The line just below each of those, return bool(self._require_confirmation), is the static branch and is fine. It is the callable branch that passes the value through.

The declared contract is a bool on both paths: require_confirmation: Union[bool, Callable[..., bool]] = False (function_tool.py:104), require_confirmation: bool | Callable[..., bool] = False (mcp_tool.py:285), and BaseTool.check_require_confirmation is annotated -> bool (base_tool.py:182-184). _invoke_callable is annotated -> Any and returns the callable's value as it is, so nothing between the predicate and if require_confirmation: narrows it (function_tool.py:267-271, mcp_tool.py:487-491).

Steps to Reproduce:

1. pip install "google-adk @ git+https://github.com/google/adk-python@c7ffcfa85a8e8970f6318306479d9c4c110583b2"
2. Save the script under Minimal Reproduction Code below as repro.py.
3. python repro.py

Expected Behavior:

A predicate that does not return a bool has not answered the question, so the gate should treat it as unanswered rather than as "no confirmation needed".

Observed Behavior:

Seven rows, six of them controls. The answered column is the value check_require_confirmation handed back, which is the part that shows the cast did nothing: the annotation says bool, the value is None.

require_confirmation -> did the guarded tool run unconfirmed?
  ran=no   True             (control: must ask)   -> asked for confirmation answered True (bool=True)
  ran=YES  False            (control: must run)   -> no confirmation asked  answered False (bool=True)
  ran=no   callable -> True (control: must ask)   -> asked for confirmation answered True (bool=True)
  ran=YES  callable -> False (control: must run)  -> no confirmation asked  answered False (bool=True)
  ran=no   callable raises  (unanswerable)        -> RuntimeError: policy backend unreachable answered RuntimeError
  ran=no   callable -> 'reason' (truthy non-bool) -> asked for confirmation answered 'amount over limit' (bool=False)
  ran=YES  callable -> None (unhandled branch)    -> no confirmation asked  answered None (bool=False)

Rows one to four are the ordinary paths and show the harness can tell a confirmation request from an execution. Row five is the other kind of non-answer: a predicate that raises propagates out of run_async, so the tool does not run. Row six matters for the fix rather than for the bug: a truthy non-bool asks for confirmation today. Row seven is the finding.

The same question put to McpTool, which needs the mcp extra installed and so is kept as a separate snippet:

  callable -> True  -> answered True (bool=True)
  callable -> False -> answered False (bool=True)
  callable -> None  -> answered None (bool=False)

Environment Details:

• ADK Library Version (pip show google-adk): 2.8.0, installed from main at c7ffcfa85a8e8970f6318306479d9c4c110583b2. I checked that function_tool.py and mcp_tool.py in site-packages are byte identical (sha256) to the same paths at that commit.
• Desktop OS: Windows 11, 10.0.26200
• Python Version (python -V): Python 3.12.10

Model Information:

• Are you using LiteLLM: No
• Which model is being used: none. The repro calls the tool directly and needs no model.

---

Optional Information

Regression:

I did not bisect. The same expression is in the 2.8.0 wheel on PyPI, so this is not new on main.

Logs:

The run prints UserWarning: [EXPERIMENTAL] feature FeatureName.TOOL_CONFIRMATION is enabled. I understand this surface is experimental. That is context for how you want to weigh the report, not an argument that it does not matter.

Additional Context:

What I read in this tracker before filing:

• #4327 (closed) and #6977 (closed) are both fixes on this same predicate path, so the callable form is live and maintained.
• #4625 asked for fail-closed behaviour when no confirmation policy is configured. It was closed, and the answer on the thread was that the current behaviour is by design, since require_confirmation defaults to False and is opt-in. This report is not that request. Here the caller has opted in with a callable, and the gate they configured still does not hold.
• #6461 (open) is a different bypass of the same gate, through A2A.
• Search check: require_confirmation returns 15 issues in this tracker and a nonsense token returns 0, so the search was working. I read the 15 and found no report of this case.

I am not proposing a specific fix, because each of the obvious ones is a behaviour change and the choice is yours. Replacing cast(bool, ...) with bool(...) would give row seven the same wrong answer with a different provenance. A strict isinstance check would start raising for the row six caller, who returns a reason string to mean yes and works today. Treating a falsy non-bool as unanswered changes only the dangerous half, but it also flips require_confirmation=lambda amount: FLAGS.get("confirm") from run to ask for anyone relying on that. Which non-bools stay legal is a design decision. I am happy to send the PR for whichever you choose, with tests.

Minimal Reproduction Code:

"""Does a require_confirmation callable that returns a non-bool skip the gate?"""

import asyncio
from typing import Any
from unittest.mock import MagicMock

from google.adk.agents.invocation_context import InvocationContext
from google.adk.sessions.session import Session
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.tool_context import ToolContext

EXECUTED: list[str] = []


def wire_money(amount: int) -> str:
  """Sends money."""
  EXECUTED.append(f"wired {amount}")
  return "wired"


def make_tool_context() -> ToolContext:
  invocation_context = MagicMock(spec=InvocationContext)
  invocation_context._state_schema = None
  invocation_context.session = MagicMock(spec=Session)
  invocation_context.session.state = MagicMock()
  invocation_context.agent = MagicMock()
  invocation_context.agent.name = "test_agent"
  tool_context = ToolContext(invocation_context=invocation_context)
  tool_context.function_call_id = "call_1"
  return tool_context


async def probe(label: str, require_confirmation: Any) -> None:
  EXECUTED.clear()
  tool = FunctionTool(wire_money, require_confirmation=require_confirmation)
  args = {"amount": 100}
  try:
    answer = await tool.check_require_confirmation(args, make_tool_context())
    answered = f"{answer!r} (bool={isinstance(answer, bool)})"
  except Exception as exc:
    answered = type(exc).__name__
  try:
    result = await tool.run_async(args=args, tool_context=make_tool_context())
    error = result.get("error", "") if isinstance(result, dict) else ""
    asked = "requires confirmation" in str(error)
    outcome = "asked for confirmation" if asked else "no confirmation asked"
  except Exception as exc:
    outcome = f"{type(exc).__name__}: {exc}"[:44]
  ran = "YES" if EXECUTED else "no "
  print(f"  ran={ran}  {label:38} -> {outcome:22} answered {answered}")


def returns_true(amount: int) -> bool:
  return True


def returns_false(amount: int) -> bool:
  return False


def raises(amount: int) -> bool:
  raise RuntimeError("policy backend unreachable")


def returns_reason(amount: int) -> bool:
  return "amount over limit"


def forgot_a_branch(amount: int) -> bool:
  if amount > 1000:
    return True
  # every other amount falls through and returns None


async def main() -> None:
  print("require_confirmation -> did the guarded tool run unconfirmed?")
  await probe("True             (control: must ask)", True)
  await probe("False            (control: must run)", False)
  await probe("callable -> True (control: must ask)", returns_true)
  await probe("callable -> False (control: must run)", returns_false)
  await probe("callable raises  (unanswerable)", raises)
  await probe("callable -> 'reason' (truthy non-bool)", returns_reason)
  await probe("callable -> None (unhandled branch)", forgot_a_branch)


asyncio.run(main())

The McpTool snippet, which additionally needs pip install mcp:

"""Same question for McpTool. Needs the mcp extra."""

import asyncio
from unittest.mock import MagicMock

from google.adk.agents.invocation_context import InvocationContext
from google.adk.sessions.session import Session
from google.adk.tools.mcp_tool.mcp_tool import McpTool
from google.adk.tools.tool_context import ToolContext
from mcp.types import Tool as McpBaseTool


def make_tool_context() -> ToolContext:
  invocation_context = MagicMock(spec=InvocationContext)
  invocation_context._state_schema = None
  invocation_context.session = MagicMock(spec=Session)
  invocation_context.session.state = MagicMock()
  invocation_context.agent = MagicMock()
  invocation_context.agent.name = "test_agent"
  tool_context = ToolContext(invocation_context=invocation_context)
  tool_context.function_call_id = "call_1"
  return tool_context


def forgot_a_branch(amount: int) -> bool:
  if amount > 1000:
    return True


async def main() -> None:
  spec = McpBaseTool(
      name="wire_money",
      description="Sends money.",
      inputSchema={"type": "object", "properties": {"amount": {"type": "integer"}}},
  )
  cases = [
      ("callable -> True ", lambda amount: True),
      ("callable -> False", lambda amount: False),
      ("callable -> None ", forgot_a_branch),
  ]
  for label, predicate in cases:
    tool = McpTool(
        mcp_tool=spec,
        mcp_session_manager=MagicMock(),
        require_confirmation=predicate,
    )
    answer = await tool.check_require_confirmation({"amount": 100}, make_tool_context())
    print(f"  {label} -> answered {answer!r} (bool={isinstance(answer, bool)})")


asyncio.run(main())

How often has this issue occurred?:

• Always (100%)