Merge branch 'ihrpr/streamablehttp-stateless' into ihrpr/get-sse

Author: ihrpr

examples/servers/simple-streamablehttp-stateless/mcp_simple_streamablehttp_stateless/server.py

@@ -141,17 +141,14 @@ async def run_server():
                     read_stream,
                     write_stream,
                     app.create_initialization_options(),
-                    # This allows the server to run without waiting for initialization
-                    require_initialization=False,
+                    # Runs in standalone mode for stateless deployments
+                    # where clients perform initialization with any node
+                    standalone_mode=True,
                 )
 
             # Start server task
             task_group.start_soon(run_server)
 
-            # Small delay to allow the server task to start
-            # This helps prevent race conditions in stateless mode
-            await anyio.sleep(0.001)
-
             # Handle the HTTP request and return the response
             await http_transport.handle_request(scope, receive, send)
 

examples/servers/simple-streamablehttp/README.md

@@ -5,6 +5,7 @@ A simple MCP server example demonstrating the StreamableHttp transport, which en
 ## Features
 
 - Uses the StreamableHTTP transport for server-client communication
+- Supports REST API operations (POST, GET, DELETE) for `/mcp` endpoint
 - Task management with anyio task groups
 - Ability to send multiple notifications over time to the client
 - Proper resource cleanup and lifespan management

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

@@ -83,6 +83,11 @@ async def call_tool(
                 level="info",
                 data=f"Notification {i+1}/{count} from caller: {caller}",
                 logger="notification_stream",
+                # Associates this notification with the original request
+                # Ensures notifications are sent to the correct response stream
+                # Without this, notifications will either go to:
+                # - a standalone SSE stream (if GET request is supported)
+                # - nowhere (if GET request isn't supported)
                 related_request_id=ctx.request_id,
             )
             if i < count - 1:  # Don't wait after the last notification

src/mcp/server/fastmcp/server.py

@@ -466,6 +466,7 @@ async def run_stdio_async(self) -> None:
     async def run_sse_async(self) -> None:
         """Run the server using SSE transport."""
         import uvicorn
