feat(client): support explicit streamable-http session resumption

Author: mrutunjay-kinagi

src/mcp/client/client.py

@@ -95,6 +95,9 @@ async def main():
     elicitation_callback: ElicitationFnT | None = None
     """Callback for handling elicitation requests."""
 
+    streamable_http_session_id: str | None = None
+    """Optional pre-existing MCP session ID used when server is a StreamableHTTP URL."""
+
     _session: ClientSession | None = field(init=False, default=None)
     _exit_stack: AsyncExitStack | None = field(init=False, default=None)
     _transport: Transport = field(init=False)
@@ -103,7 +106,7 @@ 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)
+            self._transport = streamable_http_client(self.server, session_id=self.streamable_http_session_id)
         else:
             self._transport = self.server
 

src/mcp/client/session_group.py

@@ -63,6 +63,9 @@ class StreamableHttpParameters(BaseModel):
     # Close the client session when the transport closes.
     terminate_on_close: bool = True
 
+    # Optional pre-existing MCP session ID for explicit session resumption.
+    session_id: str | None = None
+
 
 ServerParameters: TypeAlias = StdioServerParameters | SseServerParameters | StreamableHttpParameters
 
@@ -296,6 +299,7 @@ async def _establish_session(
                     url=server_params.url,
                     http_client=httpx_client,
                     terminate_on_close=server_params.terminate_on_close,
+                    session_id=server_params.session_id,
                 )
                 read, write = await session_stack.enter_async_context(client)
 

src/mcp/client/streamable_http.py

@@ -72,14 +72,15 @@ class RequestContext:
 class StreamableHTTPTransport:
     """StreamableHTTP client transport implementation."""
 
-    def __init__(self, url: str) -> None:
+    def __init__(self, url: str, session_id: str | None = None) -> None:
         """Initialize the StreamableHTTP transport.
 
         Args:
             url: The endpoint URL.
+            session_id: Optional pre-existing MCP session ID to resume.
         """
         self.url = url
-        self.session_id: str | None = None
+        self.session_id: str | None = session_id
         self.protocol_version: str | None = None
 
     def _prepare_headers(self) -> dict[str, str]:
@@ -512,6 +513,7 @@ async def streamable_http_client(
     *,
     http_client: httpx.AsyncClient | None = None,
     terminate_on_close: bool = True,
+    session_id: str | None = None,
 ) -> AsyncGenerator[TransportStreams, None]:
     """Client transport for StreamableHTTP.
 
@@ -521,6 +523,8 @@ async def streamable_http_client(
             client with recommended MCP timeouts will be created. To configure headers,
             authentication, or other HTTP settings, create an httpx.AsyncClient and pass it here.
         terminate_on_close: If True, send a DELETE request to terminate the session when the context exits.
+        session_id: Optional pre-existing MCP session ID to include in requests,
+            enabling explicit session resumption.
 
     Yields:
         Tuple containing:
@@ -538,7 +542,7 @@ async def streamable_http_client(
         # Create default client with recommended MCP timeouts
         client = create_mcp_http_client()
 
-    transport = StreamableHTTPTransport(url)
+    transport = StreamableHTTPTransport(url, session_id=session_id)
 
     logger.debug(f"Connecting to StreamableHTTP endpoint: {url}")
 

tests/client/test_client.py

@@ -307,7 +307,13 @@ 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")
+    mock.assert_called_once_with("http://localhost:8000/mcp", session_id=None)
+
+
+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")
 
 
 async def test_client_uses_transport_directly(app: MCPServer):

tests/client/test_session_group.py

@@ -294,7 +294,11 @@ async def test_client_session_group_disconnect_non_existent_server():
             "mcp.client.session_group.sse_client",
         ),  # url, headers, timeout, sse_read_timeout
         (
-            StreamableHttpParameters(url="http://test.com/stream", terminate_on_close=False),
+            StreamableHttpParameters(
+                url="http://test.com/stream",
+                terminate_on_close=False,
+                session_id="resumed-session-id",
+            ),
             "streamablehttp",
             "mcp.client.session_group.streamable_http_client",
         ),  # url, headers, timeout, sse_read_timeout, terminate_on_close
@@ -363,6 +367,7 @@ async def test_client_session_group_establish_session_parameterized(
                 call_args = mock_specific_client_func.call_args
                 assert call_args.kwargs["url"] == server_params_instance.url
                 assert call_args.kwargs["terminate_on_close"] == server_params_instance.terminate_on_close
+                assert call_args.kwargs["session_id"] == server_params_instance.session_id
                 assert isinstance(call_args.kwargs["http_client"], httpx.AsyncClient)
 
             mock_client_cm_instance.__aenter__.assert_awaited_once()

tests/shared/test_streamable_http.py

@@ -1801,6 +1801,14 @@ async def test_handle_sse_event_skips_empty_data():
         await read_stream.aclose()
 
 
+def test_streamable_http_transport_includes_seeded_session_id_header():
+    transport = StreamableHTTPTransport(url="http://localhost:8000/mcp", session_id="resume-session-id")
+
+    headers = transport._prepare_headers()
+
+    assert headers["mcp-session-id"] == "resume-session-id"
+
+
 @pytest.mark.anyio
 async def test_priming_event_not_sent_for_old_protocol_version():
     """Test that _maybe_send_priming_event skips for old protocol versions (backwards compat)."""