Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions src/google/adk/agents/invocation_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -202,6 +202,15 @@ class InvocationContext(BaseModel):
plugin_manager: PluginManager = Field(default_factory=PluginManager)
"""The manager for keeping track of plugins in this invocation."""

pending_confirmation_tool: Optional[str] = None
"""The name of the tool that is currently awaiting user confirmation.

When a tool with require_confirmation=True is called, this field is set to
the tool's name. While this field is set, other tools should be gated
(hidden from the model) to prevent bypassing the confirmation requirement.
This is cleared when confirmation is approved or rejected.
"""

_invocation_cost_manager: _InvocationCostManager = PrivateAttr(
default_factory=_InvocationCostManager
)
Expand Down Expand Up @@ -338,6 +347,23 @@ def should_pause_invocation(self, event: Event) -> bool:

return False

def set_pending_confirmation(self, tool_name: str) -> None:
"""Set a tool as pending confirmation.

Args:
tool_name: The name of the tool awaiting confirmation.
"""
self.pending_confirmation_tool = tool_name

def clear_pending_confirmation(self) -> None:
"""Clear the pending confirmation state."""
self.pending_confirmation_tool = None

@property
def has_pending_confirmation(self) -> bool:
"""Check if a tool is currently awaiting confirmation."""
return self.pending_confirmation_tool is not None

# TODO: Move this method from invocation_context to a dedicated module.
# TODO: Converge this method with find_matching_function_call in llm_flows.
def _find_matching_function_call(
Expand Down
18 changes: 18 additions & 0 deletions src/google/adk/agents/llm_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -486,6 +486,24 @@ async def canonical_tools(
tool_union, ctx, self.model, multiple_tools
)
)

# CONFIRMATION GATING: Filter tools if confirmation is pending
# When a tool requires confirmation, we hide all other tools from the model
# to prevent it from bypassing the confirmation requirement.
# See: https://github.com/google/adk-python/issues/3018
if ctx and hasattr(ctx, '_invocation_context'):
inv_ctx = ctx._invocation_context
if hasattr(inv_ctx, 'has_pending_confirmation') and inv_ctx.has_pending_confirmation:
pending_tool_name = inv_ctx.pending_confirmation_tool
logger.info(
f"Tool confirmation pending for '{pending_tool_name}'. "
f"Gating {len(resolved_tools) - 1} other tool(s)."
)
resolved_tools = [
t for t in resolved_tools
if t.name == pending_tool_name
]
Comment on lines +496 to +505

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The conditional check for has_pending_confirmation can be simplified using getattr for better readability and to gracefully handle cases where the attribute might not be present on older objects. Additionally, it's a best practice to use format strings with loggers to defer string formatting until it's certain the message will be logged.

Suggested change
if hasattr(inv_ctx, 'has_pending_confirmation') and inv_ctx.has_pending_confirmation:
pending_tool_name = inv_ctx.pending_confirmation_tool
logger.info(
f"Tool confirmation pending for '{pending_tool_name}'. "
f"Gating {len(resolved_tools) - 1} other tool(s)."
)
resolved_tools = [
t for t in resolved_tools
if t.name == pending_tool_name
]
if getattr(inv_ctx, 'has_pending_confirmation', False):
pending_tool_name = inv_ctx.pending_confirmation_tool
logger.info(
"Tool confirmation pending for '%s'. Gating %d other tool(s).",
pending_tool_name,
len(resolved_tools) - 1,
)
resolved_tools = [
t for t in resolved_tools if t.name == pending_tool_name
]


return resolved_tools

