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

[mahirhir]

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:

• Tool Context not passed to require_confirmation function in MCPTool #4327 (closed) and McpTool's require_confirmation gate doesn't set skip_summarization #6977 (closed) are both fixes on this same predicate path, so the callable form is live and maintained.
• Fail closed when high-risk tools lack explicit confirmation policy #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.
• Human-in-the-loop tool confirmation is forgeable by an A2A peer (self-approves dangerous tools) #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%)

[mahirhir]

Adding a runnable reproduction. The report above was read off main at c7ffcfa8; this runs against the released package, so the behaviour is in what users are installing today, not only on the branch.

It is offline: no API key, no network, no model call. It drives FunctionTool through its public run_async, which is the path the runner takes, rather than patching anything inside google.adk.

pip install google-adk && python demo_adk.py

Output on google-adk 2.8.0, Python 3.12.3:

google-adk 2.8.0

controls -- these fix the range of the reading below
  require_confirmation -> True                held: True   tool ran: False
  require_confirmation -> False               held: False  tool ran: True

trial
  require_confirmation -> None                held: False  tool ran: True

discrimination -- which accidental values run the tool?
  require_confirmation -> '' (empty str)      held: False  tool ran: True
  require_confirmation -> [] (empty list)     held: False  tool ran: True
  require_confirmation -> 0 (int)             held: False  tool ran: True
  require_confirmation -> 'no' (truthy str)   held: True   tool ran: False

The two controls run first and gate the trial. If True had not held the call, or False had not run it, the script prints UNTESTABLE and exits 2 instead of reporting the trial, because a reading with no established range is not evidence. Exit 1 means reproduced, 0 means not reproduced on that version.

It measures a side effect, not a return value: the tool function appends to a module-level list. What a confirmation gate protects is a real write, and a return value can be discarded while a write cannot.

The discrimination changes what the fix has to cover

It is not only None. Every falsy return value runs the tool, because the value reaches a plain truth test with nothing in between.

returned by the predicate	confirmation held	tool ran
None	no	yes
''	no	yes
[]	no	yes
0	no	yes
'no'	yes	no

'' is the realistic one. "Return the objection, empty string when there is none" is an ordinary way to write a policy function, and here it silently means "no confirmation needed". A truthy non-bool such as 'no' happens to land in the safe direction, which is why this does not surface in casual testing: the mistake that looks most like a bug behaves correctly, and the mistake that looks harmless does not.

So a repair that special-cases None still ships '' and 0 running the tool.

Suggested direction

1. Fail closed on anything that is not an explicit False. True/False behaviour is unchanged and every accidental value moves to the safe side.
2. Raise a ValueError naming the tool and the returned type, so a mistyped policy fails at the gate rather than at the write. That is a behaviour change for anyone relying on truthiness, which argues for shipping it with 1 rather than instead of it.

Either way the cast should go. A cast on a security-relevant branch reads to a reviewer as a check and is not one, so the code claims to have validated something it never looked at.

Happy to open a PR with the change plus this reproduction as a regression test, if a maintainer would like it in that form.

demo_adk.py

"""Runnable reproduction: a require_confirmation predicate that returns None runs the tool.

Offline. No API key, no network, no model call. It drives the library's own FunctionTool
through its public run_async, which is the same path the runner takes, rather than
patching anything inside google.adk. A demo that monkeypatches the library proves
nothing about the library.

    pip install google-adk && python demo_adk.py

Exit code 1 means the defect reproduced, 0 means it did not on this version, and 2 means
the controls did not behave so the trial has no established range and must not be counted.
"""

import asyncio
import sys
from typing import Any

from google.adk.agents.invocation_context import InvocationContext
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.tool_context import ToolContext

EXECUTED: list[str] = []
DECISION: list[Any] = [False]


def predicate(**_kwargs: Any) -> Any:
    # A real predicate falls off the end of an if-chain and returns None. The missing
    # branch is not a syntax error and not a runtime type error, so nothing reports it.
    return DECISION[0]


def delete_everything() -> str:
    # The evidence is that this line ran, not what the call returned. A confirmation
    # gate protects a real write, and a return value can be discarded while a write cannot.
    EXECUTED.append("delete_everything")
    return "deleted"


async def make_tool_context() -> ToolContext:
    svc = InMemorySessionService()
    session = await svc.create_session(app_name="demo", user_id="u1")
    ctx = InvocationContext(
        session_service=svc, invocation_id="i1", session=session, agent=None
    )
    return ToolContext(ctx, function_call_id="fc_1")


async def trial(label: str, decision: Any) -> object:
    EXECUTED.clear()
    DECISION[0] = decision
    tool = FunctionTool(func=delete_everything, require_confirmation=predicate)
    tc = await make_tool_context()
    try:
        result = await tool.run_async(args={}, tool_context=tc)
    except Exception as exc:
        print(f"  require_confirmation -> {label:<18}  raised {type(exc).__name__}")
        return "raised"
    held = isinstance(result, dict) and "confirmation" in str(result.get("error", ""))
    ran = bool(EXECUTED)
    print(f"  require_confirmation -> {label:<18}  held: {str(held):<5}  tool ran: {ran}")
    return ran


async def main() -> int:
    from importlib.metadata import version

    print(f"google-adk {version('google-adk')}")
    print()

    print("controls -- these fix the range of the reading below")
    held = await trial("True", True)
    ran = await trial("False", False)

    if held is not False or ran is not True:
        print()
        print("UNTESTABLE: the controls did not behave, so the trial below means nothing.")
        return 2

    print()
    print("trial")
    unanswered = await trial("None", None)

    print()
    print("discrimination -- which accidental values run the tool?")
    for label, value in (("'' (empty str)", ""), ("[] (empty list)", []),
                         ("0 (int)", 0), ("'no' (truthy str)", "no")):
        await trial(label, value)

    print()
    if unanswered is not True:
        print("Not reproduced on this version.")
        return 0

    print("DEFECT REPRODUCED: an unhandled predicate branch ran the guarded tool.")
    print("  google/adk/tools/function_tool.py, check_require_confirmation, returns")
    print("  cast(bool, await self._invoke_callable(...)). typing.cast performs no")
    print("  runtime check, so the annotation says bool while the value stays None.")
    print("  run_async then tests `if require_confirmation:` and falls through to the")
    print("  invocation. The declared return type is the only thing asserting a bool,")
    print("  and a declared type is not a check.")
    print("  google/adk/tools/mcp_tool/mcp_tool.py has the same cast and is not")
    print("  exercised here because it needs a live MCP server.")
    return 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))