confirmation request replaced by pre tool hook ask decision

Author: mkoruszowic

examples/chat/file_explorer_agent.py

@@ -32,7 +32,7 @@
 from ragbits.agents import Agent, ToolCallResult
 from ragbits.agents._main import AgentRunContext
 from ragbits.agents.confirmation import ConfirmationRequest
-from ragbits.agents.tool import requires_confirmation
+from ragbits.agents.hooks.confirmation import requires_confirmation_hook
 from ragbits.chat.interface import ChatInterface
 from ragbits.chat.interface.types import (
     ChatContext,
@@ -281,7 +281,6 @@ def search_files(pattern: str, directory: str = "") -> str:
     return json.dumps(result, indent=2)
 
 
-@requires_confirmation
 def create_file(filepath: str, content: str, context: AgentRunContext | None = None) -> str:
     """
     Create a new file with content. Requires confirmation.
@@ -324,7 +323,6 @@ def create_file(filepath: str, content: str, context: AgentRunContext | None = N
     return json.dumps(result, indent=2)
 
 
-@requires_confirmation
 def delete_file(filepath: str, context: AgentRunContext | None = None) -> str:
     """
     Delete a file. Requires confirmation.
@@ -366,7 +364,6 @@ def delete_file(filepath: str, context: AgentRunContext | None = None) -> str:
     return json.dumps(result, indent=2)
 
 
-@requires_confirmation
 def move_file(source: str, destination: str, context: AgentRunContext | None = None) -> str:
     """
     Move or rename a file. Requires confirmation.
@@ -424,7 +421,6 @@ def move_file(source: str, destination: str, context: AgentRunContext | None = N
     return json.dumps(result, indent=2)
 
 
-@requires_confirmation
 def create_directory(dirpath: str, context: AgentRunContext | None = None) -> str:
     """
     Create a new directory. Requires confirmation.
@@ -459,7 +455,6 @@ def create_directory(dirpath: str, context: AgentRunContext | None = None) -> st
     return json.dumps(result, indent=2)
 
 
-@requires_confirmation
 def delete_directory(dirpath: str, context: AgentRunContext | None = None) -> str:
     """
     Delete an empty directory. Requires confirmation.
@@ -545,6 +540,13 @@ def __init__(self) -> None:
             delete_directory,
         ]
 
+        # Create confirmation hook for destructive tools
+        self.hooks = [
+            requires_confirmation_hook(
+                tools=["create_file", "delete_file", "move_file", "create_directory", "delete_directory"]
+            )
+        ]
+
     async def chat(  # noqa: PLR0912
         self,
         message: str,
@@ -554,10 +556,10 @@ async def chat(  # noqa: PLR0912
         """
         Chat implementation with non-blocking confirmation support.
 
-        The agent will check context.confirmed_tools for any confirmations.
-        If a tool needs confirmation but hasn't been confirmed yet, it will
+        The agent will check context.confirmed_hooks for any confirmations.
+        If a hook needs confirmation but hasn't been confirmed yet, it will
         yield a ConfirmationRequest and exit. The frontend will then send a
-        new request with the confirmation in context.confirmed_tools.
+        new request with the confirmation in context.confirmed_tools (mapped to confirmed_hooks).
         """
         # Create agent with history passed explicitly
         agent: Agent = Agent(
@@ -578,15 +580,17 @@ async def chat(  # noqa: PLR0912
             Restricted to: {TEMP_DIR}
             """,
             tools=self.tools,  # type: ignore[arg-type]
+            hooks=self.hooks,
             history=history,
         )
 
-        # Create agent context with confirmed_tools from the request context
+        # Create agent context with confirmed_hooks from the request context
         agent_context: AgentRunContext = AgentRunContext()
 
-        # Pass confirmed_tools from ChatContext to AgentRunContext
+        # Pass confirmed_tools from ChatContext to AgentRunContext.confirmed_hooks
+        # (Frontend still uses confirmed_tools but we map it to confirmed_hooks)
         if context.confirmed_tools:
