150 lines
4.7 KiB
Python
150 lines
4.7 KiB
Python
"""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),
|
|
)
|