feat: support force flag

Author: radugheo

pyproject.toml

@@ -8,7 +8,7 @@ dependencies = [
     "mcp==1.26.0",
     "pysignalr==1.3.0",
     # "uipath>=2.8.23, <2.9.0",
-    "uipath==2.10.12.dev1014315279",
+    "uipath==2.10.12.dev1014315285",
     "uipath-runtime>=0.9.1, <0.10.0",
 ]
 classifiers = [
@@ -43,14 +43,13 @@ dev = [
     "mypy>=1.14.1",
     "ruff>=0.9.4",
     "pytest>=7.4.0",
-    "pytest-asyncio>=0.23.0",
+    "pytest-asyncio>=1.3.0",
     "pytest-cov>=4.1.0",
     "pytest-mock>=3.11.1",
     "pre-commit>=4.5.1",
     "filelock>=3.20.3",
     "virtualenv>=20.36.1",
     "numpy>=1.24.0",
-    "pytest-asyncio>=1.3.0",
 ]
 
 [tool.ruff]

src/uipath_mcp/_cli/_runtime/_factory.py

@@ -101,17 +101,35 @@ async def new_runtime(
                 UiPathErrorCategory.DEPLOYMENT,
             )
 
+        force = kwargs.get("force", False)
         server = mcp_config.get_server(entrypoint)
         logger.info("Creating MCP runtime for entrypoint '%s'", entrypoint)
         logger.info(server)
         if not server:
-            available = ", ".join(mcp_config.get_server_names())
-            raise UiPathMcpRuntimeError(
-                McpErrorCode.SERVER_NOT_FOUND,
-                "MCP server not found",
-                f"Server '{entrypoint}' not found. Available: {available}",
-                UiPathErrorCategory.USER,
-            )
+            if force:
+                server = mcp_config.get_single_server()
+                if server:
+                    logger.info(
+                        "Force mode: using server '%s' config with entrypoint '%s'",
+                        server.name,
+                        entrypoint,
+                    )
+                else:
+                    available = ", ".join(mcp_config.get_server_names())
+                    raise UiPathMcpRuntimeError(
+                        McpErrorCode.SERVER_NOT_FOUND,
+                        "MCP server not found",
+                        f"--force requires exactly one server in mcp.json, but found: {available}",
+                        UiPathErrorCategory.USER,
+                    )
+            else:
+                available = ", ".join(mcp_config.get_server_names())
+                raise UiPathMcpRuntimeError(
+                    McpErrorCode.SERVER_NOT_FOUND,
+                    "MCP server not found",
+                    f"Server '{entrypoint}' not found. Available: {available}",
+                    UiPathErrorCategory.USER,
+                )
 
         # Validate streamable-http configuration
         if server.is_streamable_http:

src/uipath_mcp/_cli/_utils/_config.py

@@ -109,6 +109,12 @@ def get_server(self, name: str) -> McpServer | None:
         """
         return self._servers.get(name)
 
+    def get_single_server(self) -> McpServer | None:
+        """Return the server if there is exactly one, None otherwise."""
+        if len(self._servers) == 1:
+            return next(iter(self._servers.values()))
+        return None
+
     def get_server_names(self) -> list[str]:
         """Get list of all server names."""
         return list(self._servers.keys())

tests/cli/test_config.py

@@ -0,0 +1,33 @@
+import json
+
+from uipath_mcp._cli._utils._config import McpConfig
+
+
+def test_get_single_server_with_one(tmp_path):
+    config_path = tmp_path / "mcp.json"
+    config_path.write_text(json.dumps({
+        "servers": {"only-server": {"command": "python", "args": ["s.py"]}}
+    }))
+    cfg = McpConfig(str(config_path))
+    server = cfg.get_single_server()
+    assert server is not None
+    assert server.name == "only-server"
+
+
+def test_get_single_server_with_multiple(tmp_path):
+    config_path = tmp_path / "mcp.json"
+    config_path.write_text(json.dumps({
+        "servers": {
+            "server-a": {"command": "python", "args": ["a.py"]},
+            "server-b": {"command": "python", "args": ["b.py"]},
+        }
+    }))
+    cfg = McpConfig(str(config_path))
+    assert cfg.get_single_server() is None
+
+
+def test_get_single_server_with_none(tmp_path):
+    config_path = tmp_path / "mcp.json"
+    config_path.write_text(json.dumps({"servers": {}}))
+    cfg = McpConfig(str(config_path))
+    assert cfg.get_single_server() is None

tests/cli/test_factory.py

@@ -0,0 +1,126 @@
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from uipath_mcp._cli._runtime._exception import UiPathMcpRuntimeError
+from uipath_mcp._cli._runtime._factory import UiPathMcpRuntimeFactory
+
+# Patch UiPath() constructor which requires auth env vars
+_UIPATH_PATCH = patch("uipath_mcp._cli._runtime._runtime.UiPath")
+
+
+@pytest.fixture
+def mcp_json_single(tmp_path):
+    """Create a temporary mcp.json with a single server."""
+    config = {
+        "servers": {
+            "math-server": {
+                "transport": "stdio",
+                "command": "python",
+                "args": ["server.py"],
+            }
+        }
+    }
+    config_path = tmp_path / "mcp.json"
+    config_path.write_text(json.dumps(config))
+    return str(config_path)
+
+
+@pytest.fixture
+def mcp_json_multiple(tmp_path):
+    """Create a temporary mcp.json with multiple servers."""
+    config = {
+        "servers": {
+            "math-server": {
+                "transport": "stdio",
+                "command": "python",
+                "args": ["math_server.py"],
+            },
+            "date-server": {
+                "transport": "stdio",
+                "command": "python",
+                "args": ["date_server.py"],
+            },
+        }
+    }
+    config_path = tmp_path / "mcp.json"
+    config_path.write_text(json.dumps(config))
+    return str(config_path)
+
+
+@pytest.fixture
+def factory(tmp_path):
+    context = MagicMock()
+    context.config_path = str(tmp_path / "uipath.json")
+    context.folder_key = "test-folder-key"
+    context.mcp_server_id = "test-server-id"
+    return UiPathMcpRuntimeFactory(context=context)
+
+
+@pytest.mark.asyncio
+async def test_exact_match_works(factory, mcp_json_single):
+    """Server found by exact name match."""
+    factory._mcp_config = None
+    with _UIPATH_PATCH, patch.object(factory, "_load_mcp_config") as mock_load:
+        from uipath_mcp._cli._utils._config import McpConfig
+
+        mock_load.return_value = McpConfig(mcp_json_single)
+        runtime = await factory.new_runtime("math-server", "00000000-0000-0000-0000-000000000001")
+    assert runtime._entrypoint == "math-server"
+
+
+@pytest.mark.asyncio
+async def test_wrong_name_without_force_raises(factory, mcp_json_single):
+    """Wrong entrypoint without --force raises SERVER_NOT_FOUND."""
+    with patch.object(factory, "_load_mcp_config") as mock_load:
+        from uipath_mcp._cli._utils._config import McpConfig
+
+        mock_load.return_value = McpConfig(mcp_json_single)
+        with pytest.raises(UiPathMcpRuntimeError) as exc_info:
+            await factory.new_runtime("my-mcp2", "00000000-0000-0000-0000-000000000001")
+    assert "not found" in str(exc_info.value).lower()
+
+
+@pytest.mark.asyncio
+async def test_force_single_server_uses_config_with_new_entrypoint(factory, mcp_json_single):
+    """--force with single server loads its config but uses the requested entrypoint."""
+    with _UIPATH_PATCH, patch.object(factory, "_load_mcp_config") as mock_load:
+        from uipath_mcp._cli._utils._config import McpConfig
+
+        mock_load.return_value = McpConfig(mcp_json_single)
+        runtime = await factory.new_runtime(
+            "my-mcp2", "00000000-0000-0000-0000-000000000001", force=True
+        )
+    # Entrypoint is the requested name, not the original server name
+    assert runtime._entrypoint == "my-mcp2"
+    # Server config comes from the existing server
+    assert runtime._server.command == "python"
+    assert runtime._server.args == ["server.py"]
+
+
+@pytest.mark.asyncio
+async def test_force_multiple_servers_raises(factory, mcp_json_multiple):
+    """--force with multiple servers raises an error."""
+    with patch.object(factory, "_load_mcp_config") as mock_load:
+        from uipath_mcp._cli._utils._config import McpConfig
+
+        mock_load.return_value = McpConfig(mcp_json_multiple)
+        with pytest.raises(UiPathMcpRuntimeError) as exc_info:
+            await factory.new_runtime(
+                "my-mcp2", "00000000-0000-0000-0000-000000000001", force=True
+            )
+    assert "--force requires exactly one server" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_force_with_exact_match_still_works(factory, mcp_json_single):
+    """--force with a correct entrypoint name works normally."""
+    with _UIPATH_PATCH, patch.object(factory, "_load_mcp_config") as mock_load:
+        from uipath_mcp._cli._utils._config import McpConfig
+
+        mock_load.return_value = McpConfig(mcp_json_single)
+        runtime = await factory.new_runtime(
+            "math-server", "00000000-0000-0000-0000-000000000001", force=True
+        )
+    assert runtime._entrypoint == "math-server"