fix: eliminate test port allocation race by running uvicorn in-thread

The previous pattern picked a free port via socket.bind(0), released it,
then started a uvicorn subprocess hoping to rebind — a TOCTOU race that
caused intermittent CI failures when pytest-xdist workers stole the port
between release and rebind (connection errors, 404s against wrong server,
WS 403s).

Replaced with a context manager that runs uvicorn in a background thread
with port=0 and reads the actual bound port back from the server's socket
after startup. The OS atomically assigns the port at bind time and the
server holds it until shutdown — no race window.

Also fixed a latent stream leak in SseServerTransport.connect_sse() that
the in-process server surfaced: sse_stream_reader was never closed on
normal completion or cancellation (sse_starlette only closes it on
SendTimeout). Cleanup now runs in a finally block.

Side benefits: no more subprocess spawn overhead, faster startup, shared
process state when needed, and wait_for_server() is no longer used.

Author: maxisbey

src/mcp/server/sse.py

@@ -181,13 +181,19 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
                 In this case we close our side of the streams to signal the client that
                 the connection has been closed.
                 """
-                await EventSourceResponse(content=sse_stream_reader, data_sender_callable=sse_writer)(
-                    scope, receive, send
-                )
-                await read_stream_writer.aclose()
-                await write_stream_reader.aclose()
-                self._read_stream_writers.pop(session_id, None)
-                logging.debug(f"Client session disconnected {session_id}")
+                try:
+                    await EventSourceResponse(content=sse_stream_reader, data_sender_callable=sse_writer)(
+                        scope, receive, send
+                    )
+                finally:
+                    # EventSourceResponse does not close its body iterator on
+                    # normal completion or cancellation (only on SendTimeout),
+                    # so we must close it here to avoid leaking the stream.
+                    await sse_stream_reader.aclose()
+                    await read_stream_writer.aclose()
+                    await write_stream_reader.aclose()
+                    self._read_stream_writers.pop(session_id, None)
+                    logging.debug(f"Client session disconnected {session_id}")
 
             logger.debug("Starting SSE response task")
             tg.start_soon(response_wrapper, scope, receive, send)

tests/client/test_http_unicode.py

@@ -4,8 +4,6 @@
 (server→client and client→server) using the streamable HTTP transport.
 """
 
-import multiprocessing
-import socket
 from collections.abc import AsyncGenerator, Generator
 from contextlib import asynccontextmanager
 
@@ -19,7 +17,7 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.types import TextContent, Tool
-from tests.test_helpers import wait_for_server
+from tests.test_helpers import run_uvicorn_in_thread
 
 # Test constants with various Unicode characters
 UNICODE_TEST_STRINGS = {
@@ -41,11 +39,9 @@
 }
 
 
-def run_unicode_server(port: int) -> None:  # pragma: no cover
-    """Run the Unicode test server in a separate process."""
-    import uvicorn
+def make_unicode_server_app() -> Starlette:  # pragma: no cover
+    """Create the Unicode test server app."""
 
-    # Need to recreate the server setup in this process
     async def handle_list_tools(
         ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
     ) -> types.ListToolsResult:
@@ -129,51 +125,20 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
             yield
 
     # Create an ASGI application
-    app = Starlette(
+    return Starlette(
         debug=True,
         routes=[
             Mount("/mcp", app=session_manager.handle_request),
         ],
         lifespan=lifespan,
     )
 
-    # Run the server
-    config = uvicorn.Config(
-        app=app,
-        host="127.0.0.1",
-        port=port,
-        log_level="error",
-    )
-    uvicorn_server = uvicorn.Server(config)
-    uvicorn_server.run()
-
-
-@pytest.fixture
-def unicode_server_port() -> int:
-    """Find an available port for the Unicode test server."""
-    with socket.socket() as s:
-        s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
-
 
 @pytest.fixture
-def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]:
-    """Start a Unicode test server in a separate process."""
-    proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True)
-    proc.start()
-
-    # Wait for server to be ready
-    wait_for_server(unicode_server_port)
-
-    try:
-        yield f"http://127.0.0.1:{unicode_server_port}"
-    finally:
-        # Clean up - try graceful termination first
-        proc.terminate()
-        proc.join(timeout=2)
-        if proc.is_alive():  # pragma: no cover
-            proc.kill()
-            proc.join(timeout=1)
+def running_unicode_server() -> Generator[str, None, None]:
+    """Start a Unicode test server in a background thread."""
+    with run_uvicorn_in_thread(make_unicode_server_app()) as url:
+        yield url
 
 
 @pytest.mark.anyio

tests/server/mcpserver/test_integration.py

@@ -10,13 +10,11 @@
 # pyright: reportUnknownArgumentType=false
 
 import json
-import multiprocessing
-import socket
 from collections.abc import Generator
 
 import pytest
-import uvicorn
 from inline_snapshot import snapshot
+from starlette.applications import Starlette
 
 from examples.snippets.servers import (
     basic_prompt,
@@ -58,7 +56,7 @@
     TextResourceContents,
     ToolListChangedNotification,
 )
-from tests.test_helpers import wait_for_server
+from tests.test_helpers import run_uvicorn_in_thread
 
 
 class NotificationCollector:
