Stack-2-9-finetuned / src /tools /file_read.py
walidsobhie-code
feat: Add remaining RTMP tools (FileRead, FileWrite, Sleep, AskQuestion, Brief, TaskGet, TeamDelete, MCPTool, Worktree, SyntheticOutput)
5dc5419
"""FileReadTool - Read file contents for Stack 2.9"""
import os
from pathlib import Path
from typing import Optional
from .base import BaseTool, ToolResult
from .registry import tool_registry
class FileReadTool(BaseTool):
"""Read contents of a file."""
name = "file_read"
description = "Read contents of a file"
input_schema = {
"type": "object",
"properties": {
"path": {"type": "string", "description": "File path to read"},
"offset": {"type": "number", "description": "Line offset to start reading"},
"limit": {"type": "number", "description": "Maximum lines to read"},
"show_lines": {"type": "boolean", "default": False, "description": "Include line numbers"}
},
"required": ["path"]
}
async def execute(self, path: str, offset: Optional[int] = None, limit: Optional[int] = None, show_lines: bool = False) -> ToolResult:
"""Read file."""
file_path = Path(path)
if not file_path.exists():
return ToolResult(success=False, error=f"File not found: {path}")
if not file_path.is_file():
return ToolResult(success=False, error=f"Not a file: {path}")
try:
lines = file_path.read_text().split('\n')
except Exception as e:
return ToolResult(success=False, error=f"Cannot read file: {e}")
total_lines = len(lines)
# Apply offset
if offset is not None and offset > 0:
lines = lines[offset:]
elif offset is not None:
offset = 0
# Apply limit
if limit is not None and limit > 0:
lines = lines[:limit]
if show_lines:
start_line = (offset or 0) + 1
content = '\n'.join(f"{i}: {line}" for i, line in enumerate(lines, start_line))
else:
content = '\n'.join(lines)
return ToolResult(success=True, data={
"path": path,
"content": content,
"lines_read": len(lines),
"total_lines": total_lines,
"offset": offset,
"truncated": limit is not None and len(lines) >= limit
})
class FileExistsTool(BaseTool):
"""Check if a file exists."""
name = "file_exists"
description = "Check if a file or directory exists"
input_schema = {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to check"}
},
"required": ["path"]
}
async def execute(self, path: str) -> ToolResult:
"""Check existence."""
p = Path(path)
return ToolResult(success=True, data={
"path": path,
"exists": p.exists(),
"is_file": p.is_file() if p.exists() else None,
"is_dir": p.is_dir() if p.exists() else None
})
# Register tools
tool_registry.register(FileReadTool())
tool_registry.register(FileExistsTool())