-            agent_context.confirmed_tools = context.confirmed_tools
+            agent_context.confirmed_hooks = context.confirmed_tools
 
         # Run agent in streaming mode with the message and history
         async for response in agent.run_streaming(

packages/ragbits-agents/src/ragbits/agents/__init__.py

@@ -15,7 +15,6 @@
     HookManager,
 )
 from ragbits.agents.post_processors.base import PostProcessor, StreamingPostProcessor
-from ragbits.agents.tool import requires_confirmation
 from ragbits.agents.tools import LongTermMemory, MemoryEntry, create_memory_tools
 from ragbits.agents.types import QuestionAnswerAgent, QuestionAnswerPromptInput, QuestionAnswerPromptOutput
 
@@ -40,5 +39,4 @@
     "ToolCall",
     "ToolCallResult",
     "create_memory_tools",
-    "requires_confirmation",
 ]

packages/ragbits-agents/src/ragbits/agents/_main.py

@@ -1,6 +1,4 @@
 import asyncio
-import hashlib
-import json
 import types
 import uuid
 from collections.abc import AsyncGenerator, AsyncIterator, Callable
@@ -216,9 +214,9 @@ class AgentRunContext(BaseModel, Generic[DepsT]):
     """Whether to stream events from downstream agents when tools execute other agents."""
     downstream_agents: dict[str, "Agent"] = Field(default_factory=dict)
     """Registry of all agents that participated in this run"""
