ohmu/third/server/openwebui/mcp-servers/filesystem_server.py

import os
import stat
from pathlib import Path

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("filesystem", host="0.0.0.0", port=8010)

ALLOWED_DIRS = [
    "/projects",
    "/data",
    "/scripts",
    "/backups",
    "/tmp",
]


def _resolve_path(path: str) -> Path:
    return Path(path).resolve()


def _is_path_allowed(path: str) -> bool:
    resolved = _resolve_path(path)
    for allowed in ALLOWED_DIRS:
        allowed_path = Path(allowed).resolve()
        if allowed_path in resolved.parents or allowed_path == resolved:
            return True
    return False


@mcp.tool()
def read_file(path: str) -> str:
    """Read the complete contents of a text file at the given path.

    Args:
        path: Absolute path to the file to read.
    """
    if not _is_path_allowed(path):
        return f"Error: path '{path}' is not in allowed directories: {', '.join(ALLOWED_DIRS)}"
    try:
        with open(path, "r") as f:
            return f.read()
    except Exception as e:
        return f"Error reading file: {e}"


@mcp.tool()
def write_file(path: str, content: str) -> str:
    """Write or overwrite a file with the given content.

    Creates parent directories if they do not exist.

    Args:
        path: Absolute path to the file to write.
        content: Text content to write into the file.
    """
    if not _is_path_allowed(path):
        return f"Error: path '{path}' is not in allowed directories: {', '.join(ALLOWED_DIRS)}"
    try:
        p = _resolve_path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content)
        return f"OK - wrote {len(content)} bytes to {path}"
    except Exception as e:
        return f"Error writing file: {e}"


@mcp.tool()
def list_directory(path: str) -> str:
    """List files and directories at the given path.

    Shows name, size (for files), and modification time.

    Args:
        path: Absolute path to the directory.
    """
    if not _is_path_allowed(path):
        return f"Error: path '{path}' is not in allowed directories: {', '.join(ALLOWED_DIRS)}"
    try:
        entries = []
        for entry in os.scandir(path):
            info = entry.stat()
            size = info.st_size if entry.is_file() else 0
            mtime = info.st_mtime
            kind = "F" if entry.is_file() else "D"
            entries.append(f"{kind} {entry.name:40s} {size:>10d} B")
        return "\n".join(sorted(entries))
    except Exception as e:
        return f"Error listing directory: {e}"


@mcp.tool()
def get_file_info(path: str) -> str:
    """Get metadata about a file or directory.

    Args:
        path: Absolute path to the file or directory.
    """
    if not _is_path_allowed(path):
        return f"Error: path '{path}' is not in allowed directories: {', '.join(ALLOWED_DIRS)}"
    try:
        st = os.stat(path)
        kind = "directory" if stat.S_ISDIR(st.st_mode) else "file"
        return (
            f"Path: {path}\n"
            f"Type: {kind}\n"
            f"Size: {st.st_size} B\n"
            f"Mode: {oct(st.st_mode)}\n"
            f"Modified: {st.st_mtime}"
        )
    except Exception as e:
        return f"Error: {e}"


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