tiphys.tools.fs

  1"""
  2File system tools for agents.
  3
  4Provides safe file reading, writing, and directory listing.
  5"""
  6
  7from pathlib import Path
  8
  9import aiofiles
 10import aiofiles.os
 11import structlog
 12
 13from tiphys.tools.base import ToolDefinition, tool
 14
 15logger = structlog.get_logger(__name__)
 16
 17# Maximum file size to read (10MB)
 18MAX_READ_SIZE = 10 * 1024 * 1024
 19
 20# Maximum lines to return
 21MAX_LINES = 1000
 22
 23
 24def _validate_path(path: str, must_exist: bool = False) -> Path:
 25    """
 26    Validate and normalize a file path.
 27
 28    Args:
 29        path: The path to validate.
 30        must_exist: If True, raise error if path doesn't exist.
 31
 32    Returns:
 33        Resolved Path object.
 34
 35    Raises:
 36        ValueError: If path is invalid.
 37        FileNotFoundError: If must_exist=True and path doesn't exist.
 38    """
 39    try:
 40        p = Path(path).resolve()
 41    except Exception as e:
 42        raise ValueError(f"Invalid path: {path}") from e
 43
 44    # Security: prevent path traversal attacks
 45    # Ensure the path doesn't escape to dangerous locations
 46    dangerous_prefixes = ["/etc/passwd", "/etc/shadow", "/proc", "/sys"]
 47    for prefix in dangerous_prefixes:
 48        if str(p).startswith(prefix):
 49            raise ValueError(f"Access denied: {path}")
 50
 51    if must_exist and not p.exists():
 52        raise FileNotFoundError(f"Path not found: {path}")
 53
 54    return p
 55
 56
 57@tool(
 58    name="read_file",
 59    description="Read the contents of a file. Returns the file contents as text.",
 60)
 61async def read_file(
 62    path: str,
 63    start_line: int = 1,
 64    end_line: int | None = None,
 65) -> str:
 66    """
 67    Read a file and return its contents.
 68
 69    Args:
 70        path: Absolute or relative path to the file.
 71        start_line: Starting line number (1-indexed).
 72        end_line: Ending line number (inclusive). If None, reads to end.
 73
 74    Returns:
 75        File contents as string.
 76    """
 77    try:
 78        file_path = _validate_path(path, must_exist=True)
 79    except FileNotFoundError:
 80        return f"Error: File not found: {path}"
 81    except ValueError as e:
 82        return f"Error: {e}"
 83
 84    if not file_path.is_file():
 85        return f"Error: {path} is not a file"
 86
 87    # Check file size
 88    try:
 89        stat = await aiofiles.os.stat(file_path)
 90        if stat.st_size > MAX_READ_SIZE:
 91            return (
 92                f"Error: File too large ({stat.st_size} bytes). Maximum is {MAX_READ_SIZE} bytes."
 93            )
 94    except OSError as e:
 95        return f"Error checking file: {e}"
 96
 97    try:
 98        async with aiofiles.open(file_path, encoding="utf-8", errors="replace") as f:
 99            if start_line == 1 and end_line is None:
