diff --git a/lib/crewai/src/crewai/tools/base_tool.py b/lib/crewai/src/crewai/tools/base_tool.py index 83986c9b84..cbe4ae2ab9 100644 --- a/lib/crewai/src/crewai/tools/base_tool.py +++ b/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 diff --git a/lib/crewai/src/crewai/tools/structured_tool.py b/lib/crewai/src/crewai/tools/structured_tool.py index c7191de40c..362ff72691 100644 --- a/lib/crewai/src/crewai/tools/structured_tool.py +++ b/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)) diff --git a/lib/crewai/tests/tools/test_base_tool.py b/lib/crewai/tests/tools/test_base_tool.py index b879fecf7f..4ff89ff28e 100644 --- a/lib/crewai/tests/tools/test_base_tool.py +++ b/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() \ No newline at end of file