77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
#!/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')
|