diff --git a/src/strands_tools/python_repl.py b/src/strands_tools/python_repl.py index da91ca4e..cea669f7 100644 --- a/src/strands_tools/python_repl.py +++ b/src/strands_tools/python_repl.py @@ -45,6 +45,7 @@ import threading import traceback import types +import warnings from datetime import datetime from io import StringIO from pathlib import Path @@ -73,7 +74,8 @@ "3. State Management: Maintains variables between executions, default controlled by PYTHON_REPL_RESET_STATE\n" "4. Error Handling: Captures and formats errors with suggestions\n" "5. Development Mode: Can bypass confirmation in BYPASS_TOOL_CONSENT environments\n" - "6. Interactive Control: Can enable/disable interactive PTY mode in PYTHON_REPL_INTERACTIVE environments\n\n" + "6. Interactive Control: Can enable/disable interactive PTY mode in PYTHON_REPL_INTERACTIVE environments\n" + "7. Non-Interactive Mode: Set STRANDS_NON_INTERACTIVE=true to suppress confirmation prompts\n\n" "Key Features:\n" "- Persistent state between executions\n" "- Interactive PTY support for real-time feedback\n" @@ -577,8 +579,21 @@ def python_repl(tool: ToolUse, **kwargs: Any) -> ToolResult: # Check for development mode strands_dev = os.environ.get("BYPASS_TOOL_CONSENT", "").lower() == "true" - # Check for non_interactive_mode parameter - non_interactive_mode = kwargs.get("non_interactive_mode", False) + # Non-interactive mode is driven by the environment, mirroring shell.py, so + # that suppressing the confirmation prompt is an operator decision rather + # than something a caller can request per invocation. + non_interactive_mode = os.environ.get("STRANDS_NON_INTERACTIVE", "").lower() == "true" + + # The non_interactive_mode keyword argument is no longer honored; suppressing + # the prompt is driven solely by STRANDS_NON_INTERACTIVE. Warn callers that + # still pass it so the change is visible rather than silently ignored. + if "non_interactive_mode" in kwargs: + warnings.warn( + "The 'non_interactive_mode' argument to python_repl is no longer honored. " + "Set the STRANDS_NON_INTERACTIVE=true environment variable instead.", + DeprecationWarning, + stacklevel=2, + ) try: # Handle state reset if requested diff --git a/tests/test_python_repl.py b/tests/test_python_repl.py index 63b1b392..ac6db70a 100644 --- a/tests/test_python_repl.py +++ b/tests/test_python_repl.py @@ -348,18 +348,40 @@ def test_dev_mode_bypass_confirmation(self, mock_console): assert python_repl.repl_state.get_namespace()["dev_mode_test"] == 42 def test_non_interactive_mode_bypass_confirmation(self, mock_console): - """Test that non_interactive_mode bypasses the confirmation dialog.""" + """Test that STRANDS_NON_INTERACTIVE bypasses the confirmation dialog.""" tool_use = { "toolUseId": "test-id", "input": {"code": "non_interactive_test = 'passed'", "interactive": False}, } - # Pass non_interactive_mode as True - result = python_repl.python_repl(tool=tool_use, non_interactive_mode=True) + # STRANDS_NON_INTERACTIVE drives non-interactive mode + with patch.dict(os.environ, {"STRANDS_NON_INTERACTIVE": "true"}): + result = python_repl.python_repl(tool=tool_use) assert result["status"] == "success" assert python_repl.repl_state.get_namespace()["non_interactive_test"] == "passed" + def test_non_interactive_kwarg_does_not_bypass_confirmation(self, mock_console): + """A non_interactive_mode kwarg must not suppress the confirmation prompt.""" + tool_use = { + "toolUseId": "test-id", + "input": {"code": "kwarg_bypass = True", "interactive": False}, + } + + # Without the env var set, the confirmation prompt must run even if a + # caller passes non_interactive_mode=True. Simulate the user declining. + # Passing the now-ignored kwarg must also emit a DeprecationWarning. + with ( + patch("strands_tools.python_repl.get_user_input", side_effect=["n", "declining"]), + patch.dict(os.environ, {"BYPASS_TOOL_CONSENT": "false", "STRANDS_NON_INTERACTIVE": ""}, clear=False), + pytest.warns(DeprecationWarning, match="non_interactive_mode"), + ): + result = python_repl.python_repl(tool=tool_use, non_interactive_mode=True) + + assert result["status"] == "error" + assert "cancelled by the user" in result["content"][0]["text"] + assert "kwarg_bypass" not in python_repl.repl_state.get_namespace() + def test_user_rejection_cancels_execution(self, mock_console): """Test that user rejection properly cancels execution.""" # Clear REPL state to ensure clean test environment @@ -411,9 +433,9 @@ def test_recursion_error(self, mock_console, temp_repl_state_dir): }, } - # First define the recursive function - # Pass non_interactive_mode=True to bypass confirmation - python_repl.python_repl(tool=tool_use, non_interactive_mode=True) + # First define the recursive function (non-interactive via env var) + with patch.dict(os.environ, {"STRANDS_NON_INTERACTIVE": "true"}): + python_repl.python_repl(tool=tool_use) # Now trigger the recursion error error_tool = { @@ -425,13 +447,15 @@ def test_recursion_error(self, mock_console, temp_repl_state_dir): } # Mock the clear_state method to verify it gets called - with patch.object( - python_repl.repl_state, - "clear_state", - wraps=python_repl.repl_state.clear_state, - ) as mock_clear: - # Pass non_interactive_mode=True to bypass confirmation - result = python_repl.python_repl(tool=error_tool, non_interactive_mode=True) + with ( + patch.object( + python_repl.repl_state, + "clear_state", + wraps=python_repl.repl_state.clear_state, + ) as mock_clear, + patch.dict(os.environ, {"STRANDS_NON_INTERACTIVE": "true"}), + ): + result = python_repl.python_repl(tool=error_tool) # Verify clear_state was called mock_clear.assert_called_once() @@ -459,11 +483,13 @@ def test_interactive_mode(self, mock_console): mock_pty_instance.get_output.return_value = "Interactive test\n" # Mock os.waitpid to simulate process completion - with patch("os.waitpid") as mock_waitpid: + with ( + patch("os.waitpid") as mock_waitpid, + patch.dict(os.environ, {"STRANDS_NON_INTERACTIVE": "true"}), + ): mock_waitpid.side_effect = [(12345, 0)] # Return pid and exit status 0 - # Pass non_interactive_mode=True to bypass confirmation - result = python_repl.python_repl(tool=tool_use, non_interactive_mode=True) + result = python_repl.python_repl(tool=tool_use) # Verify PtyManager was used mock_pty.assert_called_once() @@ -493,11 +519,13 @@ def test_interactive_mode_error(self, mock_console): mock_pty_instance.get_output.return_value = "Traceback... ValueError: Test error" # Mock os.waitpid to simulate process error - with patch("os.waitpid") as mock_waitpid: + with ( + patch("os.waitpid") as mock_waitpid, + patch.dict(os.environ, {"STRANDS_NON_INTERACTIVE": "true"}), + ): mock_waitpid.side_effect = [(12345, 1)] # Return pid and non-zero exit status - # Pass non_interactive_mode=True to bypass confirmation - result = python_repl.python_repl(tool=tool_use, non_interactive_mode=True) + result = python_repl.python_repl(tool=tool_use) # Verify PtyManager was used and stopped mock_pty_instance.stop.assert_called_once() @@ -523,9 +551,11 @@ def test_interactive_mode_os_error(self, mock_console): mock_pty_instance.get_output.return_value = "test output" # Mock os.waitpid to raise OSError - with patch("os.waitpid", side_effect=OSError("No such process")): - # Pass non_interactive_mode=True to bypass confirmation - result = python_repl.python_repl(tool=tool_use, non_interactive_mode=True) + with ( + patch("os.waitpid", side_effect=OSError("No such process")), + patch.dict(os.environ, {"STRANDS_NON_INTERACTIVE": "true"}), + ): + result = python_repl.python_repl(tool=tool_use) # Verify PtyManager was stopped and cleaned up mock_pty_instance.stop.assert_called_once() @@ -545,8 +575,9 @@ def test_interactive_mode_os_error(self, mock_console): ) def test_agent_interface(agent, code, expected): """Test calling python_repl through the Agent interface.""" - # Use non_interactive_mode to bypass confirmation - result = agent.tool.python_repl(code=code, interactive=False, non_interactive_mode=True) + # Use STRANDS_NON_INTERACTIVE to bypass confirmation + with patch.dict(os.environ, {"STRANDS_NON_INTERACTIVE": "true"}): + result = agent.tool.python_repl(code=code, interactive=False) # Extract the response text if isinstance(result, dict) and "content" in result and isinstance(result["content"], list):