added file permissions and fixed a duplicate output bug.
This commit is contained in:
149
app/tools/filesystem.py
Normal file
149
app/tools/filesystem.py
Normal file
@@ -0,0 +1,149 @@
|
||||
"""Filesystem tools: read_file and list_dir."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.models.tool_call import ToolResult, ToolResultStatus
|
||||
from app.tools.base import BaseTool
|
||||
from app.utils.file_helpers import (
|
||||
BinaryFileError,
|
||||
FileSizeError,
|
||||
PathSecurityError,
|
||||
resolve_safe_path,
|
||||
safe_read_file,
|
||||
)
|
||||
|
||||
|
||||
class ReadFileParams(BaseModel):
|
||||
"""Parameters for the read_file tool."""
|
||||
|
||||
file_path: str = Field(description="Path to the file to read (relative to workspace root)")
|
||||
|
||||
|
||||
class ReadFileTool(BaseTool):
|
||||
"""Read the contents of a file within the workspace."""
|
||||
|
||||
name = "read_file"
|
||||
description = "Read the full contents of a text file. Returns the file content as a string."
|
||||
params_model = ReadFileParams
|
||||
|
||||
def execute(self, *, tool_call_id: str, file_path: str, **kwargs: Any) -> ToolResult:
|
||||
fs_config = self.config.tools.filesystem
|
||||
try:
|
||||
content = safe_read_file(
|
||||
file_path,
|
||||
self.workspace_root,
|
||||
max_size_bytes=fs_config.max_file_size_bytes,
|
||||
check_binary=fs_config.binary_detection,
|
||||
)
|
||||
except PathSecurityError as exc:
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error=str(exc),
|
||||
)
|
||||
except FileNotFoundError as exc:
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error=str(exc),
|
||||
)
|
||||
except FileSizeError as exc:
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error=str(exc),
|
||||
)
|
||||
except BinaryFileError as exc:
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error=str(exc),
|
||||
)
|
||||
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.SUCCESS,
|
||||
output=content,
|
||||
)
|
||||
|
||||
|
||||
class ListDirParams(BaseModel):
|
||||
"""Parameters for the list_dir tool."""
|
||||
|
||||
directory_path: str = Field(
|
||||
default=".", description="Path to the directory to list (relative to workspace root)"
|
||||
)
|
||||
recursive: bool = Field(default=False, description="If true, list entries recursively")
|
||||
|
||||
|
||||
_MAX_ENTRIES = 500
|
||||
|
||||
|
||||
class ListDirTool(BaseTool):
|
||||
"""List files and directories within the workspace."""
|
||||
|
||||
name = "list_dir"
|
||||
description = (
|
||||
"List the contents of a directory. Directories are suffixed with '/'. "
|
||||
"Results are sorted with directories first, then files."
|
||||
)
|
||||
params_model = ListDirParams
|
||||
|
||||
def execute(
|
||||
self, *, tool_call_id: str, directory_path: str = ".", recursive: bool = False, **kwargs: Any
|
||||
) -> ToolResult:
|
||||
try:
|
||||
safe_path = resolve_safe_path(directory_path, self.workspace_root)
|
||||
except PathSecurityError as exc:
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error=str(exc),
|
||||
)
|
||||
|
||||
if not safe_path.is_dir():
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error=f"Not a directory: {safe_path}",
|
||||
)
|
||||
|
||||
entries: list[Path] = []
|
||||
if recursive:
|
||||
entries = list(safe_path.rglob("*"))
|
||||
else:
|
||||
entries = list(safe_path.iterdir())
|
||||
|
||||
# Sort: directories first, then files, alphabetical within each group
|
||||
dirs = sorted([e for e in entries if e.is_dir()])
|
||||
files = sorted([e for e in entries if e.is_file()])
|
||||
sorted_entries = dirs + files
|
||||
|
||||
lines: list[str] = []
|
||||
for entry in sorted_entries[:_MAX_ENTRIES]:
|
||||
try:
|
||||
rel = entry.relative_to(self.workspace_root)
|
||||
except ValueError:
|
||||
rel = entry
|
||||
suffix = "/" if entry.is_dir() else ""
|
||||
lines.append(f"{rel}{suffix}")
|
||||
|
||||
if len(sorted_entries) > _MAX_ENTRIES:
|
||||
lines.append(f"\n... truncated ({len(sorted_entries)} total entries, showing {_MAX_ENTRIES})")
|
||||
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.SUCCESS,
|
||||
output="\n".join(lines),
|
||||
)
|
||||
Reference in New Issue
Block a user