suna/tools/files_tool.py

import os
import asyncio
from typing import Dict, Any
from agentpress.tool import Tool, ToolResult, tool_schema
from agentpress.config import settings

class FilesTool(Tool):
    def __init__(self):
        super().__init__()
        self.workspace = settings.workspace_dir
        os.makedirs(self.workspace, exist_ok=True)

    @tool_schema({
        "name": "create_file",
        "description": "Create a new file in the workspace",
        "parameters": {
            "type": "object",
            "properties": {
                "file_path": {"type": "string", "description": "The relative path of the file to create"},
                "content": {"type": "string", "description": "The content to write to the file"}
            },
            "required": ["file_path", "content"]
        }
    })
    async def create_file(self, file_path: str, content: str) -> ToolResult:
        try:
            full_path = os.path.join(self.workspace, file_path)
            if os.path.exists(full_path):
                return self.fail_response(f"File '{file_path}' already exists. Use update_file to modify existing files.")
            os.makedirs(os.path.dirname(full_path), exist_ok=True)
            with open(full_path, 'w') as f:
                f.write(content)
            return self.success_response(f"File '{file_path}' created successfully.")
        except Exception as e:
            return self.fail_response(f"Error creating file: {str(e)}")

    @tool_schema({
        "name": "read_file",
        "description": "Read the contents of a file in the workspace",
        "parameters": {
            "type": "object",
            "properties": {
                "file_path": {"type": "string", "description": "The relative path of the file to read"}
            },
            "required": ["file_path"]
        }
    })
    async def read_file(self, file_path: str) -> ToolResult:
        try:
            full_path = os.path.join(self.workspace, file_path)
            with open(full_path, 'r') as f:
                content = f.read()
            return self.success_response({"file_path": file_path, "content": content})
        except Exception as e:
            return self.fail_response(f"Error reading file: {str(e)}")

    @tool_schema({
        "name": "update_file",
        "description": "Update the contents of a file in the workspace",
        "parameters": {
            "type": "object",
            "properties": {
                "file_path": {"type": "string", "description": "The relative path of the file to update"},
                "content": {"type": "string", "description": "The new content to write to the file"}
            },
            "required": ["file_path", "content"]
        }
    })
    async def update_file(self, file_path: str, content: str) -> ToolResult:
        try:
            full_path = os.path.join(self.workspace, file_path)
            with open(full_path, 'w') as f:
                f.write(content)
            return self.success_response(f"File '{file_path}' updated successfully.")
        except Exception as e:
            return self.fail_response(f"Error updating file: {str(e)}")


    @tool_schema({
        "name": "delete_file",
        "description": "Delete a file from the workspace",
        "parameters": {
            "type": "object",
            "properties": {
                "file_path": {"type": "string", "description": "The relative path of the file to delete"}
            },
            "required": ["file_path"]
        }
    })
    async def delete_file(self, file_path: str) -> ToolResult:
        try:
            full_path = os.path.join(self.workspace, file_path)
            os.remove(full_path)
            return self.success_response(f"File '{file_path}' deleted successfully.")
        except Exception as e:
            return self.fail_response(f"Error deleting file: {str(e)}")


if __name__ == "__main__":
    async def test_files_tool():
        files_tool = FilesTool()
        test_file_path = "test_file.txt"
        test_content = "This is a test file."
        updated_content = "This is an updated test file."

        print(f"Using workspace directory: {files_tool.workspace}")

        # Test create_file
        create_result = await files_tool.create_file(test_file_path, test_content)
        print("Create file result:", create_result)

        # Test read_file
        read_result = await files_tool.read_file(test_file_path)
        print("Read file result:", read_result)

        # Test update_file
        update_result = await files_tool.update_file(test_file_path, updated_content)
        print("Update file result:", update_result)

        # Test read_file after update
        read_updated_result = await files_tool.read_file(test_file_path)
        print("Read updated file result:", read_updated_result)

        # Test delete_file
        delete_result = await files_tool.delete_file(test_file_path)
        print("Delete file result:", delete_result)

        # Test read_file after delete (should fail)
        read_deleted_result = await files_tool.read_file(test_file_path)
        print("Read deleted file result:", read_deleted_result)

    asyncio.run(test_files_tool())