diff --git a/pyproject.toml b/pyproject.toml index fb261ec5..4604a19c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,6 +71,7 @@ dev = [ "pytest>=8.0.0,<9.0.0", "ruff>=0.13.0,<0.14.0", "responses>=0.6.1,<1.0.0", + "botocore[crt]>=1.39.7,<2.0.0", "mem0ai>=0.1.104,<1.0.0", "opensearch-py>=2.8.0,<3.0.0", "nest-asyncio>=1.5.0,<2.0.0", diff --git a/src/strands_tools/file_write.py b/src/strands_tools/file_write.py index b142479e..b6abb966 100644 --- a/src/strands_tools/file_write.py +++ b/src/strands_tools/file_write.py @@ -38,7 +38,7 @@ agent = Agent(tools=[file_write]) -# Write to a file with user confirmation +# Write to a file with user confirmation (overwrites if exists) agent.tool.file_write( path="/path/to/file.txt", content="Hello World!" @@ -49,6 +49,20 @@ path="/path/to/script.py", content="def hello():\n print('Hello world!')" ) + +# Append to an existing file +agent.tool.file_write( + path="/path/to/log.txt", + content="New log entry\n", + mode="append" +) + +# Explicitly overwrite a file +agent.tool.file_write( + path="/path/to/config.json", + content='{"setting": "value"}', + mode="write" +) ``` See the file_write function docstring for more details on usage options and parameters. @@ -82,6 +96,15 @@ "type": "string", "description": "The content to write to the file", }, + "mode": { + "type": "string", + "enum": ["write", "append"], + "description": ( + "Write mode: 'write' to overwrite file (default), " + "'append' to add content to end of existing file" + ), + "default": "write", + }, }, "required": ["path", "content"], } @@ -168,6 +191,8 @@ def file_write(tool: ToolUse, **kwargs: Any) -> ToolResult: - path: The path to the file to write. User paths with tilde (~) are automatically expanded. - content: The content to write to the file. + - mode: (Optional) Write mode - 'write' to overwrite file (default), + 'append' to add content to end of existing file. **kwargs: Additional keyword arguments (not used currently) Returns: @@ -191,6 +216,24 @@ def file_write(tool: ToolUse, **kwargs: Any) -> ToolResult: tool_input = tool["input"] path = expanduser(tool_input["path"]) content = tool_input["content"] + mode = tool_input.get("mode", "write") + + # Validate mode + if mode not in ["write", "append"]: + error_message = f"Invalid mode: '{mode}'. Must be 'write' or 'append'" + error_panel = Panel( + Text(error_message, style="bold red"), + title="[bold red]Invalid Mode", + border_style="red", + box=box.HEAVY, + expand=False, + ) + console.print(error_panel) + return { + "toolUseId": tool_use_id, + "status": "error", + "content": [{"text": error_message}], + } strands_dev = os.environ.get("BYPASS_TOOL_CONSENT", "").lower() == "true" @@ -199,6 +242,8 @@ def file_write(tool: ToolUse, **kwargs: Any) -> ToolResult: Text.assemble( ("Path: ", "cyan"), (path, "yellow"), + ("\nMode: ", "cyan"), + (mode, "yellow"), ("\nSize: ", "cyan"), (f"{len(content)} characters", "yellow"), ), @@ -256,8 +301,11 @@ def file_write(tool: ToolUse, **kwargs: Any) -> ToolResult: ) ) + # Map mode to file open mode + file_mode = "w" if mode == "write" else "a" + # Write the file - with open(path, "w") as file: + with open(path, file_mode) as file: file.write(content) success_message = f"File written successfully to {path}" diff --git a/tests/test_file_write.py b/tests/test_file_write.py index 3b05a4c7..73ff42d0 100644 --- a/tests/test_file_write.py +++ b/tests/test_file_write.py @@ -242,3 +242,162 @@ def test_file_write_alternative_rejection(mock_user_input, temp_file): # Verify file was not created assert not os.path.exists(temp_file) + + +@patch("strands_tools.file_write.get_user_input") +def test_file_write_append_mode(mock_user_input, temp_file): + """Test appending content to existing file.""" + mock_user_input.return_value = "y" + + # Write initial content + with open(temp_file, "w") as f: + f.write("Initial content\n") + + # Append using the tool + tool_use = { + "toolUseId": "test-append", + "input": {"path": temp_file, "content": "Appended content\n", "mode": "append"}, + } + + result = file_write.file_write(tool=tool_use) + assert result["status"] == "success" + + # Verify both contents are present + with open(temp_file, "r") as f: + content = f.read() + assert "Initial content\n" in content + assert "Appended content\n" in content + assert content == "Initial content\nAppended content\n" + + +@patch("strands_tools.file_write.get_user_input") +def test_file_write_append_mode_new_file(mock_user_input, temp_file): + """Test that append mode creates file if it doesn't exist.""" + mock_user_input.return_value = "y" + + # Ensure file doesn't exist + assert not os.path.exists(temp_file) + + # Append to non-existent file + tool_use = { + "toolUseId": "test-append-new", + "input": {"path": temp_file, "content": "New file content\n", "mode": "append"}, + } + + result = file_write.file_write(tool=tool_use) + assert result["status"] == "success" + + # Verify file was created with content + assert os.path.exists(temp_file) + with open(temp_file, "r") as f: + assert f.read() == "New file content\n" + + +@patch("strands_tools.file_write.get_user_input") +def test_file_write_default_mode_overwrites(mock_user_input, temp_file): + """Test that default mode (write) overwrites existing content.""" + mock_user_input.return_value = "y" + + # Write initial content + with open(temp_file, "w") as f: + f.write("Original content\n") + + # Write without specifying mode (should default to overwrite) + tool_use = { + "toolUseId": "test-default-overwrite", + "input": {"path": temp_file, "content": "New content\n"}, + } + + result = file_write.file_write(tool=tool_use) + assert result["status"] == "success" + + # Verify original content was overwritten + with open(temp_file, "r") as f: + content = f.read() + assert content == "New content\n" + assert "Original content" not in content + + +@patch("strands_tools.file_write.get_user_input") +def test_file_write_write_mode_explicit(mock_user_input, temp_file): + """Test that explicit write mode overwrites existing content.""" + mock_user_input.return_value = "y" + + # Write initial content + with open(temp_file, "w") as f: + f.write("Original content\n") + + # Write with explicit write mode + tool_use = { + "toolUseId": "test-write-mode", + "input": {"path": temp_file, "content": "Replacement content\n", "mode": "write"}, + } + + result = file_write.file_write(tool=tool_use) + assert result["status"] == "success" + + # Verify original content was overwritten + with open(temp_file, "r") as f: + content = f.read() + assert content == "Replacement content\n" + assert "Original content" not in content + + +def test_file_write_invalid_mode(temp_file): + """Test error handling for invalid mode values.""" + tool_use = { + "toolUseId": "test-invalid-mode", + "input": {"path": temp_file, "content": "Test content", "mode": "invalid"}, + } + + result = file_write.file_write(tool=tool_use) + + # Verify the error was handled correctly + assert result["status"] == "error" + assert "Invalid mode" in result["content"][0]["text"] + assert "invalid" in result["content"][0]["text"] + + +@patch.dict("os.environ", {"BYPASS_TOOL_CONSENT": "true"}) +def test_file_write_append_via_agent(agent, temp_file): + """Test append mode via the agent interface.""" + # Write initial content + with open(temp_file, "w") as f: + f.write("First line\n") + + # Append via agent + result = agent.tool.file_write(path=temp_file, content="Second line\n", mode="append") + + # Verify success + result_text = extract_result_text(result) + assert "File write success" in result_text + + # Verify both lines are present + with open(temp_file, "r") as f: + content = f.read() + assert "First line\n" in content + assert "Second line\n" in content + + +@patch("strands_tools.file_write.get_user_input") +def test_file_write_append_multiple_times(mock_user_input, temp_file): + """Test appending multiple times to the same file.""" + mock_user_input.return_value = "y" + + # Write initial content + with open(temp_file, "w") as f: + f.write("Line 1\n") + + # Append multiple times + for i in range(2, 5): + tool_use = { + "toolUseId": f"test-append-{i}", + "input": {"path": temp_file, "content": f"Line {i}\n", "mode": "append"}, + } + result = file_write.file_write(tool=tool_use) + assert result["status"] == "success" + + # Verify all lines are present in order + with open(temp_file, "r") as f: + content = f.read() + assert content == "Line 1\nLine 2\nLine 3\nLine 4\n"