fix(mcp): enforce trusted workspace roots

mcp_servers/multi_mcp.py

@@ -104,6 +104,20 @@ def reset_trace_context(self, token):
         if token is not None:
             self._trace_context.reset(token)
 
+    def _prepare_tool_arguments(self, tool_name: str, arguments: dict | None) -> dict:
+        """Apply trusted request context to sensitive tool arguments."""
+        prepared = dict(arguments or {})
+        if tool_name not in {"read_workspace_file", "write_workspace_file"}:
+            return prepared
+
+        trace_context = self._trace_context.get()
+        workspace = trace_context.get("workspace")
+        if workspace:
+            prepared["workspace_root"] = workspace
+        else:
+            prepared.pop("workspace_root", None)
+        return prepared
+
     def _load_config(self) -> dict:
         """Load server configuration from JSON"""
         if self.config_path.exists():
@@ -491,6 +505,7 @@ async def call_tool(self, server_name: str, tool_name: str, arguments: dict):
         """Call a tool on a specific server"""
         if server_name not in self.sessions:
             raise ValueError(f"Server '{server_name}' not connected")
+        arguments = self._prepare_tool_arguments(tool_name, arguments)
         trace_context = self._trace_context.get()
         if trace_context:
             print(
@@ -510,12 +525,7 @@ async def route_tool_call(self, tool_name: str, arguments: dict):
         workflow_id = trace_context.get("workflow_id", "generic")
         contract_version = trace_context.get("contract_version", "v1")
 
-        if trace_context.get("workspace") and tool_name in {
-            "read_workspace_file",
-            "write_workspace_file",
-        }:
-            arguments = dict(arguments or {})
-            arguments.setdefault("workspace_root", trace_context["workspace"])
+        arguments = self._prepare_tool_arguments(tool_name, arguments)
 
         # Get or create circuit breaker for this tool
         breaker = get_breaker(tool_name, failure_threshold=5, recovery_timeout=60.0)

tests/test_multi_mcp_modes.py

@@ -12,6 +12,20 @@
 from mcp_servers.multi_mcp import MultiMCP
 
 
+class _DummyTool:
+    def __init__(self, name: str):
+        self.name = name
+
+
+class _RecordingSession:
+    def __init__(self):
+        self.calls = []
+
+    async def call_tool(self, tool_name, arguments):
+        self.calls.append((tool_name, arguments))
+        return {"tool": tool_name, "arguments": arguments}
+
+
 class MultiMcpModeTests(unittest.TestCase):
     def tearDown(self):
         reload_settings()
@@ -47,6 +61,45 @@ def test_strict_mode_requires_required_servers_for_readiness(self):
         self.assertTrue(health["mcp_ready"])
         self.assertEqual(sorted(health["connected_servers"]), ["mockehr", "rag"])
 
+    def test_workspace_tools_use_trusted_trace_workspace(self):
+        mm = MultiMCP()
+        session = _RecordingSession()
+        mm.tools = {"sandbox": [_DummyTool("write_workspace_file")]}
+        mm.sessions = {"sandbox": session}
+        trace_token = mm.set_trace_context({"workspace": "/trusted/task"})
+        try:
+            result = asyncio.run(
+                mm.route_tool_call(
+                    "write_workspace_file",
+                    {
+                        "path": "answer.txt",
+                        "content": "ok",
+                        "workspace_root": "/",
+                    },
+                )
+            )
+        finally:
+            mm.reset_trace_context(trace_token)
+
+        self.assertEqual(result["arguments"]["workspace_root"], "/trusted/task")
+        self.assertEqual(session.calls[0][1]["workspace_root"], "/trusted/task")
+
+    def test_workspace_tools_strip_caller_root_without_trace_workspace(self):
+        mm = MultiMCP()
+        session = _RecordingSession()
+        mm.sessions = {"sandbox": session}
+
+        result = asyncio.run(
+            mm.call_tool(
+                "sandbox",
+                "read_workspace_file",
+                {"path": "notes.md", "workspace_root": "/"},
+            )
+        )
+
+        self.assertNotIn("workspace_root", result["arguments"])
+        self.assertNotIn("workspace_root", session.calls[0][1])
+
     def test_strict_mode_start_raises_when_required_server_missing(self):
         with patch.dict(
             "os.environ",