88 lines
2.5 KiB
Python
88 lines
2.5 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("filesystem", host="0.0.0.0", port=8010)
|
|
|
|
|
|
def _allowed_roots() -> list[Path]:
|
|
roots_raw = os.getenv("MCP_FILESYSTEM_ROOTS", "/workspace").split(",")
|
|
roots = [Path(root.strip()).resolve() for root in roots_raw if root.strip()]
|
|
return roots or [Path("/workspace").resolve()]
|
|
|
|
|
|
def _resolve(path: str) -> Path:
|
|
candidate = Path(path)
|
|
if not candidate.is_absolute():
|
|
candidate = Path("/workspace") / candidate
|
|
resolved = candidate.resolve()
|
|
|
|
for root in _allowed_roots():
|
|
try:
|
|
resolved.relative_to(root)
|
|
return resolved
|
|
except ValueError:
|
|
continue
|
|
|
|
allowed = ", ".join(str(r) for r in _allowed_roots())
|
|
raise ValueError(f"Path '{resolved}' is outside allowed roots: {allowed}")
|
|
|
|
|
|
@mcp.tool()
|
|
def list_roots() -> list[str]:
|
|
"""List configured filesystem roots that this MCP server can access."""
|
|
return [str(root) for root in _allowed_roots()]
|
|
|
|
|
|
@mcp.tool()
|
|
def list_dir(path: str = ".") -> list[str]:
|
|
"""List entries in a directory relative to an allowed root."""
|
|
target = _resolve(path)
|
|
if not target.exists() or not target.is_dir():
|
|
raise ValueError(f"Not a directory: {target}")
|
|
|
|
items = []
|
|
for entry in sorted(target.iterdir(), key=lambda p: p.name.lower()):
|
|
suffix = "/" if entry.is_dir() else ""
|
|
items.append(f"{entry.name}{suffix}")
|
|
return items
|
|
|
|
|
|
@mcp.tool()
|
|
def read_text(path: str, max_chars: int = 50000) -> str:
|
|
"""Read text from a file in an allowed root."""
|
|
target = _resolve(path)
|
|
if not target.exists() or not target.is_file():
|
|
raise ValueError(f"Not a file: {target}")
|
|
|
|
content = target.read_text(encoding="utf-8")
|
|
return content[:max_chars]
|
|
|
|
|
|
@mcp.tool()
|
|
def write_text(path: str, content: str, append: bool = False) -> str:
|
|
"""Write text to a file in an allowed root."""
|
|
target = _resolve(path)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
mode = "a" if append else "w"
|
|
with target.open(mode, encoding="utf-8") as fh:
|
|
fh.write(content)
|
|
|
|
return str(target)
|
|
|
|
|
|
@mcp.tool()
|
|
def find_files(path: str = ".", pattern: str = "*") -> list[str]:
|
|
"""Find files by glob pattern under a path in an allowed root."""
|
|
target = _resolve(path)
|
|
if not target.exists() or not target.is_dir():
|
|
raise ValueError(f"Not a directory: {target}")
|
|
|
|
return [str(p) for p in target.rglob(pattern) if p.is_file()]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport="sse")
|