Merge branch 'main' into swathim/mcpsessionchanges

Author: swathimshetty19

README.md

@@ -1394,6 +1394,8 @@ Run from the repository root:
     uvicorn examples.snippets.servers.streamable_http_basic_mounting:app --reload
 """
 
+import contextlib
+
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
@@ -1409,11 +1411,19 @@ def hello() -> str:
     return "Hello from MCP!"
 
 
+# Create a lifespan context manager to run the session manager
+@contextlib.asynccontextmanager
+async def lifespan(app: Starlette):
+    async with mcp.session_manager.run():
+        yield
+
+
 # Mount the StreamableHTTP server to the existing ASGI server
 app = Starlette(
     routes=[
         Mount("/", app=mcp.streamable_http_app()),
-    ]
+    ],
+    lifespan=lifespan,
 )
 ```
 
@@ -1431,6 +1441,8 @@ Run from the repository root:
     uvicorn examples.snippets.servers.streamable_http_host_mounting:app --reload
 """
 
+import contextlib
+
 from starlette.applications import Starlette
 from starlette.routing import Host
 
@@ -1446,11 +1458,19 @@ def domain_info() -> str:
     return "This is served from mcp.acme.corp"
 
 
+# Create a lifespan context manager to run the session manager
+@contextlib.asynccontextmanager
+async def lifespan(app: Starlette):
+    async with mcp.session_manager.run():
+        yield
+
+
 # Mount using Host-based routing
 app = Starlette(
     routes=[
         Host("mcp.acme.corp", app=mcp.streamable_http_app()),
-    ]
+    ],
+    lifespan=lifespan,
 )
 ```
 
@@ -1468,6 +1488,8 @@ Run from the repository root:
     uvicorn examples.snippets.servers.streamable_http_multiple_servers:app --reload
 """
 
+import contextlib
+
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
@@ -1495,12 +1517,23 @@ def send_message(message: str) -> str:
 api_mcp.settings.streamable_http_path = "/"
 chat_mcp.settings.streamable_http_path = "/"
 
+
+# Create a combined lifespan to manage both session managers
+@contextlib.asynccontextmanager
+async def lifespan(app: Starlette):
+    async with contextlib.AsyncExitStack() as stack:
+        await stack.enter_async_context(api_mcp.session_manager.run())
+        await stack.enter_async_context(chat_mcp.session_manager.run())
+        yield
+
+
 # Mount the servers
 app = Starlette(
     routes=[
         Mount("/api", app=api_mcp.streamable_http_app()),
         Mount("/chat", app=chat_mcp.streamable_http_app()),
-    ]
+    ],
+    lifespan=lifespan,
 )
 ```
 

examples/snippets/servers/streamable_http_basic_mounting.py

@@ -5,6 +5,8 @@
     uvicorn examples.snippets.servers.streamable_http_basic_mounting:app --reload
 """
 
+import contextlib
+
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
@@ -20,9 +22,17 @@ def hello() -> str:
     return "Hello from MCP!"
 
 
+# Create a lifespan context manager to run the session manager
+@contextlib.asynccontextmanager
+async def lifespan(app: Starlette):
+    async with mcp.session_manager.run():
+        yield
+
+
 # Mount the StreamableHTTP server to the existing ASGI server
 app = Starlette(
     routes=[
         Mount("/", app=mcp.streamable_http_app()),
-    ]
+    ],
+    lifespan=lifespan,
 )

examples/snippets/servers/streamable_http_host_mounting.py

@@ -5,6 +5,8 @@
     uvicorn examples.snippets.servers.streamable_http_host_mounting:app --reload
 """
 
+import contextlib
+
 from starlette.applications import Starlette
 from starlette.routing import Host
 
@@ -20,9 +22,17 @@ def domain_info() -> str:
     return "This is served from mcp.acme.corp"
 
 
+# Create a lifespan context manager to run the session manager
+@contextlib.asynccontextmanager
+async def lifespan(app: Starlette):
+    async with mcp.session_manager.run():
+        yield
+
+
 # Mount using Host-based routing
 app = Starlette(
     routes=[
         Host("mcp.acme.corp", app=mcp.streamable_http_app()),
-    ]
+    ],
+    lifespan=lifespan,
 )

