feat(client): add explicit resume flow for known streamable-http sessions

Author: mrutunjay-kinagi

src/mcp/client/client.py

@@ -98,6 +98,12 @@ async def main():
     streamable_http_session_id: str | None = None
     """Optional pre-existing MCP session ID used when server is a StreamableHTTP URL."""
 
+    streamable_http_initialize_result: InitializeResult | None = None
+    """Previously negotiated InitializeResult used to resume a StreamableHTTP session."""
+
+    streamable_http_terminate_on_close: bool = True
+    """Whether a URL-based StreamableHTTP client should terminate the session on close."""
+
     _session: ClientSession | None = field(init=False, default=None)
     _exit_stack: AsyncExitStack | None = field(init=False, default=None)
     _transport: Transport = field(init=False)
@@ -106,10 +112,31 @@ def __post_init__(self) -> None:
         if isinstance(self.server, Server | MCPServer):
             self._transport = InMemoryTransport(self.server, raise_exceptions=self.raise_exceptions)
         elif isinstance(self.server, str):
-            self._transport = streamable_http_client(self.server, session_id=self.streamable_http_session_id)
+            self._transport = streamable_http_client(
+                self.server,
+                session_id=self.streamable_http_session_id,
+                terminate_on_close=self.streamable_http_terminate_on_close,
+            )
         else:
             self._transport = self.server
 
+    @classmethod
+    def resume_session(
+        cls,
+        server: str,
+        *,
+        session_id: str,
+        initialize_result: InitializeResult,
+        **kwargs: Any,
+    ) -> Client:
+        """Create a URL-based client configured to resume an existing StreamableHTTP session."""
+        return cls(
+            server=server,
+            streamable_http_session_id=session_id,
+            streamable_http_initialize_result=initialize_result,
+            **kwargs,
+        )
+
     async def __aenter__(self) -> Client:
         """Enter the async context manager."""
         if self._session is not None:
@@ -132,7 +159,15 @@ async def __aenter__(self) -> Client:
                 )
             )
 
-            await self._session.initialize()
+            if self.streamable_http_session_id is None and self.streamable_http_initialize_result is not None:
+                raise RuntimeError(
+                    "streamable_http_initialize_result requires streamable_http_session_id for session resumption"
+                )
+
+            if self.streamable_http_session_id is not None and self.streamable_http_initialize_result is not None:
+                self._session.resume(self.streamable_http_initialize_result)
+            else:
+                await self._session.initialize()
 
             # Transfer ownership to self for __aexit__ to handle
             self._exit_stack = exit_stack.pop_all()

src/mcp/client/session.py

@@ -191,6 +191,14 @@ async def initialize(self) -> types.InitializeResult:
 
         return result
 
