Restore SSE request headers via AsyncClient.sse(); fix example fallout

httpx-sse's aconnect_sse() always sent Accept: text/event-stream and
Cache-Control: no-store; the swap to bare client.stream() dropped both.
Open the legacy SSE GET and the streamable HTTP GET/resumption/
reconnection streams with AsyncClient.sse(), which injects those
headers (explicit caller headers still take precedence), and update
the mocked sse_client test to drive the new call.

Example fixes from the same review pass:
- simple-chatbot: catch httpx2.HTTPError instead of RequestError so
  raise_for_status() failures take the handled path (the
  HTTPStatusError isinstance branch was unreachable), and drop the
  Raises section the method never honoured.
- sse-polling-client: suppress the httpcore2 logger; httpcore is no
  longer in the dependency tree, so the old suppression was a no-op.

Author: maxisbey

examples/clients/simple-chatbot/mcp_simple_chatbot/main.py

@@ -227,10 +227,7 @@ def get_response(self, messages: list[dict[str, str]]) -> str:
             messages: A list of message dictionaries.
 
         Returns:
-            The LLM's response as a string.
-
-        Raises:
-            httpx2.RequestError: If the request to the LLM fails.
+            The LLM's response as a string, or an error message if the request fails.
         """
         url = "https://api.groq.com/openai/v1/chat/completions"
 
@@ -255,7 +252,7 @@ def get_response(self, messages: list[dict[str, str]]) -> str:
                 data = response.json()
                 return data["choices"][0]["message"]["content"]
 
-        except httpx2.RequestError as e:
+        except httpx2.HTTPError as e:
             error_message = f"Error getting LLM response: {str(e)}"
             logging.error(error_message)
 

examples/clients/sse-polling-client/mcp_sse_polling_client/main.py

@@ -93,7 +93,7 @@ def main(url: str, items: int, checkpoint_every: int, log_level: str) -> None:
     )
     # Suppress noisy HTTP client logging
     logging.getLogger("httpx2").setLevel(logging.WARNING)
-    logging.getLogger("httpcore").setLevel(logging.WARNING)
+    logging.getLogger("httpcore2").setLevel(logging.WARNING)
 
     asyncio.run(run_demo(url, items, checkpoint_every))
 

src/mcp/client/sse.py

@@ -8,7 +8,7 @@
 import httpx2
 import mcp_types as types
 from anyio.abc import TaskStatus
-from httpx2 import EventSource, SSEError
+from httpx2 import SSEError
 
 from mcp.shared._compat import resync_tracer
 from mcp.shared._context_streams import create_context_streams
@@ -55,9 +55,8 @@ async def sse_client(
     async with httpx_client_factory(
         headers=headers, auth=auth, timeout=httpx2.Timeout(timeout, read=sse_read_timeout)
     ) as client:
-        async with client.stream("GET", url) as response:
-            event_source = EventSource(response)
-            response.raise_for_status()
+        async with client.sse(url) as event_source:
+            event_source.response.raise_for_status()
             logger.debug("SSE connection established")
 
             read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)

src/mcp/client/streamable_http.py

@@ -187,11 +187,11 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer
                 if last_event_id:
                     headers[LAST_EVENT_ID] = last_event_id
 
-                async with client.stream("GET", self.url, headers=headers) as response:
-                    response.raise_for_status()
+                async with client.sse(self.url, headers=headers) as event_source:
+                    event_source.response.raise_for_status()
                     logger.debug("GET SSE connection established")
 
-                    async for sse in EventSource(response):
+                    async for sse in event_source:
                         # Track last event ID for reconnection
                         if sse.id:
                             last_event_id = sse.id
@@ -230,19 +230,19 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         if isinstance(ctx.session_message.message, JSONRPCRequest):  # pragma: no branch
             original_request_id = ctx.session_message.message.id
 
-        async with ctx.client.stream("GET", self.url, headers=headers) as response:
-            response.raise_for_status()
+        async with ctx.client.sse(self.url, headers=headers) as event_source:
+            event_source.response.raise_for_status()
             logger.debug("Resumption GET SSE connection established")
 
-            async for sse in EventSource(response):  # pragma: no branch
+            async for sse in event_source:  # pragma: no branch
                 is_complete = await self._handle_sse_event(
                     sse,
                     ctx.read_stream_writer,
                     original_request_id,
                     ctx.metadata.on_resumption_token_update if ctx.metadata else None,
                 )
                 if is_complete:
-                    await response.aclose()
+                    await event_source.response.aclose()
                     break
 
     async def _handle_post_request(self, ctx: RequestContext) -> None:
@@ -408,15 +408,15 @@ async def _handle_reconnection(
             original_request_id = ctx.session_message.message.id
 
         try:
-            async with ctx.client.stream("GET", self.url, headers=headers) as response:
-                response.raise_for_status()
+            async with ctx.client.sse(self.url, headers=headers) as event_source:
+                event_source.response.raise_for_status()
                 logger.info("Reconnected to SSE stream")
 
                 # Track for potential further reconnection
                 reconnect_last_event_id: str = last_event_id
                 reconnect_retry_ms = retry_interval_ms
 
-                async for sse in EventSource(response):
+                async for sse in event_source:
                     if sse.id:  # pragma: no branch
                         reconnect_last_event_id = sse.id
                     if sse.retry is not None:

tests/shared/test_sse.py

@@ -3,7 +3,7 @@
 import json
 from collections.abc import AsyncGenerator
 from typing import Any
-from unittest.mock import AsyncMock, MagicMock, Mock, patch
+from unittest.mock import AsyncMock, MagicMock, Mock
 from urllib.parse import urlparse
 
 import anyio
@@ -416,26 +416,24 @@ async def test_sse_client_handles_empty_keepalive_pings() -> None:
     )
     response_json = response.model_dump_json(by_alias=True, exclude_none=True)
 
-    # Create mock SSE events using httpx2's ServerSentEvent
-    async def mock_aiter_sse() -> AsyncGenerator[ServerSentEvent, None]:
-        # First: endpoint event
-        yield ServerSentEvent(event="endpoint", data="/messages/?session_id=abc123")
-        # Empty data keep-alive ping - this is what we're testing
-        yield ServerSentEvent(event="message", data="")
-        # Real JSON-RPC response
-        yield ServerSentEvent(event="message", data=response_json)
+    # Mock SSE events using httpx2's ServerSentEvent: an endpoint event, an
+    # empty keep-alive ping (the case under test), then a real response.
+    mock_event_source = MagicMock()
+    mock_event_source.__aiter__.return_value = [
+        ServerSentEvent(event="endpoint", data="/messages/?session_id=abc123"),
+        ServerSentEvent(event="message", data=""),
+        ServerSentEvent(event="message", data=response_json),
+    ]
+    mock_event_source.response.raise_for_status = MagicMock()
 
-    mock_response = MagicMock()
-    mock_response.raise_for_status = MagicMock()
-
-    mock_stream = MagicMock()
-    mock_stream.__aenter__ = AsyncMock(return_value=mock_response)
-    mock_stream.__aexit__ = AsyncMock(return_value=None)
+    mock_sse = MagicMock()
+    mock_sse.__aenter__ = AsyncMock(return_value=mock_event_source)
+    mock_sse.__aexit__ = AsyncMock(return_value=None)
 
     mock_client = MagicMock()
     mock_client.__aenter__ = AsyncMock(return_value=mock_client)
     mock_client.__aexit__ = AsyncMock(return_value=None)
-    mock_client.stream = MagicMock(return_value=mock_stream)
+    mock_client.sse = MagicMock(return_value=mock_sse)
     mock_client.post = AsyncMock(return_value=MagicMock(status_code=200, raise_for_status=MagicMock()))
 
     def mock_factory(
@@ -445,14 +443,13 @@ def mock_factory(
     ) -> httpx2.AsyncClient:
         return mock_client
 
-    with patch("mcp.client.sse.EventSource", return_value=mock_aiter_sse()):
-        async with sse_client("http://test/sse", httpx_client_factory=mock_factory) as (read_stream, _):
-            # Read the message - should skip the empty one and get the real response
-            msg = await read_stream.receive()
-            # If we get here without error, the empty message was skipped successfully
-            assert not isinstance(msg, Exception)
-            assert isinstance(msg.message, types.JSONRPCResponse)
-            assert msg.message.id == 1
+    async with sse_client("http://test/sse", httpx_client_factory=mock_factory) as (read_stream, _):
+        # Read the message - should skip the empty one and get the real response
+        msg = await read_stream.receive()
+        # If we get here without error, the empty message was skipped successfully
+        assert not isinstance(msg, Exception)
+        assert isinstance(msg.message, types.JSONRPCResponse)
+        assert msg.message.id == 1
 
 
 @pytest.mark.anyio