Merge pull request aden-hive#293 from yumosx/graph

feat(file_system_toolkits): add encoding and max_size params to view_file

Author: bryanadenhq

tools/src/aden_tools/tools/file_system_toolkits/view_file/view_file.py

@@ -1,12 +1,22 @@
 import os
+
 from mcp.server.fastmcp import FastMCP
+
 from ..security import get_secure_path
 
+
 def register_tools(mcp: FastMCP) -> None:
     """Register file view tools with the MCP server."""
 
     @mcp.tool()
-    def view_file(path: str, workspace_id: str, agent_id: str, session_id: str) -> dict:
+    def view_file(
+        path: str,
+        workspace_id: str,
+        agent_id: str,
+        session_id: str,
+        encoding: str = "utf-8",
+        max_size: int = 10 * 1024 * 1024,
+    ) -> dict:
         """
         Purpose
             Read the content of a file within the session sandbox.
@@ -23,27 +33,39 @@ def view_file(path: str, workspace_id: str, agent_id: str, session_id: str) -> d
 
         Args:
             path: The path to the file (relative to session root)
-            workspace_id: The ID of the workspace
-            agent_id: The ID of the agent
+            workspace_id: The ID of workspace
+            agent_id: The ID of agent
             session_id: The ID of the current session
+            encoding: The encoding to use for reading the file (default: "utf-8")
+            max_size: The maximum size of file content to return in bytes (default: 10MB)
 
         Returns:
             Dict with file content and metadata, or error dict
         """
         try:
+            if max_size < 0:
+                return {"error": f"max_size must be non-negative, got {max_size}"}
+
             secure_path = get_secure_path(path, workspace_id, agent_id, session_id)
             if not os.path.exists(secure_path):
                 return {"error": f"File not found at {path}"}
 
-            with open(secure_path, "r", encoding="utf-8") as f:
+            if not os.path.isfile(secure_path):
+                return {"error": f"Path is not a file: {path}"}
+
+            with open(secure_path, "r", encoding=encoding) as f:
                 content = f.read()
 
+            if len(content.encode(encoding)) > max_size:
+                content = content[:max_size]
+                content += "\n\n[... Content truncated due to size limit ...]"
+
             return {
                 "success": True,
                 "path": path,
                 "content": content,
                 "size_bytes": len(content.encode("utf-8")),
-                "lines": len(content.splitlines())
+                "lines": len(content.splitlines()),
             }
         except Exception as e:
             return {"error": f"Failed to read file: {str(e)}"}

tools/tests/tools/test_file_system_toolkits.py

@@ -117,6 +117,49 @@ def test_view_nested_file(self, view_file_fn, mock_workspace, mock_secure_path,
         assert result["success"] is True
         assert result["content"] == "nested content"
 
+    def test_view_file_with_max_size_truncation(self, view_file_fn, mock_workspace, mock_secure_path, tmp_path):
+        """Viewing a file with max_size truncates content when exceeding limit."""
+        test_file = tmp_path / "large.txt"
+        content = "x" * 1000
+        test_file.write_text(content)
+
+        result = view_file_fn(path="large.txt", max_size=100, **mock_workspace)
+
+        assert result["success"] is True
+        assert len(result["content"]) <= 100 + len("\n\n[... Content truncated due to size limit ...]")
+        assert "[... Content truncated due to size limit ...]" in result["content"]
+
+    def test_view_file_with_negative_max_size(self, view_file_fn, mock_workspace, mock_secure_path, tmp_path):
+        """Viewing a file with negative max_size returns error."""
+        test_file = tmp_path / "test.txt"
+        test_file.write_text("content")
+
+        result = view_file_fn(path="test.txt", max_size=-1, **mock_workspace)
+
+        assert "error" in result
+        assert "max_size must be non-negative" in result["error"]
+
+    def test_view_file_with_custom_encoding(self, view_file_fn, mock_workspace, mock_secure_path, tmp_path):
+        """Viewing a file with custom encoding works correctly."""
+        test_file = tmp_path / "encoded.txt"
+        content = "Hello 世界"
+        test_file.write_text(content, encoding="utf-8")
+
+        result = view_file_fn(path="encoded.txt", encoding="utf-8", **mock_workspace)
+
+        assert result["success"] is True
+        assert result["content"] == content
+
+    def test_view_file_with_invalid_encoding(self, view_file_fn, mock_workspace, mock_secure_path, tmp_path):
+        """Viewing a file with invalid encoding returns error."""
+        test_file = tmp_path / "test.txt"
+        test_file.write_text("content")
+
+        result = view_file_fn(path="test.txt", encoding="invalid-encoding", **mock_workspace)
+
+        assert "error" in result
+        assert "Failed to read file" in result["error"]
+
 
 class TestWriteToFileTool:
     """Tests for write_to_file tool."""