Merge pull request 'Fix: could not handle mcp name that includes '/'' (#457) from mcp-name-with-slash into development

Reviewed-on: https://git.biggo.com/Funmula/dive-mcp-host/pulls/457

Author: LouisFunmula

dive_mcp_host/host/tools/log.py

@@ -276,17 +276,27 @@ def __init__(self, name: str, log_dir: Path, rotation_files: int = 5) -> None:
         self._logger = getLogger(self._name)
         self._logger.setLevel(INFO)
         self._logger.propagate = False
-        handler = TimedRotatingFileHandler(
-            self._path,
-            when="D",
-            interval=1,
-            backupCount=rotation_files,
-            encoding="utf-8",
-        )
-        self._logger.addHandler(handler)
+        self._error: Exception | None = None
+        try:
+            handler = TimedRotatingFileHandler(
+                self._path,
+                when="D",
+                interval=1,
+                backupCount=rotation_files,
+                encoding="utf-8",
+            )
+            self._logger.addHandler(handler)
+        except FileNotFoundError as e:
+            logger.exception(
+                "Create TimedRotatingFileHandler for %s failed, "
+                "subsquent calls will be ignored",
+                self._name,
+            )
+            self._error = e
 
     async def __call__(self, log: LogMsg) -> None:
-        self._logger.info(log.model_dump_json())
+        if not self._error:
+            self._logger.info(log.model_dump_json())
 
 
 class LogManager:

dive_mcp_host/httpd/routers/tools.py

@@ -124,7 +124,7 @@ async def list_tools(  # noqa: PLR0912, C901
     return ToolsResult(success=True, message=None, tools=list(result.values()))
 
 
-@tools.get("/{server_name}/logs/stream")
+@tools.get("/logs/stream")
 async def stream_server_logs(
     server_name: str,
     stream_until: ClientState | None = None,

tests/conftest.py

@@ -136,6 +136,45 @@ async def echo_tool_streamable_server(
         await proc.wait()
 
 
+@pytest_asyncio.fixture
+@asynccontextmanager
+async def echo_with_slash_tool_streamable_server(
+    unused_tcp_port_factory: Callable[[], int],
+) -> AsyncGenerator[tuple[int, dict[str, ServerConfig]], None]:
+    """Start the echo tool SSE server."""
+    port = unused_tcp_port_factory()
+    proc = await asyncio.create_subprocess_exec(
+        "python3",
+        "-m",
+        "dive_mcp_host.host.tools.echo",
+        "--transport=streamable",
+        "--host=localhost",
+        f"--port={port}",
+    )
+    while True:
+        try:
+            _ = await httpx.AsyncClient().get(f"http://localhost:{port}/xxxx")
+            break
+        except httpx.HTTPStatusError:
+            break
+        except:  # noqa: E722
+            await asyncio.sleep(0.1)
+    try:
+        yield (
+            port,
+            {
+                "echo/aaa/bbb/ccc": ServerConfig(
+                    name="echo/aaa/bbb/ccc",
+                    url=f"http://localhost:{port}/mcp",
+                    transport="streamable",
+                )
+            },
+        )
+    finally:
+        proc.send_signal(signal.SIGKILL)
+        await proc.wait()
+
+
 @pytest.fixture
 def log_config() -> LogConfig:
     """Fixture for log Config."""

tests/httpd/routers/test_tools.py

@@ -373,7 +373,9 @@ def test_tools_cache_after_update(test_client):
 def test_stream_logs_notfound(test_client: tuple[TestClient, DiveHostAPI]):
     """Test stream_logs function with not found server."""
     client, _ = test_client
-    response = client.get("/api/tools/missing_server/logs/stream")
+    response = client.get(
+        "/api/tools/logs/stream", params={"server_name": "missing_server"}
+    )
     for line in response.iter_lines():
         content = line.removeprefix("data: ")
         if content in ("[DONE]", ""):
@@ -413,8 +415,63 @@ def update_tools():
     with ThreadPoolExecutor(1) as executor:
         executor.submit(update_tools)
         response = client.get(
-            "/api/tools/missing_server/logs/stream",
+            "/api/tools/logs/stream",
+            params={
+                "server_name": "missing_server",
+                "stop_on_notfound": False,
+                "max_retries": 5,
+                "stream_until": "running",
+            },
+        )
+        responses: list[LogMsg] = []
+        for line in response.iter_lines():
+            content = line.removeprefix("data: ")
+            if content in ("[DONE]", ""):
+                continue
+
+            data = LogMsg.model_validate_json(content)
+            responses.append(data)
+
+        assert len(responses) >= 3
+        assert responses[-3].event == LogEvent.STREAMING_ERROR
+
+        assert responses[-2].event == LogEvent.STDERR
+        assert responses[-2].client_state == ClientState.INIT
+
+        assert responses[-1].event == LogEvent.STATUS_CHANGE
+        assert responses[-1].client_state == ClientState.RUNNING
+
+
+def test_stream_logs_name_with_slash(test_client: tuple[TestClient, DiveHostAPI]):
+    """Test stream_logs before log buffer is registered."""
+    client, _ = test_client
+
+    def update_tools():
+        _ = client.post(
+            "/api/config/mcpserver",
+            json={
+                "mcpServers": {
+                    "name/with/slash": {
+                        "transport": "stdio",
+                        "enabled": True,
+                        "command": "python",
+                        "args": [
+                            "-m",
+                            "dive_mcp_host.host.tools.echo",
+                            "--transport=stdio",
+                        ],
+                    }
+                }
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+    with ThreadPoolExecutor(1) as executor:
+        executor.submit(update_tools)
+        response = client.get(
+            "/api/tools/logs/stream",
             params={
+                "server_name": "name/with/slash",
                 "stop_on_notfound": False,
                 "max_retries": 5,
                 "stream_until": "running",

tests/test_tools.py

@@ -130,6 +130,37 @@ async def test_tool_manager_streamable(
                 assert json.loads(str(result.content)) == []
 
 
+@pytest.mark.asyncio
+async def test_tool_manager_tool_name_with_slash(
+    echo_with_slash_tool_streamable_server: AbstractAsyncContextManager[
+        tuple[int, dict[str, ServerConfig]]
+    ],
+    log_config: LogConfig,
+) -> None:
+    """Test the tool manager."""
+    async with (
+        echo_with_slash_tool_streamable_server as (port, configs),
+        ToolManager(configs, log_config) as tool_manager,
+    ):
+        await tool_manager.initialized_event.wait()
+        tools = tool_manager.langchain_tools()
+        assert sorted([i.name for i in tools]) == ["echo", "ignore"]
+        for tool in tools:
+            result = await tool.ainvoke(
+                ToolCall(
+                    name=tool.name,
+                    id="123",
+                    args={"message": "Hello, world!"},
+                    type="tool_call",
+                ),
+            )
+            assert isinstance(result, ToolMessage)
+            if tool.name == "echo":
+                assert json.loads(str(result.content))[0]["text"] == "Hello, world!"
+            else:
+                assert json.loads(str(result.content)) == []
+
+
 @pytest.mark.asyncio
 async def test_tool_manager_reload(
     echo_tool_stdio_config: dict[str, ServerConfig],