"""File operation tools - read operations."""
import glob
from pathlib import Path
from typing import Any, Dict, List
import aiofiles
from ...config import get_config
from ...logger import get_logger
from ..base import tool
logger = get_logger(__name__)
config = get_config()
@tool(
name="file_read",
description="Read the complete contents of a file. File path must be relative to workspace root.",
input_schema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to the file (e.g., 'data/input.txt')",
},
"encoding": {
"type": "string",
"description": "File encoding (default: utf-8)",
"default": "utf-8",
},
},
"required": ["file_path"],
},
)
async def read_file(file_path: str, encoding: str = "utf-8") -> Dict[str, Any]:
"""
Read complete file contents.
Args:
file_path: Relative file path
encoding: File encoding
Returns:
Dictionary with file content and metadata
Raises:
ValueError: If path is invalid
FileNotFoundError: If file doesn't exist
"""
logger.info(f"file_read called: file_path={file_path}, encoding={encoding}")
# Validate and resolve path
try:
resolved_path = config.validate_file_path(file_path)
except ValueError as e:
logger.error(f"Path validation failed: {e}")
raise
# Check if file exists
if not resolved_path.exists():
logger.error(f"File not found: {resolved_path}")
raise FileNotFoundError(f"File not found: {file_path}")
# Check if it's a file (not directory)
if not resolved_path.is_file():
logger.error(f"Path is not a file: {resolved_path}")
raise ValueError(f"Path is not a file: {file_path}")
# Check file size
file_size = resolved_path.stat().st_size
if file_size > config.max_file_size:
logger.error(
f"File too large: {file_size} bytes (max: {config.max_file_size} bytes)"
)
raise ValueError(
f"File size ({file_size} bytes) exceeds maximum allowed size "
f"({config.max_file_size} bytes)"
)
# Read file
try:
async with aiofiles.open(resolved_path, "r", encoding=encoding) as f:
content = await f.read()
except UnicodeDecodeError as e:
logger.error(f"Encoding error reading file: {e}")
raise ValueError(f"Unable to read file with encoding '{encoding}': {e}")
# Count lines
line_count = content.count("\n") + (1 if content and not content.endswith("\n") else 0)
logger.debug(f"Successfully read file: {file_size} bytes, {line_count} lines")
return {
"content": content,
"metadata": {
"size_bytes": file_size,
"encoding": encoding,
"line_count": line_count,
"path": file_path,
},
}
@tool(
name="file_read_lines",
description="Read specific lines from a file. Useful for large files or when you only need part of the file.",
input_schema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to the file",
},
"start": {
"type": "integer",
"description": "Starting line number (0-indexed, default: 0)",
"default": 0,
"minimum": 0,
},
"count": {
"type": "integer",
"description": "Number of lines to read (-1 for all lines from start, default: -1)",
"default": -1,
},
"encoding": {
"type": "string",
"description": "File encoding (default: utf-8)",
"default": "utf-8",
},
},
"required": ["file_path"],
},
)
async def read_lines(
file_path: str, start: int = 0, count: int = -1, encoding: str = "utf-8"
) -> Dict[str, Any]:
"""
Read specific lines from a file.
Args:
file_path: Relative file path
start: Starting line (0-indexed)
count: Number of lines to read (-1 for all)
encoding: File encoding
Returns:
Dictionary with lines and metadata
"""
logger.info(
f"file_read_lines called: file_path={file_path}, start={start}, "
f"count={count}, encoding={encoding}"
)
# Validate and resolve path
try:
resolved_path = config.validate_file_path(file_path)
except ValueError as e:
logger.error(f"Path validation failed: {e}")
raise
# Check if file exists
if not resolved_path.exists():
logger.error(f"File not found: {resolved_path}")
raise FileNotFoundError(f"File not found: {file_path}")
# Check if it's a file
if not resolved_path.is_file():
logger.error(f"Path is not a file: {resolved_path}")
raise ValueError(f"Path is not a file: {file_path}")
# Read file lines
try:
async with aiofiles.open(resolved_path, "r", encoding=encoding) as f:
all_lines = await f.readlines()
except UnicodeDecodeError as e:
logger.error(f"Encoding error reading file: {e}")
raise ValueError(f"Unable to read file with encoding '{encoding}': {e}")
# Extract requested lines
total_lines = len(all_lines)
if start >= total_lines:
lines = []
elif count == -1:
lines = all_lines[start:]
else:
end = min(start + count, total_lines)
lines = all_lines[start:end]
# Remove trailing newlines for cleaner output
lines = [line.rstrip("\n") for line in lines]
logger.debug(
f"Read {len(lines)} lines from file (total: {total_lines}, start: {start})"
)
return {
"lines": lines,
"metadata": {
"total_lines": total_lines,
"start_line": start,
"lines_returned": len(lines),
"encoding": encoding,
"path": file_path,
},
}
@tool(
name="file_exists",
description="Check if a file exists at the given path",
input_schema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to check",
}
},
"required": ["file_path"],
},
)
async def file_exists(file_path: str) -> Dict[str, Any]:
"""
Check if a file exists.
Args:
file_path: Relative file path
Returns:
Dictionary with existence status and metadata
"""
logger.info(f"file_exists called: file_path={file_path}")
# Validate and resolve path
try:
resolved_path = config.validate_file_path(file_path)
except ValueError as e:
logger.error(f"Path validation failed: {e}")
raise
exists = resolved_path.exists()
is_file = resolved_path.is_file() if exists else False
is_dir = resolved_path.is_dir() if exists else False
logger.debug(f"File exists check: exists={exists}, is_file={is_file}, is_dir={is_dir}")
return {
"exists": exists,
"is_file": is_file,
"is_directory": is_dir,
"path": file_path,
}
@tool(
name="file_list",
description="List files in a directory with optional glob pattern matching",
input_schema={
"type": "object",
"properties": {
"directory": {
"type": "string",
"description": "Relative directory path (default: '.' for workspace root)",
"default": ".",
},
"pattern": {
"type": "string",
"description": "Glob pattern to filter files (default: '*' for all files)",
"default": "*",
},
},
"required": [],
},
)
async def list_files(directory: str = ".", pattern: str = "*") -> Dict[str, Any]:
"""
List files in a directory.
Args:
directory: Relative directory path
pattern: Glob pattern for filtering
Returns:
Dictionary with file list and metadata
"""
logger.info(f"file_list called: directory={directory}, pattern={pattern}")
# Validate and resolve directory path
try:
resolved_dir = config.validate_file_path(directory)
except ValueError as e:
logger.error(f"Path validation failed: {e}")
raise
# Check if directory exists
if not resolved_dir.exists():
logger.error(f"Directory not found: {resolved_dir}")
raise FileNotFoundError(f"Directory not found: {directory}")
# Check if it's a directory
if not resolved_dir.is_dir():
logger.error(f"Path is not a directory: {resolved_dir}")
raise ValueError(f"Path is not a directory: {directory}")
# List files with pattern
try:
full_pattern = str(resolved_dir / pattern)
matched_paths = glob.glob(full_pattern)
# Convert to relative paths and separate files/directories
workspace_root = config.workspace_root
files = []
directories = []
for path_str in matched_paths:
path = Path(path_str)
try:
relative_path = path.relative_to(workspace_root)
relative_str = str(relative_path).replace("\\", "/")
if path.is_file():
files.append(relative_str)
elif path.is_dir():
directories.append(relative_str)
except ValueError:
# Path is outside workspace, skip it
logger.warning(f"Skipping path outside workspace: {path}")
continue
except Exception as e:
logger.error(f"Error listing directory: {e}")
raise ValueError(f"Error listing directory: {e}")
logger.debug(f"Listed {len(files)} files and {len(directories)} directories")
return {
"files": sorted(files),
"directories": sorted(directories),
"metadata": {
"directory": directory,
"pattern": pattern,
"total_files": len(files),
"total_directories": len(directories),
},
}
