fix: Replace arbitrary sleeps with active server readiness checks in tests

Eliminates race conditions in HTTP server tests by replacing fixed sleep delays
with deterministic server readiness polling. This makes tests more reliable and
often faster by eliminating unnecessary waits.

Changes:
- Created shared wait_for_server() helper in tests/test_helpers.py
- Fixed flaky time.sleep(1) in test_streamable_http_security.py
- Replaced duplicate wait_for_server() in test_sse_security.py with shared helper
- Improved consistency by updating all HTTP server test fixtures in:
  - tests/shared/test_ws.py
  - tests/shared/test_sse.py
  - tests/shared/test_streamable_http.py

The wait_for_server() helper:
- Actively polls the server port by attempting TCP connections
- Returns immediately when the server accepts connections
- Uses 0.01-second intervals between attempts
- Has a 5-second timeout that raises TimeoutError on true failures
- Makes tests deterministic instead of probabilistic

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: maxisbey

pyproject.toml

@@ -98,7 +98,7 @@ venv = ".venv"
 # those private functions instead of testing the private functions directly. It makes it easier to maintain the code source
 # and refactor code that is not public.
 executionEnvironments = [
-    { root = "tests", reportUnusedFunction = false, reportPrivateUsage = false },
+    { root = "tests", extraPaths = ["."], reportUnusedFunction = false, reportPrivateUsage = false },
     { root = "examples/servers", reportUnusedFunction = false },
 ]
 

tests/server/test_sse_security.py

@@ -3,7 +3,6 @@
 import logging
 import multiprocessing
 import socket
-import time
 
 import httpx
 import pytest
@@ -17,6 +16,7 @@
 from mcp.server.sse import SseServerTransport
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.types import Tool
+from tests.test_helpers import wait_for_server
 
 logger = logging.getLogger(__name__)
 SERVER_NAME = "test_sse_security_server"
@@ -66,26 +66,6 @@ async def handle_sse(request: Request):
     uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error")
 
 
-def wait_for_server(port: int, timeout: float = 5.0) -> None:
-    """Wait for server to be ready to accept connections.
-
-    Polls the server port until it accepts connections or timeout is reached.
-    This eliminates race conditions without arbitrary sleeps.
-    """
-    start_time = time.time()
-    while time.time() - start_time < timeout:
-        try:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-                s.settimeout(0.1)
-                s.connect(("127.0.0.1", port))
-                # Server is ready
-                return
-        except (ConnectionRefusedError, OSError):
-            # Server not ready yet, retry quickly
-            time.sleep(0.01)
-    raise TimeoutError(f"Server on port {port} did not start within {timeout} seconds")
-
-
 def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None):
     """Start server in a separate process."""
     process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings))

tests/server/test_streamable_http_security.py

@@ -3,7 +3,6 @@
 import logging
 import multiprocessing
 import socket
-import time
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
@@ -18,6 +17,7 @@
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.types import Tool
+from tests.test_helpers import wait_for_server
 
 logger = logging.getLogger(__name__)
 SERVER_NAME = "test_streamable_http_security_server"
@@ -77,8 +77,8 @@ def start_server_process(port: int, security_settings: TransportSecuritySettings
     """Start server in a separate process."""
     process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings))
     process.start()
-    # Give server time to start
-    time.sleep(1)
+    # Wait for server to be ready to accept connections
+    wait_for_server(port)
     return process
 
 

tests/shared/test_sse.py

@@ -32,6 +32,7 @@
     TextResourceContents,
     Tool,
 )
+from tests.test_helpers import wait_for_server
 
 SERVER_NAME = "test_server_for_SSE"
 
@@ -123,19 +124,8 @@ def server(server_port: int) -> Generator[None, None, None]:
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 20
-    attempt = 0
     print("waiting for server to start")
-    while attempt < max_attempts:
-        try:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-                s.connect(("127.0.0.1", server_port))
-                break
-        except ConnectionRefusedError:
-            time.sleep(0.1)
-            attempt += 1
-    else:
-        raise RuntimeError(f"Server failed to start after {max_attempts} attempts")
+    wait_for_server(server_port)
 
     yield
 

tests/shared/test_streamable_http.py

