"""File operation tools - write operations."""
from pathlib import Path
from typing import Any, Dict
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_write",
description="Write content to a file, creating it if it doesn't exist or overwriting if it does. Creates parent directories as needed.",
input_schema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to the file",
},
"content": {
"type": "string",
"description": "Content to write to the file",
},
"encoding": {
"type": "string",
"description": "File encoding (default: utf-8)",
"default": "utf-8",
},
},
"required": ["file_path", "content"],
},
)
async def write_file(file_path: str, content: str, encoding: str = "utf-8") -> Dict[str, Any]:
"""
Write content to a file (overwrite mode).
Args:
file_path: Relative file path
content: Content to write
encoding: File encoding
Returns:
Dictionary with operation result and metadata
"""
logger.info(f"file_write called: file_path={file_path}, content_length={len(content)}")
# 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
# Create parent directories if they don't exist
resolved_path.parent.mkdir(parents=True, exist_ok=True)
# Check content size
content_bytes = len(content.encode(encoding))
if content_bytes > config.max_file_size:
logger.error(
f"Content too large: {content_bytes} bytes (max: {config.max_file_size} bytes)"
)
raise ValueError(
f"Content size ({content_bytes} bytes) exceeds maximum allowed size "
f"({config.max_file_size} bytes)"
)
# Write file
try:
async with aiofiles.open(resolved_path, "w", encoding=encoding) as f:
await f.write(content)
except Exception as e:
logger.error(f"Error writing file: {e}")
raise ValueError(f"Error writing file: {e}")
# Verify file was written
file_size = resolved_path.stat().st_size
logger.info(f"Successfully wrote {file_size} bytes to {file_path}")
return {
"success": True,
"bytes_written": file_size,
"path": file_path,
"operation": "write",
}
@tool(
name="file_append",
description="Append content to the end of an existing file. Creates the file if it doesn't exist.",
input_schema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to the file",
},
"content": {
"type": "string",
"description": "Content to append to the file",
},
"encoding": {
"type": "string",
"description": "File encoding (default: utf-8)",
"default": "utf-8",
},
},
"required": ["file_path", "content"],
},
)
async def append_file(file_path: str, content: str, encoding: str = "utf-8") -> Dict[str, Any]:
"""
Append content to a file.
Args:
file_path: Relative file path
content: Content to append
encoding: File encoding
Returns:
Dictionary with operation result and metadata
"""
logger.info(f"file_append called: file_path={file_path}, content_length={len(content)}")
# 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
# Create parent directories if they don't exist
resolved_path.parent.mkdir(parents=True, exist_ok=True)
# Get original file size
original_size = resolved_path.stat().st_size if resolved_path.exists() else 0
# Check total size after append
content_bytes = len(content.encode(encoding))
total_size = original_size + content_bytes
if total_size > config.max_file_size:
logger.error(
f"Total size after append would be too large: {total_size} bytes "
f"(max: {config.max_file_size} bytes)"
)
raise ValueError(
f"Total file size after append ({total_size} bytes) would exceed "
f"maximum allowed size ({config.max_file_size} bytes)"
)
# Append to file
try:
async with aiofiles.open(resolved_path, "a", encoding=encoding) as f:
await f.write(content)
except Exception as e:
logger.error(f"Error appending to file: {e}")
raise ValueError(f"Error appending to file: {e}")
# Verify file size
new_size = resolved_path.stat().st_size
bytes_appended = new_size - original_size
logger.info(f"Successfully appended {bytes_appended} bytes to {file_path}")
return {
"success": True,
"bytes_appended": bytes_appended,
"total_size": new_size,
"path": file_path,
"operation": "append",
}
@tool(
name="file_create",
description="Create a new file with given content. Fails if the file already exists.",
input_schema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to the new file",
},
"content": {
"type": "string",
"description": "Initial content for the file",
"default": "",
},
"encoding": {
"type": "string",
"description": "File encoding (default: utf-8)",
"default": "utf-8",
},
},
"required": ["file_path"],
},
)
async def create_file(
file_path: str, content: str = "", encoding: str = "utf-8"
) -> Dict[str, Any]:
"""
Create a new file (fails if exists).
Args:
file_path: Relative file path
content: Initial content
encoding: File encoding
Returns:
Dictionary with operation result and metadata
Raises:
FileExistsError: If file already exists
"""
logger.info(f"file_create called: file_path={file_path}, content_length={len(content)}")
# 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 already exists
if resolved_path.exists():
logger.error(f"File already exists: {resolved_path}")
raise FileExistsError(f"File already exists: {file_path}")
# Create parent directories if they don't exist
resolved_path.parent.mkdir(parents=True, exist_ok=True)
# Check content size
content_bytes = len(content.encode(encoding))
if content_bytes > config.max_file_size:
logger.error(
f"Content too large: {content_bytes} bytes (max: {config.max_file_size} bytes)"
)
raise ValueError(
f"Content size ({content_bytes} bytes) exceeds maximum allowed size "
f"({config.max_file_size} bytes)"
)
# Create file
try:
async with aiofiles.open(resolved_path, "x", encoding=encoding) as f:
await f.write(content)
except FileExistsError:
logger.error(f"File already exists: {resolved_path}")
raise FileExistsError(f"File already exists: {file_path}")
except Exception as e:
logger.error(f"Error creating file: {e}")
raise ValueError(f"Error creating file: {e}")
# Verify file was created
file_size = resolved_path.stat().st_size
logger.info(f"Successfully created file {file_path} with {file_size} bytes")
return {
"success": True,
"bytes_written": file_size,
"path": file_path,
"operation": "create",
}
@tool(
name="file_delete",
description="Delete a file. Fails if the file doesn't exist or is a directory.",
input_schema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to the file to delete",
}
},
"required": ["file_path"],
},
)
async def delete_file(file_path: str) -> Dict[str, Any]:
"""
Delete a file.
Args:
file_path: Relative file path
Returns:
Dictionary with operation result
Raises:
FileNotFoundError: If file doesn't exist
ValueError: If path is a directory
"""
logger.info(f"file_delete 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
# 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 (cannot delete directories): {file_path}")
# Delete file
try:
resolved_path.unlink()
except Exception as e:
logger.error(f"Error deleting file: {e}")
raise ValueError(f"Error deleting file: {e}")
logger.info(f"Successfully deleted file {file_path}")
return {
"success": True,
"path": file_path,
"operation": "delete",
}
