file_tools.py•5.67 kB
"""
File system tools for the MCP server.
"""
import logging
import os
import shutil
from . import mcp
@mcp.tool()
async def read_file(path: str) -> str:
"""
Read the contents of a file.
Args:
path: The path to the file.
Returns:
The file content or error message.
"""
logging.info("read_file called with path: %r", path)
try:
with open(path, "r", encoding="utf-8") as f:
content = f.read()
logging.info("File read successfully: %r", path)
return f"File: {path}\n\nContent:\n{content}"
except OSError as e:
logging.error("Error reading file %r: %s", path, e)
return f"Error reading file {path}: {e}"
@mcp.tool()
async def write_file(path: str, content: str) -> str:
"""
Write content to a file.
Args:
path: The path to the file.
content: The content to write.
Returns:
Success or error message.
"""
logging.info("write_file called with path: %r", path)
try:
with open(path, "w", encoding="utf-8") as f:
f.write(content)
logging.info("File written successfully: %r", path)
return f"Successfully wrote {len(content)} characters to {path}"
except OSError as e:
logging.error("Error writing file %r: %s", path, e)
return f"Error writing file {path}: {e}"
@mcp.tool()
async def create_file(path: str) -> str:
"""
Create an empty file at the specified path.
Args:
path: The path to the file to create.
Returns:
Success or error message.
"""
logging.info("create_file called with path: %r", path)
try:
if os.path.exists(path):
return f"File already exists: {path}"
with open(path, "x", encoding="utf-8"):
pass
logging.info("File created: %r", path)
return f"File created: {path}"
except OSError as e:
logging.error("Error creating file %r: %s", path, e)
return f"Error creating file {path}: {e}"
@mcp.tool()
async def create_folder(path: str) -> str:
"""
Create a new folder (directory) at the specified path.
Args:
path: The path to the folder to create.
Returns:
Success or error message.
"""
logging.info("create_folder called with path: %r", path)
try:
if os.path.exists(path):
return f"Folder already exists: {path}"
os.makedirs(path)
logging.info("Folder created: %r", path)
return f"Folder created: {path}"
except OSError as e:
logging.error("Error creating folder %r: %s", path, e)
return f"Error creating folder {path}: {e}"
@mcp.tool()
async def list_files(path: str = ".") -> str:
"""
List files and directories in the given path.
Args:
path: Directory path to list (default: current directory).
Returns:
Formatted list of files/directories or error message.
"""
logging.info("list_files called with path: %r", path)
try:
entries = os.listdir(path)
output = f"Contents of {os.path.abspath(path)}:\n"
for entry in entries:
full_path = os.path.join(path, entry)
if os.path.isdir(full_path):
output += f"[DIR] {entry}\n"
else:
output += f" {entry}\n"
return output
except OSError as e:
logging.error("Error listing files in %r: %s", path, e)
return f"Error listing files in {path}: {e}"
@mcp.tool()
async def delete_file(path: str) -> str:
"""
Delete a file.
Args:
path: The path to the file to delete.
Returns:
Success or error message.
"""
logging.info("delete_file called with path: %r", path)
try:
if not os.path.isfile(path):
return f"File not found: {path}"
os.remove(path)
logging.info("File deleted: %r", path)
return f"File deleted: {path}"
except OSError as e:
logging.error("Error deleting file %r: %s", path, e)
return f"Error deleting file {path}: {e}"
@mcp.tool()
async def rename_file(src: str, dst: str) -> str:
"""
Rename or move a file.
Args:
src: Source file path.
dst: Destination file path.
Returns:
Success or error message.
"""
logging.info("rename_file called with src=%r, dst=%r", src, dst)
try:
os.rename(src, dst)
logging.info("File renamed/moved from %r to %r", src, dst)
return f"File renamed/moved from {src} to {dst}"
except OSError as e:
logging.error("Error renaming/moving file %r to %r: %s", src, dst, e)
return f"Error renaming/moving file {src} to {dst}"
@mcp.tool()
async def move_folder(src: str, dst: str) -> str:
"""
Move a folder (directory) from src to dst.
Args:
src: Source folder path.
dst: Destination folder path.
Returns:
Success or error message.
"""
logging.info("move_folder called with src=%r, dst=%r", src, dst)
try:
if not os.path.isdir(src):
return f"Source folder not found: {src}"
# If destination exists and is a directory, move inside it
if os.path.isdir(dst):
dst_path = os.path.join(dst, os.path.basename(src))
else:
dst_path = dst
shutil.move(src, dst_path)
logging.info("Folder moved from %r to %r", src, dst_path)
return f"Folder moved from {src} to {dst_path}"
except OSError as e:
logging.error("Error moving folder %r to %r: %s", src, dst, e)
return f"Error moving folder {src} to {dst}"