+    def resume(self, initialize_result: types.InitializeResult) -> None:
+        """Mark this session as resumed using previously negotiated initialization data.
+
+        This bypasses the initialize/initialized handshake and seeds the session with
+        server capabilities and metadata from an earlier connection.
+        """
+        self._initialize_result = initialize_result
+
     @property
     def initialize_result(self) -> types.InitializeResult | None:
         """The server's InitializeResult. None until initialize() has been called.

src/mcp/client/streamable_http.py

@@ -561,10 +561,19 @@ async def streamable_http_client(
             write_stream_reader,
             anyio.create_task_group() as tg,
         ):
+            get_stream_started = False
 
             def start_get_stream() -> None:
+                nonlocal get_stream_started
+                if get_stream_started:
+                    return
+                get_stream_started = True
                 tg.start_soon(transport.handle_get_stream, client, read_stream_writer)
 
+            # If we're resuming an existing session, start the GET stream immediately.
+            if session_id:
+                start_get_stream()
+
             tg.start_soon(
                 transport.post_writer,
                 client,

tests/client/test_client.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from unittest.mock import patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import anyio
 import pytest
@@ -307,13 +307,103 @@ async def test_complete_with_prompt_reference(simple_server: Server):
 def test_client_with_url_initializes_streamable_http_transport():
     with patch("mcp.client.client.streamable_http_client") as mock:
         _ = Client("http://localhost:8000/mcp")
-    mock.assert_called_once_with("http://localhost:8000/mcp", session_id=None)
+    mock.assert_called_once_with("http://localhost:8000/mcp", session_id=None, terminate_on_close=True)
 
 
 def test_client_with_url_and_session_id_initializes_streamable_http_transport():
     with patch("mcp.client.client.streamable_http_client") as mock:
         _ = Client("http://localhost:8000/mcp", streamable_http_session_id="resume-session-id")
-    mock.assert_called_once_with("http://localhost:8000/mcp", session_id="resume-session-id")
+    mock.assert_called_once_with(
+        "http://localhost:8000/mcp",
+        session_id="resume-session-id",
+        terminate_on_close=True,
+    )
+
+
+def test_client_with_url_and_terminate_on_close_false_initializes_streamable_http_transport():
+    with patch("mcp.client.client.streamable_http_client") as mock:
+        _ = Client("http://localhost:8000/mcp", streamable_http_terminate_on_close=False)
+    mock.assert_called_once_with("http://localhost:8000/mcp", session_id=None, terminate_on_close=False)
+
+
+def test_client_resume_session_builder_initializes_streamable_http_transport():
+    initialize_result = types.InitializeResult(
+        protocol_version="2025-03-26",
+        capabilities=types.ServerCapabilities(),
+        server_info=types.Implementation(name="server", version="1.0"),
+    )
+    with patch("mcp.client.client.streamable_http_client") as mock:
+        _ = Client.resume_session(
+            "http://localhost:8000/mcp",
+            session_id="resume-session-id",
+            initialize_result=initialize_result,
+        )
+    mock.assert_called_once_with(
+        "http://localhost:8000/mcp",
+        session_id="resume-session-id",
+        terminate_on_close=True,
+    )
+
+
+async def test_client_resume_session_skips_initialize():
+    initialize_result = types.InitializeResult(
+        protocol_version="2025-03-26",
+        capabilities=types.ServerCapabilities(),
+        server_info=types.Implementation(name="server", version="1.0"),
+    )
+
+    transport_cm = AsyncMock()
+    transport_cm.__aenter__.return_value = (MagicMock(), MagicMock())
+    transport_cm.__aexit__.return_value = None
+
+    session = MagicMock()
+    session.initialize = AsyncMock()
+    session.resume = MagicMock()
+    session.initialize_result = initialize_result
+    session_cm = AsyncMock()
+    session_cm.__aenter__.return_value = session
+    session_cm.__aexit__.return_value = None
+
+    with (
+        patch("mcp.client.client.streamable_http_client", return_value=transport_cm),
+        patch("mcp.client.client.ClientSession", return_value=session_cm),
+    ):
+        async with Client.resume_session(
+            "http://localhost:8000/mcp",
+            session_id="resume-session-id",
+            initialize_result=initialize_result,
+        ) as client:
+            assert client.initialize_result == initialize_result
+
+    session.initialize.assert_not_awaited()
+    session.resume.assert_called_once_with(initialize_result)
+
+
+async def test_client_streamable_initialize_result_requires_session_id():
+    initialize_result = types.InitializeResult(
+        protocol_version="2025-03-26",
+        capabilities=types.ServerCapabilities(),
+        server_info=types.Implementation(name="server", version="1.0"),
+    )
+
+    transport_cm = AsyncMock()
+    transport_cm.__aenter__.return_value = (MagicMock(), MagicMock())
+    transport_cm.__aexit__.return_value = None
+
+    session_cm = AsyncMock()
+    session_cm.__aenter__.return_value = AsyncMock()
+    session_cm.__aexit__.return_value = None
+
+    with (
+        patch("mcp.client.client.streamable_http_client", return_value=transport_cm),
+        patch("mcp.client.client.ClientSession", return_value=session_cm),
+    ):
+        client = Client(
+            "http://localhost:8000/mcp",
+            streamable_http_initialize_result=initialize_result,
+        )
+        with pytest.raises(RuntimeError, match="requires streamable_http_session_id"):
+            await client.__aenter__()
 
 
 async def test_client_uses_transport_directly(app: MCPServer):