fix tests binding ephemeral port and skiping if unable to reserve a local port

Author: grll

tests/__init__.py

@@ -0,0 +1 @@
+

tests/_server_utils.py

@@ -0,0 +1,57 @@
+import os
+import socket
+import subprocess
+import time
+from typing import Tuple
+
+
+def _reserve_port() -> int:
+    """Reserve an available localhost TCP port."""
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.bind(("127.0.0.1", 0))
+            # Allow the port to be reused immediately after closing
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            return sock.getsockname()[1]
+    except OSError as exc:
+        raise RuntimeError(f"Unable to reserve local TCP port: {exc}") from exc
+
+
+def launch_mcp_server(script: str) -> Tuple[subprocess.Popen[str], int]:
+    """Launch an MCP server subprocess and return the process and bound port.
+
+    Raises:
+        RuntimeError: If the subprocess exits before becoming ready.
+    """
+    port = _reserve_port()
+    env = {**os.environ, "MCP_TEST_PORT": str(port)}
+
+    process = subprocess.Popen(  # noqa: S603
+        ["python", "-c", script],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        env=env,
+    )
+
+    # Give the server a moment to bind and report startup issues
+    time.sleep(1)
+
+    if process.poll() is not None:
+        stdout, stderr = process.communicate()
+        raise RuntimeError(
+            "Failed to start MCP server subprocess. "
+            f"Exit code: {process.returncode}\nStdout:\n{stdout}\nStderr:\n{stderr}"
+        )
+
+    return process, port
+
+
+def terminate_mcp_server(process: subprocess.Popen[str]) -> Tuple[str, str]:
+    """Terminate the subprocess and return captured stdout/stderr."""
+    if process.poll() is None:
+        process.kill()
+
+    stdout, stderr = process.communicate()
+
+    return stdout, stderr

tests/test_core.py

@@ -7,6 +7,7 @@
 from mcp import StdioServerParameters
 
 from mcpadapt.core import MCPAdapt, ToolAdapter
+from tests._server_utils import launch_mcp_server, terminate_mcp_server
 
 
 class DummyAdapter(ToolAdapter):