@property
Expand Down
8 changes: 8 additions & 0 deletions src/google/adk/tools/function_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,9 @@ async def run_async(
if 'tool_context' in args_to_show:
args_to_show.pop('tool_context')

# Set pending confirmation state to gate other tools
tool_context.invocation_context.set_pending_confirmation(self.name)

tool_context.request_confirmation(
hint=(
f'Please approve or reject the tool call {self.name}() by'
Expand All @@ -212,7 +215,12 @@ async def run_async(
)
}
elif not tool_context.tool_confirmation.confirmed:
# Clear pending state when confirmation is rejected
tool_context.invocation_context.clear_pending_confirmation()
return {'error': 'This tool call is rejected.'}
else:
# Clear pending state when confirmation is approved
tool_context.invocation_context.clear_pending_confirmation()

Comment on lines 217 to 224

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The call to tool_context.invocation_context.clear_pending_confirmation() is duplicated in both the elif (rejection) and else (approval) branches. To adhere to the DRY (Don't Repeat Yourself) principle, you could refactor this by moving the call to happen once before checking if the confirmation was successful.

For example:

if require_confirmation:
  if not tool_context.tool_confirmation:
    # ... request confirmation and return
  
  # Confirmation has been provided, so clear the pending state.
  tool_context.invocation_context.clear_pending_confirmation()

  if not tool_context.tool_confirmation.confirmed:
    return {'error': 'This tool call is rejected.'}

return await self._invoke_callable(self.func, args_to_call)

Expand Down
156 changes: 156 additions & 0 deletions tests/test_confirmation_gating_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,156 @@
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Unit tests for tool confirmation gating functionality.
Tests the fix for Issue #3018: When a tool requires confirmation,
other tools should be hidden from the model to prevent bypassing confirmation.
"""

import pytest
from google.adk.agents.invocation_context import InvocationContext


def test_invocation_context_pending_confirmation():
"""Test InvocationContext pending confirmation state management."""

# Create a mock invocation context with minimal required fields
from google.adk.sessions.session import Session
from google.adk.agents.base_agent import BaseAgent
from google.adk.sessions.in_memory_session_service import InMemorySessionService
Comment on lines +30 to +32

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

According to PEP 8, imports should be at the top of the file. Placing them inside functions can hurt readability and hide dependencies. Please move these imports to the top-level scope of the module. This also applies to the imports in test_canonical_tools_filters_when_confirmation_pending.


session_service = InMemorySessionService()
session = Session(
id="test-session",
app_name="test-app",
user_id="test-user"
)

# Mock agent
class MockAgent(BaseAgent):
pass

agent = MockAgent(name="test_agent")

inv_ctx = InvocationContext(
invocation_id="test-invocation",
session_service=session_service,
session=session,
agent=agent
)

# Test initial state
assert inv_ctx.has_pending_confirmation is False
assert inv_ctx.pending_confirmation_tool is None

# Test setting pending confirmation
inv_ctx.set_pending_confirmation("my_tool")
assert inv_ctx.has_pending_confirmation is True
assert inv_ctx.pending_confirmation_tool == "my_tool"

# Test clearing pending confirmation
inv_ctx.clear_pending_confirmation()
assert inv_ctx.has_pending_confirmation is False
assert inv_ctx.pending_confirmation_tool is None

print("✅ InvocationContext confirmation state management works correctly")


@pytest.mark.asyncio
async def test_canonical_tools_filters_when_confirmation_pending():
"""Test that canonical_tools() filters tools when confirmation is pending."""

from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.function_tool import FunctionTool
from google.adk.sessions.session import Session
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.agents.invocation_context import InvocationContext
from google.adk.agents.readonly_context import ReadonlyContext
Comment on lines +75 to +80

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

According to PEP 8, imports should be at the top of the file. Placing them inside functions can hurt readability and hide dependencies. Please move these imports to the top-level scope of the module.


# Define test tools
def tool_a(x: int) -> str:
"""Tool A that requires confirmation."""
return f"A: {x}"

def tool_b(y: int) -> str:
"""Tool B that should be gated."""
return f"B: {y}"

def tool_c(z: int) -> str:
"""Tool C that should also be gated."""
return f"C: {z}"

# Create agent with multiple tools
agent = LlmAgent(
model='gemini-2.5-flash',
name='test_agent',
tools=[
FunctionTool(tool_a),
FunctionTool(tool_b),
FunctionTool(tool_c)
]
)

# Create invocation context
session_service = InMemorySessionService()
session = Session(
id="test-session",
app_name="test-app",
user_id="test-user"
)

inv_ctx = InvocationContext(
invocation_id="test-invocation",
session_service=session_service,
session=session,
agent=agent
)

readonly_ctx = ReadonlyContext(invocation_context=inv_ctx)

# Test 1: All tools available when no confirmation pending
all_tools = await agent.canonical_tools(readonly_ctx)
assert len(all_tools) == 3, f"Expected 3 tools, got {len(all_tools)}"
tool_names = {t.name for t in all_tools}
assert tool_names == {"tool_a", "tool_b", "tool_c"}
print("✅ All tools available when no confirmation pending")

# Test 2: Only pending tool available when confirmation pending
inv_ctx.set_pending_confirmation("tool_a")
filtered_tools = await agent.canonical_tools(readonly_ctx)
assert len(filtered_tools) == 1, f"Expected 1 tool, got {len(filtered_tools)}"
assert filtered_tools[0].name == "tool_a"
print("✅ Only tool_a available when tool_a confirmation pending")

# Test 3: All tools available again after clearing
inv_ctx.clear_pending_confirmation()
all_tools_again = await agent.canonical_tools(readonly_ctx)
assert len(all_tools_again) == 3
print("✅ All tools available again after clearing confirmation")


if __name__ == "__main__":
import asyncio

print("Running Confirmation Gating Unit Tests")
print("=" * 50)

test_invocation_context_pending_confirmation()
print()

asyncio.run(test_canonical_tools_filters_when_confirmation_pending())
print()

print("All unit tests passed! ✅")
Loading