feat: add tool call confirmation before execution (human-in-the-loop)

Add an admin-configurable option to require user confirmation before
tool/MCP calls are executed. When enabled, the backend emits a
'confirmation' event via the existing WebSocket event_caller mechanism,
showing the tool name and parameters. The user can approve or deny
each call.

This addresses a privacy/security concern where LLMs may send
sensitive conversation data to external tool servers without user
awareness.

Changes:
- config.py: Add ENABLE_TOOL_CALL_CONFIRMATION env var and register
  in config dict under 'tool.call.confirmation.enable'
- main.py: Expose as config.features.enable_tool_call_confirmation
  in /api/config endpoint
- middleware.py: Add confirmation check in tool_call_handler() before
  tool execution, with _ToolCallDenied sentinel for clean flow control
- stores/index.ts: Add enable_tool_call_confirmation to Config type

Addresses #16940
This commit is contained in:
GOVINSAGA 2026-07-06 23:26:33 +05:30
parent 10724d057a
commit 8282e35782
4 changed files with 37 additions and 0 deletions

View file

@ -1128,6 +1128,12 @@ WEB_SEARCH_CONFIRMATION_CONTENT = os.getenv(
'Your query will be sent to the configured web search provider.',
)
####################################
# Tool Call Confirmation
####################################
ENABLE_TOOL_CALL_CONFIRMATION = os.getenv('ENABLE_TOOL_CALL_CONFIRMATION', 'False').lower() == 'true'
WEB_SEARCH_ENGINE = os.getenv('WEB_SEARCH_ENGINE', '')
BYPASS_WEB_SEARCH_EMBEDDING_AND_RETRIEVAL = (
@ -2861,6 +2867,7 @@ DEFAULT_CONFIG = {
'web.search.enable': ENABLE_WEB_SEARCH,
'web.search.confirmation.enable': ENABLE_WEB_SEARCH_CONFIRMATION,
'web.search.confirmation.content': WEB_SEARCH_CONFIRMATION_CONTENT,
'tool.call.confirmation.enable': ENABLE_TOOL_CALL_CONFIRMATION,
'web.search.engine': WEB_SEARCH_ENGINE,
'web.search.bypass_embedding_and_retrieval': BYPASS_WEB_SEARCH_EMBEDDING_AND_RETRIEVAL,
'web.search.bypass_web_loader': BYPASS_WEB_SEARCH_WEB_LOADER,

View file

@ -1868,6 +1868,7 @@ async def get_app_config(request: Request):
'web.search.enable',
'web.search.confirmation.enable',
'web.search.confirmation.content',
'tool.call.confirmation.enable',
'code_execution.enable',
'code_interpreter.enable',
'image_generation.enable',
@ -1936,6 +1937,7 @@ async def get_app_config(request: Request):
'enable_web_search': config.get('web.search.enable'),
'enable_web_search_confirmation': config.get('web.search.confirmation.enable'),
'web_search_confirmation_content': config.get('web.search.confirmation.content'),
'enable_tool_call_confirmation': config.get('tool.call.confirmation.enable'),
'enable_code_execution': config.get('code_execution.enable'),
'enable_code_interpreter': config.get('code_interpreter.enable'),
'enable_image_generation': config.get('image_generation.enable'),

View file

@ -1082,6 +1082,11 @@ async def terminal_event_handler(
)
class _ToolCallDenied(Exception):
"""Sentinel exception raised when the user denies a tool call confirmation."""
pass
async def chat_completion_tools_handler(
request: Request, body: dict, extra_params: dict, user: UserModel, models, tools
) -> tuple[dict, dict]:
@ -1189,6 +1194,26 @@ async def chat_completion_tools_handler(
allowed_params = spec.get('parameters', {}).get('properties', {}).keys()
tool_function_params = {k: v for k, v in tool_function_params.items() if k in allowed_params}
# Tool call confirmation: ask user before executing
if await Config.get('tool.call.confirmation.enable') and event_caller:
params_display = json.dumps(tool_function_params, indent=2, ensure_ascii=False)
confirmation = await event_caller(
{
'type': 'confirmation',
'data': {
'title': f'Allow tool call: {tool_function_name}?',
'message': f'The model wants to call **{tool_function_name}** with:\n```json\n{params_display}\n```',
},
}
)
if not confirmation:
tool_result = json.dumps({
'status': 'denied',
'message': f'Tool call to {tool_function_name} was denied by user.',
})
# Skip tool execution; jump to result processing
raise _ToolCallDenied()
if tool.get('direct', False):
tool_result = await event_caller(
{
@ -1206,6 +1231,8 @@ async def chat_completion_tools_handler(
tool_function = tool['callable']
tool_result = await tool_function(**tool_function_params)
except _ToolCallDenied:
pass # tool_result already set with denial message
except Exception as e:
tool_result = str(e)

View file

@ -294,6 +294,7 @@ type Config = {
enable_web_search?: boolean;
enable_web_search_confirmation?: boolean;
web_search_confirmation_content?: string;
enable_tool_call_confirmation?: boolean;
enable_google_drive_integration: boolean;
enable_onedrive_integration: boolean;
enable_image_generation: boolean;