Spaces:
Paused
Paused
File size: 5,670 Bytes
0d3f7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | """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}
|