From ba6634aa716b0caea645e66a5040c083e763dd39 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 15 Jul 2026 17:53:13 -0400 Subject: [PATCH] MCP server source --- Dockerfile | 8 +++++ requirements.txt | 1 + server.py | 76 ++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 85 insertions(+) create mode 100644 Dockerfile create mode 100644 requirements.txt create mode 100644 server.py diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..e8385ea --- /dev/null +++ b/Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.13-slim +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY server.py ./ +ENV PYTHONUNBUFFERED=1 +EXPOSE 8900 +CMD ["python3", "server.py"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..886b069 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +fastmcp>=3.0.0 diff --git a/server.py b/server.py new file mode 100644 index 0000000..d133436 --- /dev/null +++ b/server.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Filesystem MCP Server — Safe file operations for Open WebUI.""" +from fastmcp import FastMCP +import os, glob, fnmatch + +ALLOWED_ROOTS = ['/root/projects', '/root/docker'] +mcp = FastMCP('Filesystem', version='1.0.0') + +def _safe_path(path: str) -> str: + """Resolve and validate path is within allowed roots.""" + real = os.path.realpath(path) + for root in ALLOWED_ROOTS: + if real.startswith(os.path.realpath(root) + os.sep) or real == os.path.realpath(root): + return real + raise ValueError(f'Path outside allowed roots: {path}. Allowed: {ALLOWED_ROOTS}') + +@mcp.tool(description="List files and folders in a directory") +async def list_directory(path: str = '/root/projects') -> str: + """List contents of a directory.""" + real = _safe_path(path) + if not os.path.isdir(real): + return f'Not a directory: {path}' + items = os.listdir(real) + result = [f'Directory: {real}'] + for item in sorted(items): + full = os.path.join(real, item) + typ = 'DIR' if os.path.isdir(full) else 'FILE' + size = os.path.getsize(full) if os.path.isfile(full) else 0 + result.append(f' [{typ}] {item} ({size:,} bytes)') + return '\n'.join(result[:100]) + +@mcp.tool(description="Read a file's contents, optionally limited to N lines") +async def read_file(path: str, max_lines: int = 100) -> str: + """Read file contents.""" + real = _safe_path(path) + if not os.path.isfile(real): + return f'Not a file: {path}' + with open(real) as f: + lines = f.readlines() + if max_lines and len(lines) > max_lines: + return ''.join(lines[:max_lines]) + f'\n... ({len(lines)} total lines, showing first {max_lines})' + return ''.join(lines) + +@mcp.tool(description="Write content to a file (overwrites existing)") +async def write_file(path: str, content: str) -> str: + """Write content to a file.""" + real = _safe_path(path) + os.makedirs(os.path.dirname(real), exist_ok=True) + with open(real, 'w') as f: + f.write(content) + return f'Written {len(content):,} bytes to {real}' + +@mcp.tool(description="Search for files by name pattern (glob)") +async def search_files(pattern: str, directory: str = '/root/projects') -> str: + """Search for files matching a glob pattern.""" + real = _safe_path(directory) + if not os.path.isdir(real): + return f'Not a directory: {directory}' + matches = [] + for root, dirs, files in os.walk(real): + for name in files + dirs: + if fnmatch.fnmatch(name, f'*{pattern}*'): + matches.append(os.path.join(root, name)) + if not matches: + return f'No files matching "{pattern}" in {real}' + return f'Found {len(matches)} matches:\n' + '\n'.join(matches[:50]) + +@mcp.tool(description="Create a new directory") +async def create_directory(path: str) -> str: + """Create a directory and parents.""" + real = _safe_path(path) + os.makedirs(real, exist_ok=True) + return f'Created directory: {real}' + +if __name__ == '__main__': + mcp.run(transport='http', host='0.0.0.0', port=8900, path='/mcp')