Skip to content
Open
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
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi, why do we need a dependency change as part of this PR?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Which test would fail regarding Added botocore[crt] to dev dependencies to support agent interface tests

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The botocore[crt] dependency fixes a pre-existing test failure in the agent interface tests.

The failure:

  • test_file_write_via_agent
  • test_file_write_append_via_agent

Both error with:

botocore.exceptions.MissingDependencyException: Using the login credential 
provider requires an additional dependency. You will need to pip install 
"botocore[crt]" before proceeding.

This isn't related to append mode - it's been broken. I caught it while running the full test suite. With botocore[crt] added, all 16 tests pass.

If you'd prefer, I can split this into a separate PR and keep this one focused on just the append mode feature. The core functionality (14 tests) works fine without it.

"mem0ai>=0.1.104,<1.0.0",
"opensearch-py>=2.8.0,<3.0.0",
"nest-asyncio>=1.5.0,<2.0.0",
Expand Down
52 changes: 50 additions & 2 deletions src/strands_tools/file_write.py
Original file line number Diff line number Diff line change
Expand Up @@ -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!"
Expand All @@ -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.
Expand Down Expand Up @@ -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"],
}
Expand Down Expand Up @@ -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:
Expand All @@ -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"

Expand All @@ -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"),
),
Expand Down Expand Up @@ -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}"
Expand Down
159 changes: 159 additions & 0 deletions tests/test_file_write.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Loading