Allow pre-bound HTTP sockets (#4222)

fastmcp_slim/fastmcp/server/mixins/transport.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import socket
 from collections.abc import Awaitable, Callable
 from functools import partial
 from typing import TYPE_CHECKING, Any, Literal
@@ -236,6 +237,7 @@ class TransportMixin:
         json_response: bool | None = None,
         stateless_http: bool | None = None,
         stateless: bool | None = None,
+        sockets: list[socket.socket] | None = None,
     ) -> None:
         """Run the server using HTTP transport.
 
@@ -250,6 +252,7 @@ class TransportMixin:
             json_response: Whether to use JSON response format (defaults to settings.json_response)
             stateless_http: Whether to use stateless HTTP (defaults to settings.stateless_http)
             stateless: Alias for stateless_http for CLI consistency
+            sockets: Pre-bound sockets to pass to Uvicorn
         """
         # Allow stateless as alias for stateless_http
         if stateless is not None and stateless_http is None:
@@ -302,7 +305,10 @@ class TransportMixin:
                     f"Starting MCP server {self.name!r} with transport {transport!r}{mode} on http://{host}:{port}/{path}"
                 )
 
-                await server.serve()
+                if sockets is not None:
+                    await server.serve(sockets=sockets)
+                else:
+                    await server.serve()
 
     def http_app(
         self: FastMCP,

tests/server/test_log_level.py

@@ -1,6 +1,7 @@
 """Test log_level parameter support in FastMCP server."""
 
 import asyncio
+import socket
 from unittest.mock import AsyncMock, patch
 
 from fastmcp import FastMCP
@@ -50,6 +51,26 @@ class TestLogLevelParameter:
             # Verify serve was called
             mock_instance.serve.assert_called_once()
 
+    async def test_run_http_passes_sockets_to_uvicorn(self):
+        """Test that run_http_async forwards pre-bound sockets to Uvicorn."""
+        server = FastMCP("TestServer")
+
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            with patch(
+                "fastmcp.server.mixins.transport.uvicorn.Server"
+            ) as mock_server_class:
+                mock_instance = mock_server_class.return_value
+                mock_instance.serve = AsyncMock()
+
+                await server.run_http_async(
+                    show_banner=False,
+                    host="127.0.0.1",
+                    port=8000,
+                    sockets=[sock],
+                )
+
+                mock_instance.serve.assert_called_once_with(sockets=[sock])
+
     async def test_run_async_passes_log_level(self):
         """Test that run_async passes log_level to transport methods."""
         server = FastMCP("TestServer")