diff --git a/mcp_servers/multi_mcp.py b/mcp_servers/multi_mcp.py index 362a005..d8a4a0b 100644 --- a/mcp_servers/multi_mcp.py +++ b/mcp_servers/multi_mcp.py @@ -515,7 +515,7 @@ async def route_tool_call(self, tool_name: str, arguments: dict): "write_workspace_file", }: arguments = dict(arguments or {}) - arguments.setdefault("workspace_root", trace_context["workspace"]) + arguments["workspace_root"] = trace_context["workspace"] # Get or create circuit breaker for this tool breaker = get_breaker(tool_name, failure_threshold=5, recovery_timeout=60.0) diff --git a/tests/test_mockehr_mcp.py b/tests/test_mockehr_mcp.py index abeb40b..6d05e0b 100644 --- a/tests/test_mockehr_mcp.py +++ b/tests/test_mockehr_mcp.py @@ -186,3 +186,33 @@ async def _fake_call_tool(server_name, tool_name, arguments): assert called["server"] == "mockehr" assert result["ok"] is True + +def test_route_tool_call_forces_trusted_workspace_root(): + mm = MultiMCP() + mm.tools = {"sandbox": [_DummyTool("write_workspace_file")]} + mm.sessions = {"sandbox": object()} + called = {"arguments": None} + + async def _fake_call_tool(server_name, tool_name, arguments): + called["arguments"] = arguments + return {"ok": True} + + mm.call_tool = _fake_call_tool # type: ignore[assignment] + token = mm.set_trace_context({"workspace": "/trusted/workspace"}) + try: + result = asyncio.run( + mm.route_tool_call( + "write_workspace_file", + { + "workspace_root": "/workspace", + "path": "config/settings.json", + "content": "{}", + }, + ) + ) + finally: + mm.reset_trace_context(token) + + assert result["ok"] is True + assert called["arguments"]["workspace_root"] == "/trusted/workspace" +