@@ -7,7 +7,6 @@
 import json
 import multiprocessing
 import socket
-import time
 from collections.abc import Generator
 from typing import Any
 
@@ -43,6 +42,7 @@
 from mcp.shared.message import ClientMessageMetadata
 from mcp.shared.session import RequestResponder
 from mcp.types import InitializeResult, TextContent, TextResourceContents, Tool
+from tests.test_helpers import wait_for_server
 
 # Test constants
 SERVER_NAME = "test_streamable_http_server"
@@ -344,18 +344,7 @@ def basic_server(basic_server_port: int) -> Generator[None, None, None]:
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 20
-    attempt = 0
-    while attempt < max_attempts:
-        try:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-                s.connect(("127.0.0.1", basic_server_port))
-                break
-        except ConnectionRefusedError:
-            time.sleep(0.1)
-            attempt += 1
-    else:
-        raise RuntimeError(f"Server failed to start after {max_attempts} attempts")
+    wait_for_server(basic_server_port)
 
     yield
 
@@ -391,18 +380,7 @@ def event_server(
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 20
-    attempt = 0
-    while attempt < max_attempts:
-        try:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-                s.connect(("127.0.0.1", event_server_port))
-                break
-        except ConnectionRefusedError:
-            time.sleep(0.1)
-            attempt += 1
-    else:
-        raise RuntimeError(f"Server failed to start after {max_attempts} attempts")
+    wait_for_server(event_server_port)
 
     yield event_store, f"http://127.0.0.1:{event_server_port}"
 
@@ -422,18 +400,7 @@ def json_response_server(json_server_port: int) -> Generator[None, None, None]:
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 20
-    attempt = 0
-    while attempt < max_attempts:
-        try:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-                s.connect(("127.0.0.1", json_server_port))
-                break
-        except ConnectionRefusedError:
-            time.sleep(0.1)
-            attempt += 1
-    else:
-        raise RuntimeError(f"Server failed to start after {max_attempts} attempts")
+    wait_for_server(json_server_port)
 
     yield
 
@@ -1407,18 +1374,7 @@ def context_aware_server(basic_server_port: int) -> Generator[None, None, None]:
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 20
-    attempt = 0
-    while attempt < max_attempts:
-        try:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-                s.connect(("127.0.0.1", basic_server_port))
-                break
-        except ConnectionRefusedError:
-            time.sleep(0.1)
-            attempt += 1
-    else:
-        raise RuntimeError(f"Context-aware server failed to start after {max_attempts} attempts")
+    wait_for_server(basic_server_port)
 
     yield
 

tests/shared/test_ws.py

@@ -26,6 +26,7 @@
     TextResourceContents,
     Tool,
 )
+from tests.test_helpers import wait_for_server
 
 SERVER_NAME = "test_server_for_WS"
 
@@ -110,19 +111,8 @@ def server(server_port: int) -> Generator[None, None, None]:
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 20
-    attempt = 0
     print("waiting for server to start")
-    while attempt < max_attempts:
-        try:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-                s.connect(("127.0.0.1", server_port))
-                break
-        except ConnectionRefusedError:
-            time.sleep(0.1)
-            attempt += 1
-    else:
-        raise RuntimeError(f"Server failed to start after {max_attempts} attempts")
+    wait_for_server(server_port)
 
     yield
 

tests/test_helpers.py

@@ -0,0 +1,31 @@
+"""Common test utilities for MCP server tests."""
+
+import socket
+import time
+
+
+def wait_for_server(port: int, timeout: float = 5.0) -> None:
+    """Wait for server to be ready to accept connections.
+
+    Polls the server port until it accepts connections or timeout is reached.
+    This eliminates race conditions without arbitrary sleeps.
+
+    Args:
+        port: The port number to check
+        timeout: Maximum time to wait in seconds (default 5.0)
+
+    Raises:
+        TimeoutError: If server doesn't start within the timeout period
+    """
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        try:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                s.settimeout(0.1)
+                s.connect(("127.0.0.1", port))
+                # Server is ready
+                return
+        except (ConnectionRefusedError, OSError):
+            # Server not ready yet, retry quickly
+            time.sleep(0.01)
+    raise TimeoutError(f"Server on port {port} did not start within {timeout} seconds")