fix(client): cover streamable transport error paths

Author: go165

src/mcp/client/session_group.py

@@ -147,7 +147,7 @@ def __init__(
         self._session_exit_stacks = {}
         self._component_name_hook = component_name_hook
 
-    async def __aenter__(self) -> Self:  # pragma: no cover
+    async def __aenter__(self) -> Self:
         # Enter the exit stack only if we created it ourselves
         if self._owns_exit_stack:
             await self._exit_stack.__aenter__()
@@ -158,7 +158,7 @@ async def __aexit__(
         _exc_type: type[BaseException] | None,
         _exc_val: BaseException | None,
         _exc_tb: TracebackType | None,
-    ) -> bool | None:  # pragma: no cover
+    ) -> bool | None:
         """Closes session exit stacks and main exit stack upon completion."""
 
         # Only close the main exit stack if we created it
@@ -323,7 +323,7 @@ async def _establish_session(
             await self._exit_stack.enter_async_context(session_stack)
 
             return result.server_info, session
-        except Exception:  # pragma: no cover
+        except Exception:
             # If anything during this setup fails, ensure the session-specific
             # stack is closed.
             await session_stack.aclose()

src/mcp/client/streamable_http.py

@@ -468,25 +468,33 @@ async def _handle_message(session_message: SessionMessage) -> None:
                         read_stream_writer=read_stream_writer,
                     )
 
-                    async def handle_request_async() -> None:
-                        try:
-                            if is_resumption:
-                                await self._handle_resumption_request(ctx)
-                            else:
-                                await self._handle_post_request(ctx)
-                        except httpx.HTTPError as exc:
-                            logger.exception("Transport error handling request")
-                            if isinstance(message, JSONRPCRequest):
+                    # If this is a request, start a new task to handle it
+                    if isinstance(message, JSONRPCRequest):
+                        request_id = message.id
+
+                        async def handle_request_async() -> None:
+                            try:
+                                if is_resumption:
+                                    await self._handle_resumption_request(ctx)
+                                else:
+                                    await self._handle_post_request(ctx)
+                            except httpx.HTTPError as exc:
+                                logger.exception("Transport error handling request")
                                 error_data = ErrorData(code=INTERNAL_ERROR, message=f"Transport error: {exc}")
-                                error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
+                                error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))
                                 with contextlib.suppress(anyio.BrokenResourceError, anyio.ClosedResourceError):
                                     await read_stream_writer.send(error_msg)
 
-                    # If this is a request, start a new task to handle it
-                    if isinstance(message, JSONRPCRequest):
                         tg.start_soon(handle_request_async)
                     else:
-                        await handle_request_async()
+
+                        async def handle_notification_async() -> None:
+                            try:
+                                await self._handle_post_request(ctx)
+                            except httpx.HTTPError:
+                                logger.debug("Transport error handling notification", exc_info=True)
+
+                        await handle_notification_async()
 
                 async for session_message in write_stream_reader:
                     sender_ctx = write_stream_reader.last_context