100                # Read entire file
101                content = await f.read()
102                lines = content.splitlines()
103                if len(lines) > MAX_LINES:
104                    return (
105                        f"File has {len(lines)} lines. Showing first {MAX_LINES}:\n\n"
106                        + "\n".join(lines[:MAX_LINES])
107                    )
108                return content
109            else:
110                # Read specific line range
111                lines = []
112                line_num = 0
113                async for line in f:
114                    line_num += 1
115                    if line_num >= start_line:
116                        lines.append(line.rstrip("\n"))
117                    if end_line and line_num >= end_line:
118                        break
119                    if len(lines) >= MAX_LINES:
120                        break
121
122                return "\n".join(lines)
123
124    except UnicodeDecodeError:
125        return f"Error: {path} is not a text file"
126    except PermissionError:
127        return f"Error: Permission denied reading {path}"
128    except Exception as e:
129        await logger.aerror("Failed to read file", path=path, error=str(e))
130        return f"Error reading file: {e}"
131
132
133@tool(
134    name="write_file",
135    description="Write content to a file. Creates the file if it doesn't exist, overwrites if it does.",
136    requires_permission=True,
137)
138async def write_file(
139    path: str,
140    content: str,
141    append: bool = False,
142) -> str:
143    """
144    Write content to a file.
145
146    Args:
147        path: Path to the file.
148        content: Content to write.
149        append: If True, append to file instead of overwriting.
150
151    Returns:
152        Success message or error.
153    """
154    try:
155        file_path = _validate_path(path)
156    except ValueError as e:
157        return f"Error: {e}"
158
159    # Create parent directories if needed
160    try:
161        file_path.parent.mkdir(parents=True, exist_ok=True)
162    except PermissionError:
163        return f"Error: Permission denied creating directory {file_path.parent}"
164
165    mode = "a" if append else "w"
166
167    try:
168        async with aiofiles.open(file_path, mode, encoding="utf-8") as f:  # type: ignore[call-overload]
169            await f.write(content)
170
171        action = "appended to" if append else "written to"
172        return f"Successfully {action} {path} ({len(content)} characters)"
173
174    except PermissionError:
175        return f"Error: Permission denied writing to {path}"
176    except Exception as e:
177        await logger.aerror("Failed to write file", path=path, error=str(e))
178        return f"Error writing file: {e}"
179
180
181@tool(
182    name="list_directory",
183    description="List the contents of a directory. Returns files and subdirectories.",
184)
185async def list_directory(
186    path: str = ".",
187    show_hidden: bool = False,
188    recursive: bool = False,
189    max_depth: int = 2,
190) -> str:
191    """
192    List directory contents.
193
194    Args:
195        path: Directory path. Defaults to current directory.
196        show_hidden: Include hidden files (starting with .).
197        recursive: List subdirectories recursively.
198        max_depth: Maximum recursion depth (only if recursive=True).
199
200    Returns:
201        Formatted directory listing.
202    """
203    try:
204        dir_path = _validate_path(path, must_exist=True)
205    except (ValueError, FileNotFoundError) as e:
206        return f"Error: {e}"
207
208    if not dir_path.is_dir():
209        return f"Error: {path} is not a directory"
210
211    try:
212        lines: list[str] = [f"Contents of {dir_path}:", ""]
213
214        if recursive:
215            lines.extend(_list_recursive(dir_path, show_hidden, max_depth, 0))
216        else:
217            lines.extend(_list_flat(dir_path, show_hidden))
218
219        return "\n".join(lines)
220
221    except PermissionError:
222        return f"Error: Permission denied accessing {path}"
223    except Exception as e:
224        await logger.aerror("Failed to list directory", path=path, error=str(e))
225        return f"Error listing directory: {e}"
226
227
228def _list_flat(dir_path: Path, show_hidden: bool) -> list[str]:
229    """List directory contents without recursion."""
230    items: list[str] = []
231
232    try:
233        entries = sorted(dir_path.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower()))
234
235        for entry in entries:
236            if not show_hidden and entry.name.startswith("."):
237                continue
238
239            if entry.is_dir():
240                items.append(f"📁 {entry.name}/")
241            elif entry.is_symlink():
242                items.append(f"🔗 {entry.name} -> {entry.resolve()}")
243            else:
244                size = _format_size(entry.stat().st_size)
245                items.append(f"📄 {entry.name} ({size})")
246
247    except PermissionError:
248        items.append("  (permission denied)")
249
250    return items
251
252
253def _list_recursive(
254    dir_path: Path, show_hidden: bool, max_depth: int, current_depth: int
255) -> list[str]:
256    """List directory contents recursively."""
257    items: list[str] = []
258    indent = "  " * current_depth
259
260    try:
261        entries = sorted(dir_path.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower()))
262
263        for entry in entries:
264            if not show_hidden and entry.name.startswith("."):
265                continue
266
267            if entry.is_dir():
268                items.append(f"{indent}📁 {entry.name}/")
269                if current_depth < max_depth:
270                    items.extend(_list_recursive(entry, show_hidden, max_depth, current_depth + 1))
271            else:
272                size = _format_size(entry.stat().st_size)
273                items.append(f"{indent}📄 {entry.name} ({size})")
274
275    except PermissionError:
276        items.append(f"{indent}  (permission denied)")
277
278    return items
279
280
281def _format_size(size_bytes: int) -> str:
282    """Format file size in human-readable format."""
283    size: float = float(size_bytes)
284    for unit in ["B", "KB", "MB", "GB"]:
285        if size < 1024:
286            return f"{size:.1f}{unit}" if unit != "B" else f"{int(size)}{unit}"
287        size /= 1024
288    return f"{size:.1f}TB"
289
290
291# Export tool definitions
292def get_fs_tools() -> list[ToolDefinition]:
293    """Get all file system tool definitions."""
294    tools: list[ToolDefinition] = []
295
296    for func in [read_file, write_file, list_directory]:
297        if hasattr(func, "_tool_definition"):
298            tools.append(func._tool_definition)
299
300    return tools