Extending MCP Servers with Dependencies

"""
Enhanced Filesystem Server — wraps filesystem-mcp-server and adds:
  - audit_log: logs every tool call
  - safe_write: validates paths before writing (overrides write_file)
"""
import asyncio
import json
import sys
from datetime import datetime, timezone

# The upstream server runs as a subprocess via stdio
# Pharos installed it, so it's available as a local MCP server

UPSTREAM_COMMAND = "python"
UPSTREAM_ARGS = ["-m", "filesystem_mcp_server"]


class EnhancedFilesystemServer:
    def __init__(self):
        self.upstream = None
        self.audit_log: list[dict] = []

    async def start(self):
        """Start the upstream server as a subprocess."""
        self.upstream = await asyncio.create_subprocess_exec(
            UPSTREAM_COMMAND,
            *UPSTREAM_ARGS,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

    async def _call_upstream(self, method: str, params: dict) -> dict:
        """Forward a JSON-RPC request to the upstream server."""
        request = {
            "jsonrpc": "2.0",
            "id": len(self.audit_log) + 1,
            "method": method,
            "params": params,
        }
        line = json.dumps(request) + "\n"
        self.upstream.stdin.write(line.encode())
        await self.upstream.stdin.drain()

        response_line = await self.upstream.stdout.readline()
        return json.loads(response_line)

    def _audit(self, tool: str, args: dict, result: str):
        """Record every tool call."""
        self.audit_log.append({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "tool": tool,
            "args": args,
            "result_summary": result[:200],
        })

    # ─── Pass-through tools ──────────────────────────────────

    async def read_file(self, path: str) -> str:
        """Read a file — delegates to upstream, logs the call."""
        result = await self._call_upstream("tools/call", {
            "name": "read_file",
            "arguments": {"path": path},
        })
        content = result.get("result", "")
        self._audit("read_file", {"path": path}, content)
        return content

    async def list_directory(self, path: str) -> str:
        """List a directory — delegates to upstream, logs the call."""
        result = await self._call_upstream("tools/call", {
            "name": "list_directory",
            "arguments": {"path": path},
        })
        content = result.get("result", "")
        self._audit("list_directory", {"path": path}, content)
        return content

    # ─── Overridden tools ────────────────────────────────────

    BLOCKED_PATTERNS = ["/etc/", "/sys/", "/proc/", "~/.ssh/"]

    async def write_file(self, path: str, content: str) -> str:
        """
        Override write_file — validates the path before delegating.
        The upstream server's write_file is never called for blocked paths.
        """
        for pattern in self.BLOCKED_PATTERNS:
            if pattern in path:
                self._audit("write_file", {"path": path}, "BLOCKED")
                return f"Error: path '{path}' is in a protected directory"

        result = await self._call_upstream("tools/call", {
            "name": "write_file",
            "arguments": {"path": path, "content": content},
        })
        response = result.get("result", "")
        self._audit("write_file", {"path": path}, response)
        return response

    # ─── New tools ───────────────────────────────────────────

    async def get_audit_log(self) -> str:
        """New tool — returns the audit log. Not in the upstream server."""
        return json.dumps(self.audit_log, indent=2)

# Validate first
pharos publish --dry-run

# Publish to the registry
pharos publish

# Users install your server — dependencies come automatically
pharos install enhanced-filesystem-server