Add MCP roots support

When the client advertises the roots capability during initialize,
the server reciprocates with roots.listChanged in its capabilities,
fires a roots/list request once notifications/initialized arrives,
and re-fetches on every notifications/roots/list_changed. The first
file:// URI in the response becomes the FileSystemProvider root, so
one MCP server can follow the user across projects without a
restart. ResourceProvider gains a no-op set_workspace_root hook;
ChainProvider fans out, FileSystemProvider re-targets.

Author: JE-Chen

je_auto_control/utils/mcp_server/resources.py

@@ -41,6 +41,10 @@ def read(self, uri: str) -> Optional[Dict[str, Any]]:  # pragma: no cover - abst
         """Return one content block (``{uri, mimeType, text}``) or ``None``."""
         raise NotImplementedError
 
+    def set_workspace_root(self, root: str) -> None:
+        """Hook for MCP roots. Default: no-op. FS-backed providers override."""
+        del root
+
 
 class FileSystemProvider(ResourceProvider):
     """Expose ``*.json`` action files in ``root`` under ``<scheme>://files/<name>``."""
@@ -50,6 +54,10 @@ def __init__(self, root: str = ".",
         self.root = os.path.realpath(root)
         self.scheme = scheme
 
+    def set_workspace_root(self, root: str) -> None:
+        """Re-target the provider at a new directory (e.g. via MCP roots)."""
+        self.root = os.path.realpath(os.fspath(root))
+
     def list(self) -> List[MCPResource]:
         if not os.path.isdir(self.root):
             return []
@@ -160,6 +168,11 @@ def read(self, uri: str) -> Optional[Dict[str, Any]]:
                 return content
         return None
 
+    def set_workspace_root(self, root: str) -> None:
+        """Forward the root to every child provider."""
+        for provider in self.providers:
+            provider.set_workspace_root(root)
+
 
 def default_resource_provider(root: str = ".") -> ResourceProvider:
     """Return the resource provider exposed by the default MCP server."""

je_auto_control/utils/mcp_server/server.py

@@ -74,8 +74,10 @@ def __init__(self, tools: Optional[List[MCPTool]] = None,
         self._calls_lock = threading.Lock()
         self._write_lock = threading.Lock()
         self._sampling_id_counter = itertools.count(1)
+        self._outbound_id_counter = itertools.count(1)
         self._pending_outbound: Dict[Any, Dict[str, Any]] = {}
         self._outbound_lock = threading.Lock()
+        self._client_capabilities: Dict[str, Any] = {}
 
     def register_tool(self, tool: MCPTool) -> None:
         """Add or replace a tool in the live registry.
@@ -251,12 +253,76 @@ def _handle_notification(self, method: Optional[str],
         if method == "notifications/initialized":
             self._initialized = True
             autocontrol_logger.info("MCP client initialized")
+            self._maybe_request_roots_async()
             return
         if method == "notifications/cancelled":
             self._cancel_active_call(params)
             return
+        if method == "notifications/roots/list_changed":
+            self._maybe_request_roots_async()
+            return
         autocontrol_logger.debug("MCP notification ignored: %s", method)
 
+    def _maybe_request_roots_async(self) -> None:
+        """Fire a roots/list request when the client supports it."""
+        if "roots" not in self._client_capabilities:
+            return
+        if self._writer is None:
+            return
+        threading.Thread(
+            target=self._refresh_roots_safely, daemon=True,
+            name="MCPRootsRefresh",
+        ).start()
+
+    def _refresh_roots_safely(self) -> None:
+        try:
+            self.refresh_roots(timeout=5.0)
+        except (RuntimeError, TimeoutError) as error:
+            autocontrol_logger.info("MCP roots refresh skipped: %r", error)
+
+    def refresh_roots(self, timeout: float = 10.0) -> List[Dict[str, Any]]:
+        """Send ``roots/list`` to the client and apply the first root."""
+        result = self._send_outbound_request(
+            "roots/list", params={}, timeout=timeout,
+        )
+        roots_list = (result or {}).get("roots") or []
+        if not isinstance(roots_list, list) or not roots_list:
+            return []
+        first_uri = roots_list[0].get("uri") if isinstance(roots_list[0],
+                                                            dict) else None
+        if isinstance(first_uri, str):
+            local_path = _file_uri_to_path(first_uri)
+            if local_path:
+                self._resources.set_workspace_root(local_path)
+                autocontrol_logger.info("MCP workspace root → %s", local_path)
+        return roots_list
+
+    def _send_outbound_request(self, method: str,
+                               params: Dict[str, Any],
+                               timeout: float = 10.0) -> Dict[str, Any]:
+        """Send a server-initiated request and wait for the response."""
+        writer = self._writer
+        if writer is None:
+            raise RuntimeError(f"{method} requires an outbound writer")
+        request_id = f"srv-{next(self._outbound_id_counter)}"
+        slot = {"event": threading.Event()}
+        with self._outbound_lock:
+            self._pending_outbound[request_id] = slot
+        envelope = json.dumps({
+            "jsonrpc": "2.0", "id": request_id,
+            "method": method, "params": params,
+        }, ensure_ascii=False, default=str)
+        try:
+            writer(envelope)
+            if not slot["event"].wait(timeout=timeout):
+                raise TimeoutError(f"{method} timed out after {timeout}s")
+        finally:
+            with self._outbound_lock:
+                self._pending_outbound.pop(request_id, None)
+        if "error" in slot:
+            raise RuntimeError(f"{method} failed: {slot['error']}")
+        return slot.get("result") or {}
+
     def _cancel_active_call(self, params: Dict[str, Any]) -> None:
         """Mark the matching active tool call as cancelled, if any."""
         request_id = params.get("requestId")
@@ -293,17 +359,22 @@ def _dispatch(self, msg_id: Any, method: Optional[str],
             return self._handle_prompts_get(params)
         raise _MCPError(-32601, f"Method not found: {method}")
 
-    @staticmethod
-    def _handle_initialize(params: Dict[str, Any]) -> Dict[str, Any]:
+    def _handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
         client_version = params.get("protocolVersion", PROTOCOL_VERSION)
+        client_caps = params.get("capabilities") or {}
+        if isinstance(client_caps, dict):
+            self._client_capabilities = client_caps
+        capabilities: Dict[str, Any] = {
+            "tools": {"listChanged": True},
+            "resources": {"listChanged": False, "subscribe": False},
+            "prompts": {"listChanged": False},
+            "sampling": {},
+        }
+        if "roots" in self._client_capabilities:
+            capabilities["roots"] = {"listChanged": True}
         return {
             "protocolVersion": client_version or PROTOCOL_VERSION,
-            "capabilities": {
-                "tools": {"listChanged": True},
-                "resources": {"listChanged": False, "subscribe": False},
-                "prompts": {"listChanged": False},
-                "sampling": {},
-            },
+            "capabilities": capabilities,
             "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
         }
 
@@ -465,6 +536,20 @@ def _stringify_result(value: Any) -> str:
         return repr(value)
 
 
+def _file_uri_to_path(uri: str) -> Optional[str]:
+    """Convert a ``file://`` URI to a local filesystem path; ``None`` otherwise."""
+    if not isinstance(uri, str) or not uri.startswith("file://"):
+        return None
+    from urllib.parse import unquote, urlparse
+    parsed = urlparse(uri)
+    raw_path = unquote(parsed.path)
+    # Windows: file:///C:/foo strips the leading slash before the drive letter.
+    if sys.platform.startswith("win") and raw_path.startswith("/") and \
+            len(raw_path) > 2 and raw_path[2] == ":":
+        raw_path = raw_path[1:]
+    return raw_path or None
+
+
 def _notification_message(method: str, params: Dict[str, Any]) -> str:
     return json.dumps({"jsonrpc": "2.0", "method": method, "params": params},
                       ensure_ascii=False, default=str)

test/unit_test/headless/test_mcp_server.py

@@ -6,6 +6,7 @@
 """
 import io
 import json
+import os
 import threading
 from typing import Any, Dict, List
 
@@ -1135,6 +1136,63 @@ def test_rate_limiter_zero_rate_means_unlimited():
         assert limiter.try_acquire() is True
 
 
+def test_initialize_advertises_roots_when_client_supports_it():
+    server = MCPServer(tools=[])
+    response = _decode(server.handle_line(_request("initialize", params={
+        "capabilities": {"roots": {"listChanged": True}},
+    })))
+    assert "roots" in response["result"]["capabilities"]
+
+
+def test_initialize_omits_roots_when_client_lacks_capability():
+    server = MCPServer(tools=[])
+    response = _decode(server.handle_line(_request("initialize", params={
+        "capabilities": {},
+    })))
+    assert "roots" not in response["result"]["capabilities"]
+
+
+def test_refresh_roots_updates_filesystem_provider(tmp_path):
+    from je_auto_control.utils.mcp_server.resources import (
+        ChainProvider, FileSystemProvider,
+    )
+    fs_provider = FileSystemProvider(root=str(tmp_path / "initial"))
+    chain = ChainProvider([fs_provider])
+    captured_lines = []
+    server = MCPServer(tools=[], resource_provider=chain,
+                       concurrent_tools=True)
+    server.set_writer(captured_lines.append)
+    # Simulate client capability so refresh is allowed.
+    server._client_capabilities = {"roots": {"listChanged": True}}
+
+    target = tmp_path / "ws"
+    target.mkdir()
+
+    def run_refresh():
+        server.refresh_roots(timeout=2.0)
+
+    t = threading.Thread(target=run_refresh)
+    t.start()
+    deadline = threading.Event()
+    for _ in range(200):
+        if any('"roots/list"' in line for line in captured_lines):
+            break
+        deadline.wait(0.01)
+    request_lines = [line for line in captured_lines
+                      if '"roots/list"' in line]
+    assert request_lines, "expected outbound roots/list"
+    request_id = json.loads(request_lines[-1])["id"]
+
+    file_uri = "file:///" + str(target).replace("\\", "/").lstrip("/")
+    server.handle_line(json.dumps({
+        "jsonrpc": "2.0", "id": request_id,
+        "result": {"roots": [{"uri": file_uri, "name": "ws"}]},
+    }))
+    t.join(timeout=2.0)
+    assert not t.is_alive()
+    assert os.path.realpath(fs_provider.root) == os.path.realpath(str(target))
+
+
 def test_default_registry_lists_core_automation_tools():
     names = {tool.name for tool in build_default_tool_registry()}
     expected = {