106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.tools.base import Tool
|
|
|
|
|
|
class FilesTool(Tool):
|
|
name = "files"
|
|
description = "Read, list, and write files within the workspace."
|
|
|
|
def __init__(self, workspace_root: Path) -> None:
|
|
self.workspace_root = workspace_root.resolve()
|
|
|
|
def parameters_schema(self) -> dict[str, Any]:
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"action": {
|
|
"type": "string",
|
|
"enum": ["read", "list", "write"],
|
|
"description": "Use read to read a file, list to list a directory, or write to create/update a file.",
|
|
},
|
|
"path": {
|
|
"type": "string",
|
|
"description": "Absolute or relative path inside the workspace.",
|
|
},
|
|
"content": {
|
|
"type": "string",
|
|
"description": "File content for write operations.",
|
|
},
|
|
},
|
|
"required": ["action", "path"],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
action = str(payload.get("action", "read")).strip()
|
|
raw_path = str(payload.get("path", "")).strip()
|
|
path = self._resolve_path(raw_path)
|
|
|
|
if action == "read":
|
|
if not path.exists():
|
|
return {"tool": self.name, "status": "error", "message": f"Path not found: {path}"}
|
|
if path.is_dir():
|
|
return {"tool": self.name, "status": "error", "message": f"Path is a directory: {path}"}
|
|
content = path.read_text(encoding="utf-8", errors="replace")
|
|
return {
|
|
"tool": self.name,
|
|
"status": "ok",
|
|
"action": action,
|
|
"path": str(path),
|
|
"content": content[:12000],
|
|
"truncated": len(content) > 12000,
|
|
}
|
|
|
|
if action == "list":
|
|
if not path.exists():
|
|
return {"tool": self.name, "status": "error", "message": f"Path not found: {path}"}
|
|
if not path.is_dir():
|
|
return {"tool": self.name, "status": "error", "message": f"Path is not a directory: {path}"}
|
|
entries = []
|
|
for child in sorted(path.iterdir(), key=lambda item: item.name.lower())[:200]:
|
|
entries.append(
|
|
{
|
|
"name": child.name,
|
|
"type": "dir" if child.is_dir() else "file",
|
|
}
|
|
)
|
|
return {
|
|
"tool": self.name,
|
|
"status": "ok",
|
|
"action": action,
|
|
"path": str(path),
|
|
"entries": entries,
|
|
"truncated": len(entries) >= 200,
|
|
}
|
|
|
|
if action == "write":
|
|
content = str(payload.get("content", ""))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
return {
|
|
"tool": self.name,
|
|
"status": "ok",
|
|
"action": action,
|
|
"path": str(path),
|
|
"bytes_written": len(content.encode("utf-8")),
|
|
}
|
|
|
|
return {
|
|
"tool": self.name,
|
|
"status": "error",
|
|
"message": f"Unsupported action: {action}. Allowed actions are read, list, and write.",
|
|
}
|
|
|
|
def _resolve_path(self, raw_path: str) -> Path:
|
|
candidate = Path(raw_path).expanduser()
|
|
if not candidate.is_absolute():
|
|
candidate = (self.workspace_root / candidate).resolve()
|
|
else:
|
|
candidate = candidate.resolve()
|
|
|
|
if self.workspace_root not in candidate.parents and candidate != self.workspace_root:
|
|
raise ValueError(f"Path is outside the workspace: {candidate}")
|
|
return candidate
|