@@ -81,9 +82,12 @@ def echo_tool(text: str) -> str:
 def echo_server_sse_script():
     return dedent(
         '''
+        import os
         from mcp.server.fastmcp import FastMCP
 
-        mcp = FastMCP("Echo Server", host="127.0.0.1", port=8000)
+        port = int(os.environ.get("MCP_TEST_PORT", "8000"))
+
+        mcp = FastMCP("Echo Server", host="127.0.0.1", port=port)
 
         @mcp.tool()
         def echo_tool(text: str) -> str:
@@ -97,31 +101,27 @@ def echo_tool(text: str) -> str:
 
 @pytest.fixture
 async def echo_sse_server(echo_server_sse_script):
-    import subprocess
-
-    # Start the SSE server process with its own process group
-    process = subprocess.Popen(
-        ["python", "-c", echo_server_sse_script],
-    )
-
-    # Give the server a moment to start up
-    time.sleep(1)
+    try:
+        process, port = launch_mcp_server(echo_server_sse_script)
+    except RuntimeError as exc:
+        pytest.skip(str(exc))
 
     try:
-        yield {"url": "http://127.0.0.1:8000/sse"}
+        yield {"url": f"http://127.0.0.1:{port}/sse"}
     finally:
-        # Clean up the process when test is done
-        process.kill()
-        process.wait()
+        terminate_mcp_server(process)
 
 
 @pytest.fixture
 def echo_server_streamable_http_script():
     return dedent(
         '''
+        import os
         from mcp.server.fastmcp import FastMCP
         
-        mcp = FastMCP("Echo Server", host="127.0.0.1", port=8000, stateless_http=True, json_response=True)
+        port = int(os.environ.get("MCP_TEST_PORT", "8000"))
+
+        mcp = FastMCP("Echo Server", host="127.0.0.1", port=port, stateless_http=True, json_response=True)
         
         @mcp.tool()
         def echo_tool(text: str) -> str:
@@ -135,22 +135,15 @@ def echo_tool(text: str) -> str:
 
 @pytest.fixture
 async def echo_streamable_http_server(echo_server_streamable_http_script):
-    import subprocess
-
-    # Start the SSE server process with its own process group
-    process = subprocess.Popen(
-        ["python", "-c", echo_server_streamable_http_script],
-    )
-
-    # Give the server a moment to start up
-    time.sleep(1)
+    try:
+        process, port = launch_mcp_server(echo_server_streamable_http_script)
+    except RuntimeError as exc:
+        pytest.skip(str(exc))
 
     try:
-        yield {"url": "http://127.0.0.1:8000/mcp", "transport": "streamable-http"}
+        yield {"url": f"http://127.0.0.1:{port}/mcp", "transport": "streamable-http"}
     finally:
-        # Clean up the process when test is done
-        process.kill()
-        process.wait()
+        terminate_mcp_server(process)
 
 
 @pytest.fixture

tests/test_crewai_adapter.py

@@ -8,6 +8,7 @@
 
 from mcpadapt.core import MCPAdapt
 from mcpadapt.crewai_adapter import CrewAIAdapter
+from tests._server_utils import launch_mcp_server, terminate_mcp_server
 
 
 def extract_and_eval_dict(text):
@@ -102,9 +103,12 @@ def custom_tool(
 def echo_server_sse_script():
     return dedent(
         '''
+        import os
         from mcp.server.fastmcp import FastMCP
 
-        mcp = FastMCP("Echo Server", host="127.0.0.1", port=8000)
+        port = int(os.environ.get("MCP_TEST_PORT", "8000"))
+
+        mcp = FastMCP("Echo Server", host="127.0.0.1", port=port)
 
         @mcp.tool()
         def echo_tool(text: str) -> str:
@@ -150,23 +154,15 @@ def echo_tool_union_none(text: str | None) -> str:
 
 @pytest.fixture
 async def echo_sse_server(echo_server_sse_script):
-    import subprocess
-    import time
-
-    # Start the SSE server process with its own process group
-    process = subprocess.Popen(
-        ["python", "-c", echo_server_sse_script],
-    )
-
-    # Give the server a moment to start up
-    time.sleep(1)
+    try:
+        process, port = launch_mcp_server(echo_server_sse_script)
+    except RuntimeError as exc:
+        pytest.skip(str(exc))
 
     try:
-        yield {"url": "http://127.0.0.1:8000/sse"}
+        yield {"url": f"http://127.0.0.1:{port}/sse"}
     finally:
-        # Clean up the process when test is done
-        process.kill()
-        process.wait()
+        terminate_mcp_server(process)
 
 
 def test_basic_sync(echo_server_script):

tests/test_google_genai_adapter.py

@@ -5,6 +5,7 @@
 
 from mcpadapt.core import MCPAdapt
 from mcpadapt.google_genai_adapter import GoogleGenAIAdapter
+from tests._server_utils import launch_mcp_server, terminate_mcp_server
 
 
 @pytest.fixture
@@ -29,9 +30,12 @@ def echo_tool(text: str) -> str:
 def echo_server_sse_script():
     return dedent(
         '''
+        import os
         from mcp.server.fastmcp import FastMCP
 
-        mcp = FastMCP("Echo Server", host="127.0.0.1", port=8000)
+        port = int(os.environ.get("MCP_TEST_PORT", "8000"))
+
+        mcp = FastMCP("Echo Server", host="127.0.0.1", port=port)
 
         @mcp.tool()
         def echo_tool(text: str) -> str:
@@ -77,23 +81,15 @@ def echo_tool_union_none(text: str | None) -> str:
 
 @pytest.fixture
 async def echo_sse_server(echo_server_sse_script):
-    import subprocess
-    import time
-
-    # Start the SSE server process with its own process group
-    process = subprocess.Popen(
-        ["python", "-c", echo_server_sse_script],
-    )
-
-    # Give the server a moment to start up
-    time.sleep(1)
+    try:
+        process, port = launch_mcp_server(echo_server_sse_script)
+    except RuntimeError as exc:
+        pytest.skip(str(exc))
 
     try:
-        yield {"url": "http://127.0.0.1:8000/sse"}
+        yield {"url": f"http://127.0.0.1:{port}/sse"}
     finally:
-        # Clean up the process when test is done
-        process.kill()
-        process.wait()
+        terminate_mcp_server(process)
 
 
 def test_basic_sync(echo_server_script):

tests/test_langchain_adapter.py

@@ -5,6 +5,7 @@
 
 from mcpadapt.core import MCPAdapt
 from mcpadapt.langchain_adapter import LangChainAdapter
+from tests._server_utils import launch_mcp_server, terminate_mcp_server
 
 
 @pytest.fixture
@@ -55,9 +56,12 @@ def echo_tool(text: str) -> str:
 def echo_server_sse_script():
     return dedent(
         '''
+        import os
         from mcp.server.fastmcp import FastMCP
 
-        mcp = FastMCP("Echo Server", host="127.0.0.1", port=8000)
+        port = int(os.environ.get("MCP_TEST_PORT", "8000"))
+
+        mcp = FastMCP("Echo Server", host="127.0.0.1", port=port)
 
         @mcp.tool()
         def echo_tool(text: str) -> str:
@@ -71,23 +75,15 @@ def echo_tool(text: str) -> str:
 
 @pytest.fixture
 async def echo_sse_server(echo_server_sse_script):
-    import subprocess
-    import time
-
-    # Start the SSE server process with its own process group
-    process = subprocess.Popen(
-        ["python", "-c", echo_server_sse_script],
-    )
-
-    # Give the server a moment to start up
-    time.sleep(1)
+    try:
+        process, port = launch_mcp_server(echo_server_sse_script)
+    except RuntimeError as exc:
+        pytest.skip(str(exc))
 
     try:
-        yield {"url": "http://127.0.0.1:8000/sse"}
+        yield {"url": f"http://127.0.0.1:{port}/sse"}
     finally:
-        # Clean up the process when test is done
-        process.kill()
-        process.wait()
+        terminate_mcp_server(process)
 
 
 @pytest.mark.asyncio

tests/test_smolagents_adapter.py

@@ -7,6 +7,7 @@
 
 from mcpadapt.core import MCPAdapt
 from mcpadapt.smolagents_adapter import SmolAgentsAdapter
+from tests._server_utils import launch_mcp_server, terminate_mcp_server
 
 
 @pytest.fixture
@@ -31,9 +32,12 @@ def echo_tool(text: str) -> str:
 def echo_server_sse_script():
     return dedent(
         '''
+        import os
         from mcp.server.fastmcp import FastMCP
 
-        mcp = FastMCP("Echo Server", host="127.0.0.1", port=8000)
+        port = int(os.environ.get("MCP_TEST_PORT", "8000"))
+
+        mcp = FastMCP("Echo Server", host="127.0.0.1", port=port)
 
         @mcp.tool()
         def echo_tool(text: str) -> str:
@@ -79,23 +83,15 @@ def echo_tool_union_none(text: str | None) -> str:
 
 @pytest.fixture
 async def echo_sse_server(echo_server_sse_script):
-    import subprocess
-    import time
-
-    # Start the SSE server process with its own process group
-    process = subprocess.Popen(
-        ["python", "-c", echo_server_sse_script],
-    )
-
-    # Give the server a moment to start up
-    time.sleep(1)
+    try:
+        process, port = launch_mcp_server(echo_server_sse_script)
+    except RuntimeError as exc:
+        pytest.skip(str(exc))
 
     try:
-        yield {"url": "http://127.0.0.1:8000/sse"}
+        yield {"url": f"http://127.0.0.1:{port}/sse"}
     finally:
-        # Clean up the process when test is done
-        process.kill()
-        process.wait()
+        terminate_mcp_server(process)
 
 
 def test_basic_sync(echo_server_script):