fix(remaining): unwrap ExceptionGroup in remaining task groups

ROOT CAUSE:
Remaining task group usages propagate ExceptionGroup.

CHANGES:
- Added exception unwrapping in StreamableHTTPManager
- Added exception unwrapping in lowlevel server
- Added exception unwrapping in experimental task support
- Added exception unwrapping in task result handler
- Added exception unwrapping in session group
- Added exception unwrapping in memory transport

IMPACT:
- All task groups now properly unwrap ExceptionGroups

FILES MODIFIED:
- src/mcp/server/streamable_http_manager.py
- src/mcp/server/lowlevel/server.py
- src/mcp/server/experimental/task_support.py
- src/mcp/server/experimental/task_result_handler.py
- src/mcp/client/session_group.py
- src/mcp/client/_memory.py

Author: peter-luminova

src/mcp/client/_memory.py

@@ -49,20 +49,27 @@ async def _connect(self) -> AsyncIterator[TransportStreams]:
             server_read, server_write = server_streams
 
             async with anyio.create_task_group() as tg:
-                # Start server in background
-                tg.start_soon(
-                    lambda: actual_server.run(
-                        server_read,
-                        server_write,
-                        actual_server.create_initialization_options(),
-                        raise_exceptions=self._raise_exceptions,
+                try:
+                    # Start server in background
+                    tg.start_soon(
+                        lambda: actual_server.run(
+                            server_read,
+                            server_write,
+                            actual_server.create_initialization_options(),
+                            raise_exceptions=self._raise_exceptions,
+                        )
                     )
-                )
 
-                try:
-                    yield client_read, client_write
-                finally:
-                    tg.cancel_scope.cancel()
+                    try:
+                        yield client_read, client_write
+                    finally:
+                        tg.cancel_scope.cancel()
+                except BaseExceptionGroup as e:
+                    from mcp.shared.exceptions import unwrap_task_group_exception
+
+                    real_exc = unwrap_task_group_exception(e)
+                    if real_exc is not e:
+                        raise real_exc
 
     async def __aenter__(self) -> TransportStreams:
         """Connect to the server and return streams for communication."""

src/mcp/client/session_group.py

@@ -167,8 +167,15 @@ async def __aexit__(
 
         # Concurrently close session stacks.
         async with anyio.create_task_group() as tg:
-            for exit_stack in self._session_exit_stacks.values():
-                tg.start_soon(exit_stack.aclose)
+            try:
+                for exit_stack in self._session_exit_stacks.values():
+                    tg.start_soon(exit_stack.aclose)
+            except BaseExceptionGroup as e:
+                from mcp.shared.exceptions import unwrap_task_group_exception
+
+                real_exc = unwrap_task_group_exception(e)
+                if real_exc is not e:
+                    raise real_exc
 
     @property
     def sessions(self) -> list[mcp.ClientSession]:

src/mcp/server/experimental/task_result_handler.py

@@ -163,25 +163,31 @@ async def _wait_for_task_update(self, task_id: str) -> None:
         Races between store update and queue message - first one wins.
         """
         async with anyio.create_task_group() as tg:
-
-            async def wait_for_store() -> None:
-                try:
-                    await self._store.wait_for_update(task_id)
-                except Exception:
-                    pass
-                finally:
-                    tg.cancel_scope.cancel()
-
-            async def wait_for_queue() -> None:
-                try:
-                    await self._queue.wait_for_message(task_id)
-                except Exception:
-                    pass
-                finally:
-                    tg.cancel_scope.cancel()
-
-            tg.start_soon(wait_for_store)
-            tg.start_soon(wait_for_queue)
+            try:
+                async def wait_for_store() -> None:
+                    try:
+                        await self._store.wait_for_update(task_id)
+                    except Exception:
+                        pass
+                    finally:
+                        tg.cancel_scope.cancel()
+
+                async def wait_for_queue() -> None:
+                    try:
+                        await self._queue.wait_for_message(task_id)
+                    except Exception:
+                        pass
+                    finally:
+                        tg.cancel_scope.cancel()
+
+                tg.start_soon(wait_for_store)
+                tg.start_soon(wait_for_queue)
+            except BaseExceptionGroup as e:
+                from mcp.shared.exceptions import unwrap_task_group_exception
+
+                real_exc = unwrap_task_group_exception(e)
+                if real_exc is not e:
+                    raise real_exc
 
     def route_response(self, request_id: RequestId, response: dict[str, Any]) -> bool:
         """Route a response back to the waiting resolver.

src/mcp/server/experimental/task_support.py

@@ -80,11 +80,18 @@ async def run(self) -> AsyncIterator[None]:
                 ...
         """
         async with anyio.create_task_group() as tg:
-            self._task_group = tg
             try:
-                yield
-            finally:
-                self._task_group = None
+                self._task_group = tg
+                try:
+                    yield
+                finally:
+                    self._task_group = None
+            except BaseExceptionGroup as e:
+                from mcp.shared.exceptions import unwrap_task_group_exception
+
+                real_exc = unwrap_task_group_exception(e)
+                if real_exc is not e:
+                    raise real_exc
 
     def configure_session(self, session: ServerSession) -> None:
         """Configure a session for task support.

src/mcp/server/lowlevel/server.py

@@ -390,16 +390,23 @@ async def run(
                 await stack.enter_async_context(task_support.run())
 
             async with anyio.create_task_group() as tg:
-                async for message in session.incoming_messages:
-                    logger.debug("Received message: %s", message)
-
-                    tg.start_soon(
-                        self._handle_message,
-                        message,
-                        session,
-                        lifespan_context,
-                        raise_exceptions,
-                    )
+                try:
+                    async for message in session.incoming_messages:
+                        logger.debug("Received message: %s", message)
+
+                        tg.start_soon(
+                            self._handle_message,
+                            message,
+                            session,
+                            lifespan_context,
+                            raise_exceptions,
+                        )
+                except BaseExceptionGroup as e:
+                    from mcp.shared.exceptions import unwrap_task_group_exception
+
+                    real_exc = unwrap_task_group_exception(e)
+                    if real_exc is not e:
+                        raise real_exc
 
     async def _handle_message(
         self,

src/mcp/server/streamable_http_manager.py

@@ -123,18 +123,25 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
             self._has_started = True
 
         async with anyio.create_task_group() as tg:
-            # Store the task group for later use
-            self._task_group = tg
-            logger.info("StreamableHTTP session manager started")
             try:
-                yield  # Let the application run
-            finally:
-                logger.info("StreamableHTTP session manager shutting down")
-                # Cancel task group to stop all spawned tasks
-                tg.cancel_scope.cancel()
-                self._task_group = None
-                # Clear any remaining server instances
-                self._server_instances.clear()
+                # Store the task group for later use
+                self._task_group = tg
+                logger.info("StreamableHTTP session manager started")
+                try:
+                    yield  # Let the application run
+                finally:
+                    logger.info("StreamableHTTP session manager shutting down")
+                    # Cancel task group to stop all spawned tasks
+                    tg.cancel_scope.cancel()
+                    self._task_group = None
+                    # Clear any remaining server instances
+                    self._server_instances.clear()
+            except BaseExceptionGroup as e:
+                from mcp.shared.exceptions import unwrap_task_group_exception
+
+                real_exc = unwrap_task_group_exception(e)
+                if real_exc is not e:
+                    raise real_exc
 
     async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> None:
         """Process ASGI request with proper session handling and transport setup.