Draft pull: feature request_human_approval [work in progress]

lib/crewai/src/crewai/tools/base_tool.py

@@ -41,6 +41,7 @@
 from crewai.tools.tool_failure import ToolFailure, ToolFailurePolicy, ToolFailureReason
 from crewai.types.callback import SerializableCallable, _resolve_dotted_path
 from crewai.utilities.string_utils import sanitize_tool_name
+import logging, time
 
 
 P = ParamSpec("P")
@@ -181,6 +182,10 @@ def _serialize_result_schema(
         default=False,
         description="Flag to check if the tool should be the final agent answer.",
     )
+    requires_human_approval: Any = Field(
+        default=False,
+        description="Flag to check if tool execution requires explicit human approval before running. Accepts a boolean or a predicate function evaluating args/kwargs.",
+    )
     max_usage_count: int | None = Field(
         default=None,
         description="Maximum number of times this tool can be used. None means unlimited usage.",
@@ -322,7 +327,46 @@ def _claim_usage(self) -> ToolFailure | None:
                 )
             self.current_usage_count += 1
             return None
-
+
+    def _should_require_approval(self, *args: Any, **kwargs: Any) -> bool:
+        """Evaluates if human approval is needed, supporting both booleans and predicate functions."""
+        logger = logging.getLogger(__name__)
+
+        approval_flag = getattr(self, "requires_human_approval", False)
+        if callable(approval_flag):
+            try:
+                return approval_flag(*args, **kwargs)
+            except Exception as e:
+                logger.error(f"Error in requires_human_approval predicate for {self.name}: {e}")
+                return True
+        return bool(approval_flag)
+
+    def _request_human_approval(self, *args: Any, **kwargs: Any) -> bool:
+        """Requests human approval, logs the audit trail, and handles timeouts securely."""
+        logger = logging.getLogger(__name__)
+
+        print(f"Human approval required for tool: {self.name}")
+        print(f"Description: {self.description}")
+        print(f"Args: {args}, Kwargs: {kwargs}")
+
+        audit_data = {
+            "tool": self.name,
+            "args": args,
+            "kwargs": kwargs,
+            "timestamp": time.time()
+        }
+
+        try:
+            # Note: For async/Slack workflows, developers can subclass and override this specific block.
+            response = input("Approve execution? (yes/no): ").strip().lower()
+            approved = response in ["yes", "y", "approve"]
+            logger.info(f"Audit [HITL]: Tool={self.name}, Approved={approved}, Timestamp={audit_data['timestamp']}, Args={args}, Kwargs={kwargs}")
+            return approved
+        except Exception as e:
+            logger.error(f"Audit [HITL]: Tool={self.name}, Approved=False (Timeout/Error), Error={e}")
+            # Fail closed on timeout to prevent agent from retrying endlessly
+            raise TimeoutError(f"Approval timed out for '{self.name}'. Tool execution blocked.")
+
     def run(
         self,
         *args: Any,
@@ -335,6 +379,10 @@ def run(
         if limit_error:
             return limit_error
 
+        if self._should_require_approval(*args, **kwargs):
+            if not self._request_human_approval(*args, **kwargs):
+                raise PermissionError(f"Action '{self.name}' explicitly rejected by human.")
+
         result = self._run(*args, **kwargs)
 
         if asyncio.iscoroutine(result):
@@ -363,6 +411,10 @@ async def arun(
         if limit_error:
             return limit_error
 
+        if self._should_require_approval(*args, **kwargs):
+            if not self._request_human_approval(*args, **kwargs):
+                raise PermissionError(f"Action '{self.name}' explicitly rejected by human.")
+
         return await self._arun(*args, **kwargs)
 
     async def _arun(
@@ -416,6 +468,7 @@ def to_structured_tool(self) -> CrewStructuredTool:
             current_usage_count=self.current_usage_count,
             cache_function=self.cache_function,
             tool_failure_policy=self.tool_failure_policy,
+            requires_human_approval=self.requires_human_approval,
         )
         structured_tool._original_tool = self
         return structured_tool
@@ -686,6 +739,7 @@ def tool(
     result_schema: type[BaseModel] | None = ...,
     result_as_answer: bool = ...,
     max_usage_count: int | None = ...,
+    requires_human_approval: Any = ...,
 ) -> Callable[[Callable[P2, R2]], Tool[P2, R2]]: ...
 
 
@@ -695,6 +749,7 @@ def tool(
     result_schema: type[BaseModel] | None = ...,
     result_as_answer: bool = ...,
     max_usage_count: int | None = ...,
+    requires_human_approval: Any = ...,
 ) -> Callable[[Callable[P2, R2]], Tool[P2, R2]]: ...
 
 
@@ -703,6 +758,7 @@ def tool(
     result_schema: type[BaseModel] | None = None,
     result_as_answer: bool = False,
     max_usage_count: int | None = None,
+    requires_human_approval: Any = False,
 ) -> Tool[P2, R2] | Callable[[Callable[P2, R2]], Tool[P2, R2]]:
     """Decorator to create a Tool from a function.
 
@@ -769,6 +825,7 @@ def _make_tool(f: Callable[P2, R2]) -> Tool[P2, R2]:
                 result_as_answer=result_as_answer,
                 max_usage_count=max_usage_count,
                 current_usage_count=0,
+                requires_human_approval=requires_human_approval,
             )
 
         return _make_tool

lib/crewai/src/crewai/tools/structured_tool.py

@@ -28,6 +28,7 @@
     generate_model_description,
 )
 from crewai.utilities.string_utils import sanitize_tool_name
+import logging
 
 
 def _serialize_schema(v: type[BaseModel] | None) -> dict[str, Any] | None:
@@ -210,6 +211,7 @@ class CrewStructuredTool(BaseModel):
     func: Any = Field(default=None, exclude=True)
     result_as_answer: bool = Field(default=False)
     max_usage_count: int | None = Field(default=None)
+    requires_human_approval: Any = Field(default=False)
     current_usage_count: int = Field(default=0)
     tool_failure_policy: ToolFailurePolicy | None = Field(default=None)
     cache_function: Any = Field(default=None, exclude=True)
@@ -377,6 +379,45 @@ def _parse_args(self, raw_args: str | dict[str, Any]) -> dict[str, Any]:
             hint = build_schema_hint(self.args_schema)
             raise ValueError(f"Arguments validation failed: {e}{hint}") from e
 
+    def _should_require_approval(self, *args: Any, **kwargs: Any) -> bool:
+            """Evaluates if human approval is needed, supporting both booleans and predicate functions."""
+            logger = logging.getLogger(__name__)
+
+            approval_flag = getattr(self, "requires_human_approval", False)
+            if callable(approval_flag):
+                try:
+                    return approval_flag(*args, **kwargs)
+                except Exception as e:
+                    logger.error(f"Error in requires_human_approval predicate for {self.name}: {e}")
+                    return True
+            return bool(approval_flag)
+
+    def _request_human_approval(self, *args: Any, **kwargs: Any) -> bool:
+        """Requests human approval, logs the audit trail, and handles timeouts securely."""
+        logger = logging.getLogger(__name__)
+
+        print(f"Human approval required for tool: {self.name}")
+        print(f"Description: {self.description}")
+        print(f"Args: {args}, Kwargs: {kwargs}")
+
+        audit_data = {
+            "tool": self.name,
+            "args": args,
+            "kwargs": kwargs,
+            "timestamp": time.time()
+            }
+
+        try:
+            # Note: For async/Slack workflows, developers can subclass and override this specific block.
+            response = input("Approve execution? (yes/no): ").strip().lower()
+            approved = response in ["yes", "y", "approve"]
+            logger.info(f"Audit [HITL]: Tool={self.name}, Approved={approved}, Timestamp={audit_data['timestamp']}, Args={args}, Kwargs={kwargs}")
+            return approved
+        except Exception as e:
+            logger.error(f"Audit [HITL]: Tool={self.name}, Approved=False (Timeout/Error), Error={e}")
+            # Fail closed on timeout to prevent agent from retrying endlessly
+            raise TimeoutError(f"Approval timed out for '{self.name}'. Tool execution blocked.")    
+
     async def ainvoke(
         self,
         input: str | dict[str, Any],
@@ -402,6 +443,10 @@ async def ainvoke(
 
         self._increment_usage_count()
 
+        if self._should_require_approval(**parsed_args):
+            if not self._request_human_approval(**parsed_args):
+                raise PermissionError(f"Action '{self.name}' explicitly rejected by human.")
+
         try:
             if inspect.iscoroutinefunction(self.func):
                 return await self.func(**parsed_args, **kwargs)
@@ -437,6 +482,10 @@ def invoke(
 
         self._increment_usage_count()
 
+        if self._should_require_approval(**parsed_args):
+            if not self._request_human_approval(**parsed_args):
+                raise PermissionError(f"Action '{self.name}' explicitly rejected by human.")
+
         if inspect.iscoroutinefunction(self.func):
             return asyncio.run(self.func(**parsed_args, **kwargs))
 

lib/crewai/tests/tools/test_base_tool.py

@@ -792,3 +792,79 @@ def test_prompt_rendering_still_uses_composite(self):
             assert "Tool Name: get_temperature" in rendered
             assert "Tool Arguments:" in rendered
             assert f"Tool Description: {self.AUTHORED}" in rendered
+
+def test_tool_requires_approval_approved():
+    """Test that the tool executes if the human approves."""
+    class DummyTestTool(BaseTool):
+        name: str = "dummy_tool"
+        description: str = "A dummy tool for testing"
+
+        def _run(self, question: str = "") -> str:
+            return "success_output"
+
+    tool = DummyTestTool(requires_human_approval=True) 
+    with patch.object(tool, '_request_human_approval', return_value=True):
+        with patch.object(tool, '_run', return_value="success_output") as mock_run:
+            result = tool.run(question="test")
+            assert result == "success_output"
+            mock_run.assert_called_once()
+
+
+def test_tool_requires_approval_rejected():
+    """Test that the tool hard-fails and DOES NOT execute if the human rejects."""
+    class DummyTestTool(BaseTool):
+        name: str = "dummy_tool"
+        description: str = "A dummy tool for testing"
+
+        def _run(self, question: str = "") -> str:
+            return "success_output"
+
+    tool = DummyTestTool(requires_human_approval=True)
+    with patch.object(tool, '_request_human_approval', return_value=False):
+        with patch.object(tool, '_run') as mock_run:
+            with pytest.raises(PermissionError, match="explicitly rejected by human"):
+                tool.run(question="test")
+            mock_run.assert_not_called()
+
+
+def test_tool_requires_approval_timeout():
+    """Test that timeouts fail closed securely."""
+    class DummyTestTool(BaseTool):
+        name: str = "dummy_tool"
+        description: str = "A dummy tool for testing"
+
+        def _run(self, question: str = "") -> str:
+            return "success_output"
+
+    tool = DummyTestTool(requires_human_approval=True)
+    with patch.object(tool, '_request_human_approval', side_effect=TimeoutError("Approval timed out")):
+        with patch.object(tool, '_run') as mock_run:
+            with pytest.raises(TimeoutError, match="Approval timed out"):
+                tool.run(question="test")
+            mock_run.assert_not_called()
+
+
+def test_tool_requires_approval_predicate():
+    """Test that requires_human_approval works as a callable predicate."""
+    class DummyTestTool(BaseTool):
+        name: str = "dummy_tool"
+        description: str = "A dummy tool for testing"
+
+        def _run(self, question: str = "") -> str:
+            return "success_output"
+
+    tool = DummyTestTool(requires_human_approval=lambda **kwargs: kwargs.get('question', '') == 'sensitive')
+
+    with patch.object(tool, '_request_human_approval', return_value=True) as mock_approve:
+        with patch.object(tool, '_run', return_value="ok") as mock_run:
+            # Should NOT request approval (question != 'sensitive')
+            tool.run(question="normal")
+            mock_approve.assert_not_called()
+            mock_run.assert_called_once()
+
+            mock_run.reset_mock()
+
+            # SHOULD request approval (question == 'sensitive')
+            tool.run(question="sensitive")
+            mock_approve.assert_called_once()
+            mock_run.assert_called_once()