examples/snippets/servers/streamable_http_multiple_servers.py

@@ -5,6 +5,8 @@
     uvicorn examples.snippets.servers.streamable_http_multiple_servers:app --reload
 """
 
+import contextlib
+
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
@@ -32,10 +34,21 @@ def send_message(message: str) -> str:
 api_mcp.settings.streamable_http_path = "/"
 chat_mcp.settings.streamable_http_path = "/"
 
+
+# Create a combined lifespan to manage both session managers
+@contextlib.asynccontextmanager
+async def lifespan(app: Starlette):
+    async with contextlib.AsyncExitStack() as stack:
+        await stack.enter_async_context(api_mcp.session_manager.run())
+        await stack.enter_async_context(chat_mcp.session_manager.run())
+        yield
+
+
 # Mount the servers
 app = Starlette(
     routes=[
         Mount("/api", app=api_mcp.streamable_http_app()),
         Mount("/chat", app=chat_mcp.streamable_http_app()),
-    ]
+    ],
+    lifespan=lifespan,
 )

src/mcp/server/streamable_http.py

@@ -238,30 +238,50 @@ def _create_session_message(  # pragma: no cover
         message: JSONRPCMessage,
         request: Request,
         request_id: RequestId,
+        protocol_version: str,
     ) -> SessionMessage:
-        """Create a session message with metadata including close_sse_stream callback."""
+        """Create a session message with metadata including close_sse_stream callback.
 