-    confirmed_tools: list[dict[str, Any]] | None = Field(
+    confirmed_hooks: list[dict[str, Any]] | None = Field(
         default=None,
-        description="List of confirmed/declined tools from the frontend",
+        description="List of confirmed/declined hooks. Each entry has 'confirmation_id' and 'confirmed' (bool)",
     )
 
     def register_agent(self, agent: "Agent") -> None:
@@ -968,6 +966,7 @@ async def _execute_tool(
         # Execute PRE_TOOL hooks with chaining
         pre_tool_result = await self.hook_manager.execute_pre_tool(
             tool_call=tool_call,
+            context=context,
         )
 
         # Check decision
@@ -979,81 +978,28 @@ async def _execute_tool(
                 result=pre_tool_result.reason or "Tool execution denied",
             )
             return
-        # TODO: possible place/way of replacing confirmation by handling ask decision
-        # elif pre_tool_result.decision == "ask":
-        #     confirmation_id = hashlib.sha256(
-        #         f"{tool_call.name}:{json.dumps(tool_call.arguments, sort_keys=True)}".encode()
-        #     ).hexdigest()[:CONFIRMATION_ID_LENGTH]
-        #
-        #     request = ConfirmationRequest(
-        #         confirmation_id=confirmation_id,
-        #         tool_name=tool_call.name,
-        #         tool_description=tool.description or "",
-        #         arguments=tool_call.arguments,
-        #     )
-        #
-        #     # Yield confirmation request (will be streamed to frontend)
-        #     yield request
-        #
-        #     yield ToolCallResult(
-        #         id=tool_call.id,
-        #         name=tool_call.name,
-        #         arguments=tool_call.arguments,
-        #         result=pre_tool_result.reason or "Tool requires user confirmation",
-        #     )
-        #     return
-
-        # Always update arguments (chained from hooks)
-        tool_call.arguments = pre_tool_result.arguments
-
-        # Check if tool requires confirmation
-        if tool.requires_confirmation:
-            # Check if this tool has been confirmed in the context
-            confirmed_tools = context.confirmed_tools or []
-
-            # Generate a stable confirmation ID based on tool name and arguments
-            confirmation_id = hashlib.sha256(
-                f"{tool_call.name}:{json.dumps(tool_call.arguments, sort_keys=True)}".encode()
-            ).hexdigest()[:CONFIRMATION_ID_LENGTH]
-
-            # Check if this specific tool call has been confirmed or declined
-            is_confirmed = any(
-                ct.get("confirmation_id") == confirmation_id and ct.get("confirmed") for ct in confirmed_tools
-            )
-            is_declined = any(
-                ct.get("confirmation_id") == confirmation_id and not ct.get("confirmed", True) for ct in confirmed_tools
+        # Handle "ask" decision from hooks
+        elif pre_tool_result.decision == "ask":
+            request = ConfirmationRequest(
+                confirmation_id=pre_tool_result.confirmation_id or "",
+                tool_name=tool_call.name,
+                tool_description=pre_tool_result.reason or "Hook requires user confirmation",
+                arguments=pre_tool_result.arguments,
             )
 
-            if is_declined:
-                # Tool was explicitly declined - skip execution entirely
-                yield ToolCallResult(
-                    id=tool_call.id,
-                    name=tool_call.name,
-                    arguments=tool_call.arguments,
-                    result="❌ Action declined by user",
-                )
-                return
-
-            if not is_confirmed:
-                # Tool not confirmed yet - create and yield confirmation request
-                request = ConfirmationRequest(
-                    confirmation_id=confirmation_id,
-                    tool_name=tool_call.name,
-                    tool_description=tool.description or "",
-                    arguments=tool_call.arguments,
-                )
+            # Yield confirmation request (will be streamed to frontend)
+            yield request
 
-                # Yield confirmation request (will be streamed to frontend)
-                yield request
+            yield ToolCallResult(
+                id=tool_call.id,
+                name=tool_call.name,
+                arguments=tool_call.arguments,
+                result=pre_tool_result.reason or "Hook requires user confirmation",
+            )
+            return
 
-                # Yield a pending result and exit without executing
-                yield ToolCallResult(
-                    id=tool_call.id,
-                    name=tool_call.name,
-                    arguments=tool_call.arguments,
-                    result="⏳ Awaiting user confirmation",
-                )
-                return
+        # Always update arguments (chained from hooks)
+        tool_call.arguments = pre_tool_result.arguments
 
         tool_error: Exception | None = None
 

packages/ragbits-agents/src/ragbits/agents/hooks/__init__.py

@@ -45,6 +45,7 @@ async def validate_input(input_data: PreToolInput) -> PreToolOutput | None:
 """
 
 from ragbits.agents.hooks.base import Hook
+from ragbits.agents.hooks.confirmation import requires_confirmation_hook
 from ragbits.agents.hooks.manager import HookManager
 from ragbits.agents.hooks.types import (
     EventType,
@@ -65,4 +66,6 @@ async def validate_input(input_data: PreToolInput) -> PreToolOutput | None:
     "PostToolOutput",
     "PreToolInput",
     "PreToolOutput",
+    # Helper functions
+    "requires_confirmation_hook",
 ]

packages/ragbits-agents/src/ragbits/agents/hooks/confirmation.py

@@ -0,0 +1,49 @@
+"""
+Helper functions for creating common hooks.
+
+This module provides factory functions for creating commonly used hooks.
+"""
+
+from ragbits.agents.hooks.base import Hook
+from ragbits.agents.hooks.types import EventType, PreToolInput, PreToolOutput
+
+
+def requires_confirmation_hook(tools: list[str] | None = None, priority: int = 1) -> Hook:
+    """
+    Create a hook that requires user confirmation before tool execution.
+
+    This replaces the @requires_confirmation decorator with a hook-based approach.
+    The hook returns "ask" decision, which causes the agent to yield a ConfirmationRequest
+    and wait for user approval/decline.
+
+    Args:
+        tools: List of tool names to require confirmation for. If None, applies to all tools.
+        priority: Hook priority (default: 1, runs first)
+
+    Returns:
+        Hook configured to require confirmation
+
+    Example:
+        ```python
+        from ragbits.agents import Agent
+        from ragbits.agents.hooks.helpers import requires_confirmation_hook
+
+        agent = Agent(tools=[delete_file, send_email],
+                      hooks=[requires_confirmation_hook(tools=["delete_file", "send_email"])])
+        ```
+    """
+
+    async def confirm_hook(input_data: PreToolInput) -> PreToolOutput:
+        """Hook that always returns 'ask' to require confirmation."""
+        return PreToolOutput(
+            arguments=input_data.tool_call.arguments,
+            decision="ask",
+            reason=f"Tool '{input_data.tool_call.name}' requires user confirmation",
+        )
+
+    return Hook(
+        event_type=EventType.PRE_TOOL,
+        callback=confirm_hook,  # type: ignore[arg-type]
+        tools=tools,
+        priority=priority,
+    )