Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions src/strands_tools/python_repl.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
import threading
import traceback
import types
import warnings
from datetime import datetime
from io import StringIO
from pathlib import Path
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand Down
79 changes: 55 additions & 24 deletions tests/test_python_repl.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 = {
Expand All @@ -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()
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()
Expand All @@ -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()
Expand All @@ -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):
Expand Down
Loading