+
         starlette_app = self.sse_app()
 
         config = uvicorn.Config(
@@ -673,7 +674,10 @@ async def log(
             **extra: Additional structured data to include
         """
         await self.request_context.session.send_log_message(
-            level=level, data=message, logger=logger_name
+            level=level,
+            data=message,
+            logger=logger_name,
+            related_request_id=self.request_id,
         )
 
     @property

src/mcp/server/lowlevel/server.py

@@ -479,10 +479,11 @@ async def run(
         # but also make tracing exceptions much easier during testing and when using
         # in-process servers.
         raise_exceptions: bool = False,
-        # When True, the server will wait for the client to send an initialization
-        # message before processing any other messages.
-        # False should be used for stateless servers.
-        require_initialization: bool = True,
+        # When True, the server runs in standalone mode for stateless deployments where
+        # clients can perform initialization with any node. The client must still follow
+        # the initialization lifecycle, but can do so with any available node
+        # rather than requiring initialization for each connection.
+        standalone_mode: bool = False,
     ):
         async with AsyncExitStack() as stack:
             lifespan_context = await stack.enter_async_context(self.lifespan(self))
@@ -491,7 +492,7 @@ async def run(
                     read_stream,
                     write_stream,
                     initialization_options,
-                    require_initialization,
+                    standalone_mode=standalone_mode,
                 )
             )
 

src/mcp/server/session.py

@@ -85,15 +85,15 @@ def __init__(
         read_stream: MemoryObjectReceiveStream[types.JSONRPCMessage | Exception],
         write_stream: MemoryObjectSendStream[types.JSONRPCMessage],
         init_options: InitializationOptions,
-        require_initialization: bool = True,
+        standalone_mode: bool = False,
     ) -> None:
         super().__init__(
             read_stream, write_stream, types.ClientRequest, types.ClientNotification
         )
-        if require_initialization:
-            self._initialization_state = InitializationState.NotInitialized
-        else:
+        if standalone_mode:
             self._initialization_state = InitializationState.Initialized
+        else:
+            self._initialization_state = InitializationState.NotInitialized
         self._init_options = init_options
         self._incoming_message_stream_writer, self._incoming_message_stream_reader = (
             anyio.create_memory_object_stream[ServerRequestResponder](0)

src/mcp/server/streamableHttp.py

@@ -656,11 +656,6 @@ async def connect(
         """
 
         # Create the memory streams for this connection
-        read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception]
-        read_stream_writer: MemoryObjectSendStream[JSONRPCMessage | Exception]
-
-        write_stream: MemoryObjectSendStream[JSONRPCMessage]
-        write_stream_reader: MemoryObjectReceiveStream[JSONRPCMessage]
 
         read_stream_writer, read_stream = anyio.create_memory_object_stream[
             JSONRPCMessage | Exception
@@ -684,10 +679,13 @@ async def message_router():
                         if isinstance(
                             message.root, JSONRPCNotification | JSONRPCRequest
                         ):
-                            # Extract related_request_id from params if it exists
-                            if (params := getattr(message.root, "params", None)) and (
-                                related_id := params.get("related_request_id")
-                            ) is not None:
+                            # Extract related_request_id from meta if it exists
+                            if (
+                                (params := getattr(message.root, "params", None))
+                                and (meta := params.get("_meta"))
+                                and (related_id := meta.get("related_request_id"))
+                                is not None
+                            ):
                                 target_request_id = str(related_id)
                         else:
                             target_request_id = str(message.root.id)

src/mcp/shared/session.py

@@ -6,7 +6,6 @@
 from typing import Any, Generic, TypeVar
 
 import anyio
-import anyio.lowlevel
 import httpx
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from pydantic import BaseModel
@@ -24,6 +23,7 @@
     JSONRPCNotification,
     JSONRPCRequest,
     JSONRPCResponse,
+    NotificationParams,
     RequestParams,
     ServerNotification,
     ServerRequest,
@@ -276,8 +276,19 @@ async def send_notification(
         Emits a notification, which is a one-way message that does not expect
         a response.
         """
+        # Some transport implementations may need to set the related_request_id
+        # to attribute to the notifications to the request that triggered them.
         if related_request_id is not None and notification.root.params is not None:
-            notification.root.params.related_request_id = related_request_id
+            # Create meta if it doesn't exist
+            if notification.root.params.meta is None:
+                meta_dict = {"related_request_id": related_request_id}
+
+            else:
+                meta_dict = notification.root.params.meta.model_dump(
+                    by_alias=True, mode="json", exclude_none=True
+                )
+                meta_dict["related_request_id"] = related_request_id
+            notification.root.params.meta = NotificationParams.Meta(**meta_dict)
         jsonrpc_notification = JSONRPCNotification(
             jsonrpc="2.0",
             **notification.model_dump(by_alias=True, mode="json", exclude_none=True),

src/mcp/types.py

@@ -58,7 +58,6 @@ class Meta(BaseModel):
         model_config = ConfigDict(extra="allow")
 
     meta: Meta | None = Field(alias="_meta", default=None)
-    related_request_id: RequestId | None = None
     """
     This parameter name is reserved by MCP to allow clients and servers to attach
     additional metadata to their notifications.

tests/client/test_logging_callback.py

@@ -9,6 +9,7 @@
 from mcp.shared.session import RequestResponder
 from mcp.types import (
     LoggingMessageNotificationParams,
+    NotificationParams,
     TextContent,
 )
 
@@ -78,6 +79,11 @@ async def message_handler(
         )
         assert log_result.isError is False
         assert len(logging_collector.log_messages) == 1
-        assert logging_collector.log_messages[0] == LoggingMessageNotificationParams(
-            level="info", logger="test_logger", data="Test log message"
-        )
+        # Create meta object with related_request_id added dynamically
+        meta = NotificationParams.Meta()
+        setattr(meta, "related_request_id", "2")
+        log = logging_collector.log_messages[0]
+        assert log.level == "info"
+        assert log.logger == "test_logger"
+        assert log.data == "Test log message"
+        assert log.meta == meta