server.py

import os
import shutil
import glob
import hashlib
import zipfile
from typing import List, Optional, Dict, Any
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("Filesystem")

@mcp.tool()
def list_directory(path: str = ".") -> List[str]:
    """Lists files and directories."""
    try: return sorted(os.listdir(path))
    except Exception as e: return [f"Error: {str(e)}"]

@mcp.tool()
def read_file(path: str, start_line: Optional[int] = None, end_line: Optional[int] = None) -> str:
    """Reads file content, optionally within line range."""
    try:
        with open(path, "r") as f: lines = f.readlines()
        if start_line or end_line:
            start = (start_line - 1) if start_line else 0
            end = end_line if end_line else len(lines)
            return "".join(lines[start:end])
        return "".join(lines)
    except Exception as e: return f"Error: {str(e)}"

@mcp.tool()
def write_file(path: str, content: str) -> str:
    """Writes content to a file."""
    try:
        dir_p = os.path.dirname(path)
        if dir_p: os.makedirs(dir_p, exist_ok=True)
        with open(path, "w") as f: f.write(content)
        return f"Wrote to {path}"
    except Exception as e: return f"Error: {str(e)}"

@mcp.tool()
def search_files(pattern: str, root: str = ".") -> List[str]:
    """Recursive glob search."""
    try: return sorted(glob.glob(os.path.join(root, "**", pattern), recursive=True))
    except Exception as e: return [f"Error: {str(e)}"]

@mcp.tool()
def find_and_replace_in_files(directory: str, search_str: str, replace_str: str, pattern: str = "*.*"):
    """Bulk search and replace in a directory."""
    results = []
    files = glob.glob(os.path.join(directory, "**", pattern), recursive=True)
    for f_path in files:
        if os.path.isfile(f_path):
            try:
                with open(f_path, "r") as f: content = f.read()
                if search_str in content:
                    with open(f_path, "w") as f: f.write(content.replace(search_str, replace_str))
                    results.append(f"Updated: {f_path}")
            except: continue
    return results

@mcp.tool()
def get_tree_view(path: str = ".", depth: int = 2) -> str:
    """Returns a visual markdown tree of the directory."""
    def _tree(dir_path, level):
        if level > depth: return ""
        text = ""
        try:
            for item in sorted(os.listdir(dir_path)):
                if item.startswith('.'): continue
                full_p = os.path.join(dir_path, item)
                text += "  " * level + "├── " + item + "\n"
                if os.path.isdir(full_p):
                    text += _tree(full_p, level + 1)
        except: pass
        return text
    return f"# Tree: {os.path.abspath(path)}\n" + _tree(path, 0)

@mcp.tool()
def calculate_hash(path: str) -> str:
    """SHA256 hash."""
    sha = hashlib.sha256()
    try:
        with open(path, "rb") as f:
            for b in iter(lambda: f.read(4096), b""): sha.update(b)
        return sha.hexdigest()
    except Exception as e: return str(e)

@mcp.tool()
def zip_directory(src: str, out: str) -> str:
    """Zips a folder."""
    try:
        shutil.make_archive(out.replace('.zip', ''), 'zip', src)
        return f"Archive {out} created."
    except Exception as e: return str(e)

@mcp.tool()
def move_path(src: str, dst: str):
    """Moves/Renames a path."""
    try:
        shutil.move(src, dst)
        return f"Moved to {dst}"
    except Exception as e: return str(e)

if __name__ == "__main__":
    mcp.run()