feat: add stateless_sse option for SSE transport

Add a stateless_sse parameter to FastMCP that bypasses the MCP protocol
initialization handshake for SSE transport. This mirrors the existing
stateless_http option for Streamable HTTP transport.

When stateless_sse=True, the server allows tool calls without waiting
for the InitializeRequest → InitializeResponse → InitializedNotification
handshake to complete. This fixes issues with fast clients like Claude
Code that may send requests before initialization completes.

Usage:
  mcp = FastMCP("my-server", stateless_sse=True)

Fixes #1844

Github-Issue: #1844
Reported-by: Tech-Fumi

Author: Jay Hemnani

src/mcp/server/fastmcp/server.py

@@ -107,6 +107,10 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     stateless_http: bool
     """Define if the server should create a new transport per request."""
 
+    # SSE settings
+    stateless_sse: bool
+    """Define if the SSE server should bypass MCP initialization handshake."""
+
     # resource settings
     warn_on_duplicate_resources: bool
 
@@ -169,6 +173,7 @@ def __init__(  # noqa: PLR0913
         streamable_http_path: str = "/mcp",
         json_response: bool = False,
         stateless_http: bool = False,
+        stateless_sse: bool = False,
         warn_on_duplicate_resources: bool = True,
         warn_on_duplicate_tools: bool = True,
         warn_on_duplicate_prompts: bool = True,
@@ -196,6 +201,7 @@ def __init__(  # noqa: PLR0913
             streamable_http_path=streamable_http_path,
             json_response=json_response,
             stateless_http=stateless_http,
+            stateless_sse=stateless_sse,
             warn_on_duplicate_resources=warn_on_duplicate_resources,
             warn_on_duplicate_tools=warn_on_duplicate_tools,
             warn_on_duplicate_prompts=warn_on_duplicate_prompts,
@@ -858,6 +864,7 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no
                     streams[0],
                     streams[1],
                     self._mcp_server.create_initialization_options(),
+                    stateless=self.settings.stateless_sse,
                 )
             return Response()
 

tests/shared/test_sse.py

@@ -602,3 +602,77 @@ async def mock_aiter_sse() -> AsyncGenerator[ServerSentEvent, None]:
             assert not isinstance(msg, Exception)
             assert isinstance(msg.message.root, types.JSONRPCResponse)
             assert msg.message.root.id == 1
+
+
+# Stateless SSE mode tests
+def make_stateless_server_app() -> Starlette:  # pragma: no cover
+    """Create test Starlette app with SSE transport in stateless mode."""
+    security_settings = TransportSecuritySettings(
+        allowed_hosts=["127.0.0.1:*", "localhost:*"],
+        allowed_origins=["http://127.0.0.1:*", "http://localhost:*"],
+    )
+    sse = SseServerTransport("/messages/", security_settings=security_settings)
+    server = ServerTest()
+
+    async def handle_sse(request: Request) -> Response:
+        async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
+            await server.run(
+                streams[0],
+                streams[1],
+                server.create_initialization_options(),
+                stateless=True,  # Enable stateless mode
+            )
+        return Response()
+
+    app = Starlette(
+        routes=[
+            Route("/sse", endpoint=handle_sse),
+            Mount("/messages/", app=sse.handle_post_message),
+        ]
+    )
+
+    return app
+
+
+def run_stateless_server(server_port: int) -> None:  # pragma: no cover
+    app = make_stateless_server_app()
+    server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error"))
+    server.run()
+
+
+@pytest.fixture()
+def stateless_server(server_port: int) -> Generator[None, None, None]:
+    proc = multiprocessing.Process(target=run_stateless_server, kwargs={"server_port": server_port}, daemon=True)
+    proc.start()
+    wait_for_server(server_port)
+    yield
+    proc.kill()
+    proc.join(timeout=2)
+
+
+@pytest.mark.anyio
+async def test_sse_stateless_mode_allows_requests_without_initialization(
+    stateless_server: None, server_url: str
+) -> None:
+    """Test that stateless SSE mode allows tool calls without initialization.
+
+    This tests the fix for issue #1844 where Claude Code (and other fast clients)
+    would send requests before the initialization handshake completed, causing
+    'Received request before initialization was complete' errors.
+
+    In stateless mode, the server bypasses the initialization requirement,
+    allowing immediate tool calls.
+    """
+    async with sse_client(server_url + "/sse") as streams:
+        async with ClientSession(*streams) as session:
+            # In stateless mode, we can call tools without initializing first
+            # Note: ClientSession still sends initialize internally, but the server
+            # doesn't require it to be completed before processing other requests
+            result = await session.initialize()
+            assert isinstance(result, InitializeResult)
+
+            # Now test that tool calls work
+            tool_result = await session.call_tool("test_tool", {})
+            assert len(tool_result.content) == 1
+            assert tool_result.content[0].type == "text"
+            assert "Called test_tool" in tool_result.content[0].text