-        async def close_stream_callback() -> None:
-            self.close_sse_stream(request_id)
+        The close_sse_stream callbacks are only provided when the client supports
+        resumability (protocol version >= 2025-11-25). Old clients can't resume if
+        the stream is closed early because they didn't receive a priming event.
+        """
+        # Only provide close callbacks when client supports resumability
+        if self._event_store and protocol_version >= "2025-11-25":
 
-        async def close_standalone_stream_callback() -> None:
-            self.close_standalone_sse_stream()
+            async def close_stream_callback() -> None:
+                self.close_sse_stream(request_id)
+
+            async def close_standalone_stream_callback() -> None:
+                self.close_standalone_sse_stream()
+
+            metadata = ServerMessageMetadata(
+                request_context=request,
+                close_sse_stream=close_stream_callback,
+                close_standalone_sse_stream=close_standalone_stream_callback,
+            )
+        else:
+            metadata = ServerMessageMetadata(request_context=request)
 
-        metadata = ServerMessageMetadata(
-            request_context=request,
-            close_sse_stream=close_stream_callback,
-            close_standalone_sse_stream=close_standalone_stream_callback,
-        )
         return SessionMessage(message, metadata=metadata)
 
-    async def _send_priming_event(  # pragma: no cover
+    async def _maybe_send_priming_event(
         self,
         request_id: RequestId,
         sse_stream_writer: MemoryObjectSendStream[dict[str, Any]],
+        protocol_version: str,
     ) -> None:
-        """Send priming event for SSE resumability if event_store is configured."""
+        """Send priming event for SSE resumability if event_store is configured.
+
+        Only sends priming events to clients with protocol version >= 2025-11-25,
+        which includes the fix for handling empty SSE data. Older clients would
+        crash trying to parse empty data as JSON.
+        """
         if not self._event_store:
             return
+        # Priming events have empty data which older clients cannot handle.
+        if protocol_version < "2025-11-25":
+            return
         priming_event_id = await self._event_store.store_event(
             str(request_id),  # Convert RequestId to StreamId (str)
             None,  # Priming event has no payload
@@ -499,6 +519,15 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
 
                 return
 
+            # Extract protocol version for priming event decision.
+            # For initialize requests, get from request params.
+            # For other requests, get from header (already validated).
+            protocol_version = (
+                str(message.root.params.get("protocolVersion", DEFAULT_NEGOTIATED_VERSION))
+                if is_initialization_request and message.root.params
+                else request.headers.get(MCP_PROTOCOL_VERSION_HEADER, DEFAULT_NEGOTIATED_VERSION)
+            )
+
             # Extract the request ID outside the try block for proper scope
             request_id = str(message.root.id)  # pragma: no cover
             # Register this stream for the request ID
@@ -560,7 +589,7 @@ async def sse_writer():
                     try:
                         async with sse_stream_writer, request_stream_reader:
                             # Send priming event for SSE resumability
-                            await self._send_priming_event(request_id, sse_stream_writer)
+                            await self._maybe_send_priming_event(request_id, sse_stream_writer, protocol_version)
 
                             # Process messages from the request-specific stream
                             async for event_message in request_stream_reader:
@@ -605,7 +634,7 @@ async def sse_writer():
                     async with anyio.create_task_group() as tg:
                         tg.start_soon(response, scope, receive, send)
                         # Then send the message to be processed by the server
-                        session_message = self._create_session_message(message, request, request_id)
+                        session_message = self._create_session_message(message, request, request_id, protocol_version)
                         await writer.send(session_message)
                 except Exception:
                     logger.exception("SSE response error")
@@ -864,6 +893,9 @@ async def _replay_events(self, last_event_id: str, request: Request, send: Send)
             if self.mcp_session_id:
                 headers[MCP_SESSION_ID_HEADER] = self.mcp_session_id
 
+            # Get protocol version from header (already validated in _validate_protocol_version)
+            replay_protocol_version = request.headers.get(MCP_PROTOCOL_VERSION_HEADER, DEFAULT_NEGOTIATED_VERSION)
+
             # Create SSE stream for replay
             sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[dict[str, str]](0)
 
@@ -884,7 +916,7 @@ async def send_event(event_message: EventMessage) -> None:
                             self._sse_stream_writers[stream_id] = sse_stream_writer
 
                             # Send priming event for this new connection
-                            await self._send_priming_event(stream_id, sse_stream_writer)
+                            await self._maybe_send_priming_event(stream_id, sse_stream_writer, replay_protocol_version)
 
                             # Create new request streams for this connection
                             self._request_streams[stream_id] = anyio.create_memory_object_stream[EventMessage](0)
@@ -1000,11 +1032,16 @@ async def message_router():  # pragma: no cover
                                 # Stream might be closed, remove from registry
                                 self._request_streams.pop(request_stream_id, None)
                         else:
-                            logging.debug(
+                            logger.debug(
                                 f"""Request stream {request_stream_id} not found 
                                 for message. Still processing message as the client
                                 might reconnect and replay."""
                             )
+                except anyio.ClosedResourceError:
+                    if self._terminated:
+                        logger.debug("Read stream closed by client")
+                    else:
+                        logger.exception("Unexpected closure of read stream in message router")
                 except Exception:
                     logger.exception("Error in message router")
 

src/mcp/shared/version.py

@@ -1,3 +1,3 @@
 from mcp.types import LATEST_PROTOCOL_VERSION
 
-SUPPORTED_PROTOCOL_VERSIONS: list[str] = ["2024-11-05", "2025-03-26", LATEST_PROTOCOL_VERSION]
+SUPPORTED_PROTOCOL_VERSIONS: list[str] = ["2024-11-05", "2025-03-26", "2025-06-18", LATEST_PROTOCOL_VERSION]

src/mcp/types.py

@@ -25,7 +25,7 @@
   not separate types in the schema.
 """
 
-LATEST_PROTOCOL_VERSION = "2025-06-18"
+LATEST_PROTOCOL_VERSION = "2025-11-25"
 
 """
 The default negotiated version of the Model Context Protocol when no version is specified.

tests/issues/test_1027_win_unreachable_cleanup.py

@@ -95,7 +95,7 @@ def echo(text: str) -> str:
             async with ClientSession(read, write) as session:
                 # Initialize the session
                 result = await session.initialize()
-                assert result.protocolVersion in ["2024-11-05", "2025-06-18"]
+                assert result.protocolVersion in ["2024-11-05", "2025-06-18", "2025-11-25"]
 
                 # Verify startup marker was created
                 assert Path(startup_marker).exists(), "Server startup marker not created"