@@ -85,23 +83,8 @@ async def handle_generic_notification(
                 self.tool_notifications.append(message.params)
 
 
-# Common fixtures
-@pytest.fixture
-def server_port() -> int:
-    """Get a free port for testing."""
-    with socket.socket() as s:
-        s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
-
-
-@pytest.fixture
-def server_url(server_port: int) -> str:
-    """Get the server URL for testing."""
-    return f"http://127.0.0.1:{server_port}"
-
-
-def run_server_with_transport(module_name: str, port: int, transport: str) -> None:  # pragma: no cover
-    """Run server with specified transport."""
+def make_transport_app(module_name: str, transport: str) -> Starlette:  # pragma: no cover
+    """Create server app for the specified example module and transport."""
     # Get the MCP instance based on module name
     if module_name == "basic_tool":
         mcp = basic_tool.mcp
@@ -128,46 +111,27 @@ def run_server_with_transport(module_name: str, port: int, transport: str) -> No
 
     # Create app based on transport type
     if transport == "sse":
-        app = mcp.sse_app()
+        return mcp.sse_app()
     elif transport == "streamable-http":
-        app = mcp.streamable_http_app()
+        return mcp.streamable_http_app()
     else:
         raise ValueError(f"Invalid transport for test server: {transport}")
 
-    server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=port, log_level="error"))
-    print(f"Starting {transport} server on port {port}")
-    server.run()
-
 
 @pytest.fixture
-def server_transport(request: pytest.FixtureRequest, server_port: int) -> Generator[str, None, None]:
-    """Start server in a separate process with specified MCP instance and transport.
+def server_transport(request: pytest.FixtureRequest) -> Generator[tuple[str, str], None, None]:
+    """Start server in a background thread with specified MCP instance and transport.
 
     Args:
         request: pytest request with param tuple of (module_name, transport)
-        server_port: Port to run the server on
 
     Yields:
-        str: The transport type ('sse' or 'streamable_http')
+        tuple[str, str]: The (transport, url) pair.
     """
     module_name, transport = request.param
 
-    proc = multiprocessing.Process(
-        target=run_server_with_transport,
-        args=(module_name, server_port, transport),
-        daemon=True,
-    )
-    proc.start()
-
-    # Wait for server to be ready
-    wait_for_server(server_port)
-
-    yield transport
-
-    proc.kill()
-    proc.join(timeout=2)
-    if proc.is_alive():  # pragma: no cover
-        print("Server process failed to terminate")
+    with run_uvicorn_in_thread(make_transport_app(module_name, transport)) as url:
+        yield transport, url
 
 
 # Helper function to create client based on transport
@@ -220,9 +184,9 @@ async def elicitation_callback(context: RequestContext[ClientSession], params: E
     ],
     indirect=True,
 )
-async def test_basic_tools(server_transport: str, server_url: str) -> None:
+async def test_basic_tools(server_transport: tuple[str, str]) -> None:
     """Test basic tool functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):
@@ -256,9 +220,9 @@ async def test_basic_tools(server_transport: str, server_url: str) -> None:
     ],
     indirect=True,
 )
-async def test_basic_resources(server_transport: str, server_url: str) -> None:
+async def test_basic_resources(server_transport: tuple[str, str]) -> None:
     """Test basic resource functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):
@@ -296,9 +260,9 @@ async def test_basic_resources(server_transport: str, server_url: str) -> None:
     ],
     indirect=True,
 )
-async def test_basic_prompts(server_transport: str, server_url: str) -> None:
+async def test_basic_prompts(server_transport: tuple[str, str]) -> None:
     """Test basic prompt functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):
@@ -348,9 +312,9 @@ async def test_basic_prompts(server_transport: str, server_url: str) -> None:
     ],
     indirect=True,
 )
-async def test_tool_progress(server_transport: str, server_url: str) -> None:
+async def test_tool_progress(server_transport: tuple[str, str]) -> None:
     """Test tool progress reporting."""
-    transport = server_transport
+    transport, server_url = server_transport
     collector = NotificationCollector()
 
     async def message_handler(message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception):
@@ -404,9 +368,9 @@ async def progress_callback(progress: float, total: float | None, message: str |
     ],
     indirect=True,
 )
-async def test_sampling(server_transport: str, server_url: str) -> None:
+async def test_sampling(server_transport: tuple[str, str]) -> None:
     """Test sampling (LLM interaction) functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):
@@ -434,9 +398,9 @@ async def test_sampling(server_transport: str, server_url: str) -> None:
     ],
     indirect=True,
 )
-async def test_elicitation(server_transport: str, server_url: str) -> None:
+async def test_elicitation(server_transport: tuple[str, str]) -> None:
     """Test elicitation (user interaction) functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):
@@ -483,9 +447,9 @@ async def test_elicitation(server_transport: str, server_url: str) -> None:
     ],
     indirect=True,
 )
-async def test_notifications(server_transport: str, server_url: str) -> None:
+async def test_notifications(server_transport: tuple[str, str]) -> None:
     """Test notifications and logging functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     collector = NotificationCollector()
 
     async def message_handler(message: RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception):
@@ -530,9 +494,9 @@ async def message_handler(message: RequestResponder[ServerRequest, ClientResult]
     ],
     indirect=True,
 )
-async def test_completion(server_transport: str, server_url: str) -> None:
+async def test_completion(server_transport: tuple[str, str]) -> None:
     """Test completion (autocomplete) functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):
@@ -582,9 +546,9 @@ async def test_completion(server_transport: str, server_url: str) -> None:
     ],
     indirect=True,
 )
-async def test_mcpserver_quickstart(server_transport: str, server_url: str) -> None:
+async def test_mcpserver_quickstart(server_transport: tuple[str, str]) -> None:
     """Test MCPServer quickstart example."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):
@@ -617,9 +581,9 @@ async def test_mcpserver_quickstart(server_transport: str, server_url: str) -> N
     ],
     indirect=True,
 )
-async def test_structured_output(server_transport: str, server_url: str) -> None:
+async def test_structured_output(server_transport: tuple[str, str]) -> None:
     """Test structured output functionality."""
-    transport = server_transport
+    transport, server_url = server_transport
     client_cm = create_client_for_transport(transport, server_url)
 
     async with client_cm as (read_stream, write_stream):