Spaces:
Paused
Paused
| """Filesystem tool implementation.""" | |
| from __future__ import annotations | |
| import logging | |
| import os | |
| from pathlib import Path | |
| from typing import Any | |
| from hermes.tools.base.tool import BaseTool, ToolSchema | |
| logger = logging.getLogger(__name__) | |
| class FilesystemTool(BaseTool): | |
| """Tool for filesystem operations.""" | |
| def __init__(self, base_path: str | None = None) -> None: | |
| super().__init__() | |
| self.base_path = Path(base_path) if base_path else Path.cwd() | |
| def _define_schema(self) -> ToolSchema: | |
| return ToolSchema( | |
| name="file_reader", | |
| description="Read, write, and list files on the filesystem", | |
| parameters={ | |
| "action": { | |
| "type": "string", | |
| "description": "Action to perform: read, write, list, exists, mkdir", | |
| }, | |
| "path": { | |
| "type": "string", | |
| "description": "File or directory path", | |
| }, | |
| "content": { | |
| "type": "string", | |
| "description": "Content to write (for write action)", | |
| }, | |
| "pattern": { | |
| "type": "string", | |
| "description": "Glob pattern for listing files", | |
| "default": "*", | |
| }, | |
| }, | |
| required=["action", "path"], | |
| category="filesystem", | |
| tags=["file", "read", "write", "filesystem"], | |
| ) | |
| def _resolve_path(self, path: str) -> Path: | |
| """Resolve and validate path with symlink protection.""" | |
| base = self.base_path.resolve(strict=False) | |
| candidate = (self.base_path / path).resolve(strict=False) | |
| try: | |
| candidate.relative_to(base) | |
| except ValueError: | |
| raise ValueError(f"Path escapes base directory: {path}") from None | |
| if candidate.is_symlink(): | |
| raise ValueError(f"Symlinks not allowed: {path}") | |
| return candidate | |
| async def execute(self, **kwargs: Any) -> dict[str, Any]: | |
| """Execute filesystem operation.""" | |
| action = kwargs["action"] | |
| path = kwargs["path"] | |
| try: | |
| if action == "read": | |
| return await self._read_file(path) | |
| elif action == "write": | |
| return await self._write_file(path, kwargs.get("content", "")) | |
| elif action == "list": | |
| return await self._list_directory(path, kwargs.get("pattern", "*")) | |
| elif action == "exists": | |
| return await self._check_exists(path) | |
| elif action == "mkdir": | |
| return await self._make_directory(path) | |
| else: | |
| return {"error": f"Unknown action: {action}"} | |
| except Exception as e: | |
| logger.error(f"Filesystem error: {e}") | |
| return {"error": str(e)} | |
| async def _read_file(self, path: str) -> dict[str, Any]: | |
| """Read a file.""" | |
| resolved = self._resolve_path(path) | |
| if not resolved.exists(): | |
| return {"error": f"File not found: {path}"} | |
| if not resolved.is_file(): | |
| return {"error": f"Not a file: {path}"} | |
| content = resolved.read_text(encoding="utf-8", errors="replace") | |
| return { | |
| "path": str(resolved), | |
| "content": content, | |
| "size": resolved.stat().st_size, | |
| "modified": os.path.getmtime(resolved), | |
| } | |
| async def _write_file(self, path: str, content: str) -> dict[str, Any]: | |
| """Write to a file.""" | |
| resolved = self._resolve_path(path) | |
| resolved.parent.mkdir(parents=True, exist_ok=True) | |
| resolved.write_text(content, encoding="utf-8") | |
| return { | |
| "path": str(resolved), | |
| "size": len(content), | |
| "success": True, | |
| } | |
| async def _list_directory(self, path: str, pattern: str) -> dict[str, Any]: | |
| """List directory contents.""" | |
| resolved = self._resolve_path(path) | |
| if not resolved.exists(): | |
| return {"error": f"Directory not found: {path}"} | |
| if not resolved.is_dir(): | |
| return {"error": f"Not a directory: {path}"} | |
| base_resolved = self.base_path.resolve() | |
| items = [] | |
| for item in sorted(resolved.glob(pattern)): | |
| try: | |
| rel_path = str(item.relative_to(base_resolved)) | |
| except ValueError: | |
| rel_path = item.name | |
| items.append( | |
| { | |
| "name": item.name, | |
| "path": rel_path, | |
| "type": "directory" if item.is_dir() else "file", | |
| "size": item.stat().st_size if item.is_file() else 0, | |
| } | |
| ) | |
| return {"path": str(resolved), "items": items, "count": len(items)} | |
| async def _check_exists(self, path: str) -> dict[str, Any]: | |
| """Check if path exists.""" | |
| resolved = self._resolve_path(path) | |
| return { | |
| "path": str(resolved), | |
| "exists": resolved.exists(), | |
| "is_file": resolved.is_file() if resolved.exists() else False, | |
| "is_dir": resolved.is_dir() if resolved.exists() else False, | |
| } | |
| async def _make_directory(self, path: str) -> dict[str, Any]: | |
| """Create a directory.""" | |
| resolved = self._resolve_path(path) | |
| resolved.mkdir(parents=True, exist_ok=True) | |
| return {"path": str(resolved), "created": True} | |