feat: add exception group unwrapping utility

ROOT CAUSE:
Task groups wrap real errors with CancelledError from siblings,
making error handling difficult for callers.

CHANGES:
- Added unwrap_task_group_exception() utility function
- Extracts real error from ExceptionGroup, ignores cancelled siblings

IMPACT:
- Enables clean error handling for SDK users

FILES MODIFIED:
- src/mcp/shared/exceptions.py: Added unwrap_task_group_exception()
- tests/shared/test_exceptions.py: Added tests for unwrapping behavior

Author: peter-luminova

src/mcp/shared/exceptions.py

@@ -104,3 +104,47 @@ def from_error(cls, error: ErrorData) -> UrlElicitationRequiredError:
         raw_elicitations = cast(list[dict[str, Any]], data.get("elicitations", []))
         elicitations = [ElicitRequestURLParams.model_validate(e) for e in raw_elicitations]
         return cls(elicitations, error.message)
+
+
+def unwrap_task_group_exception(exc: BaseException) -> BaseException:
+    """Unwrap an exception from a task group, extracting only the real error.
+
+    When anyio task groups fail, they raise BaseExceptionGroup containing:
+    - The original error that caused the failure
+    - CancelledError from sibling tasks that were cancelled
+
+    This function extracts only the real error, ignoring cancelled siblings.
+
+    Args:
+        exc: The exception to unwrap (could be any exception)
+
+    Returns:
+        The unwrapped exception if it was an ExceptionGroup with a real error,
+        otherwise the original exception
+
+    Example:
+        ```python
+        try:
+            async with anyio.create_task_group() as tg:
+                tg.start_soon(task1)
+                tg.start_soon(task2)
+        except BaseExceptionGroup as e:
+            # Extract only the real error, ignore CancelledError
+            real_exc = unwrap_task_group_exception(e)
+            raise real_exc
+        ```
+    """
+    import anyio
+
+    # If not an exception group, return as-is
+    if not isinstance(exc, BaseExceptionGroup):
+        return exc
+
+    # Find the first non-cancelled exception
+    cancelled_exc_class = anyio.get_cancelled_exc_class()
+    for sub_exc in exc.exceptions:
+        if not isinstance(sub_exc, cancelled_exc_class):
+            return sub_exc
+
+    # All were cancelled, return the group
+    return exc

tests/shared/test_exceptions.py

@@ -162,3 +162,66 @@ def test_url_elicitation_required_error_exception_message() -> None:
 
     # The exception's string representation should match the message
     assert str(error) == "URL elicitation required"
+
+
+# Tests for unwrap_task_group_exception
+import anyio
+
+
+@pytest.mark.anyio
+async def test_unwrap_single_error() -> None:
+    """Test that a single exception is returned as-is."""
+    from mcp.shared.exceptions import unwrap_task_group_exception
+
+    error = ValueError("test error")
+    result = unwrap_task_group_exception(error)
+    assert result is error
+
+
+@pytest.mark.anyio
+async def test_unwrap_exception_group_with_real_error() -> None:
+    """Test that real error is extracted from ExceptionGroup."""
+    from mcp.shared.exceptions import unwrap_task_group_exception
+
+    real_error = ConnectionError("connection failed")
+
+    # Simulate what anyio does: create exception group with real error + cancelled
+    try:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(lambda: (_ for _ in ()).throw(real_error))
+            tg.start_soon(anyio.sleep, 999)  # Will be cancelled
+    except BaseExceptionGroup as e:
+        result = unwrap_task_group_exception(e)
+        assert isinstance(result, ConnectionError)
+        assert str(result) == "connection failed"
+
+
+@pytest.mark.anyio
+async def test_unwrap_exception_group_all_cancelled() -> None:
+    """Test that when all exceptions are cancelled, the group is re-raised."""
+    from mcp.shared.exceptions import unwrap_task_group_exception
+
+    try:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(anyio.sleep, 999)
+            tg.cancel_scope.cancel()
+    except BaseExceptionGroup as e:
+        # Should return the group if all are cancelled
+        result = unwrap_task_group_exception(e)
+        assert isinstance(result, BaseExceptionGroup)
+
+
+@pytest.mark.anyio
+async def test_unwrap_preserves_non_cancelled_errors() -> None:
+    """Test that all non-cancelled exceptions are preserved."""
+    from mcp.shared.exceptions import unwrap_task_group_exception
+
+    error1 = ValueError("error 1")
+    error2 = RuntimeError("error 2")
+
+    # Create an exception group with multiple real errors
+    group = BaseExceptionGroup("multiple", [error1, error2])
+
+    result = unwrap_task_group_exception(group)
+    # Should return the first non-cancelled error
+    assert result is error1