[OSS-137] Report MCP HTTP auth failures instead of cancelled connections (#7067)

* feat(mcp): add shared error classifier for HTTP auth failures

When an MCP server refuses a streamable-HTTP connection, the HTTP status is
observed by the client but often buried inside anyio teardown. Add typed
connection exceptions and helpers to recover the status from exception groups,
CancelledError context chains, and httpx errors so later call sites can report
authentication failures instead of guessing.

Groundwork only; no call sites wired yet.

* feat(mcp): raise typed errors from HTTPTransport.connect on HTTP status

When streamable-HTTP connect fails with an httpx HTTPStatusError, classify
the status via the shared MCP exception helpers and raise MCPAuthenticationError
for 401/403 or MCPHTTPError for other refused statuses instead of a generic
ConnectionError that hides the credential problem.

* refactor(mcp): centralize raise_connection_failure and simplify connect

Move connection failure classification into exceptions.py so transports and
clients share one helper. Flatten HTTPTransport.connect to a single except
path that cleans up once and raises, avoiding the outer handler re-classifying
errors the inner handler already typed.

* fix(mcp): classify auth failures in MCPClient.connect before reporting cancelled

When a streamable-HTTP server refuses the connection, the awaiting coroutine
often sees only CancelledError while the HTTP status surfaces during transport
unwind. Inspect cleanup for the status before emitting error_type=cancelled,
and fix HTTPTransport.disconnect so it raises typed errors instead of
suppressing exception groups that carry the refusal.

* fix(mcp): replace speculative tool resolver errors with classifier

Use raise_connection_failure for native MCP discovery instead of hedged
cancel-scope wording, preserve typed MCPConnectionError from setup, and
detect event-loop presence explicitly so ConnectionError is not mistaken
for a missing running loop. Update HTTPS discovery to classify HTTP status
codes via find_http_status.

* refactor(mcp): collapse native tool resolver failure handlers

CancelledError is not an Exception subclass, so handle it alongside
Exception in one except clause and delegate to a shared helper.

* refactor(mcp): call raise_connection_failure directly in tool resolver

* fix(mcp): classify tool execution auth failures in events

Add tool_execution_error_type so call_tool_result emits authentication
instead of server_error for MCPAuthenticationError and HTTP 401/403.
Preserve typed MCPConnectionError in _retry_operation instead of flattening
them into a generic ConnectionError first.

* feat(mcp): add status_code to MCPConnectionFailedEvent

Surface the HTTP status observed during connection failures on the event
payload and in verbose console output, so executions and checkpoints record
401/403 alongside error_type=authentication instead of only the message text.

* fix(mcp): handle cancellation and exception groups in auth paths

Ensure discovery cleanup runs on CancelledError, classify mixed
BaseExceptionGroups during HTTP connect, and fix ExceptionGroup imports
on Python 3.10 with regression tests.

* fix(mcp): preserve auth errors from discovery disconnect cleanup

Re-raise MCPConnectionError from disconnect during cancellation cleanup
instead of logging and swallowing it, with a regression test.

* fix(mcp): unwind transport context to recover auth on cancel

Always exit pending streamable-HTTP contexts before classifying failures,
handle CancelledError during client cleanup, propagate typed HTTPS discovery
errors, and add regression tests for the teardown recovery path.

* fix(mcp): classify auth from groups and timeout teardown

Handle BaseExceptionGroup in HTTPS discovery and recover HTTP 401 from
streamable-HTTP context exit after connect timeouts, with regression tests.

* refactor(mcp): consolidate client connection failure reporting

Extract _report_connection_failure and delegate _http_failure and
_connection_failure to it without changing connect error behavior.

* refactor(mcp): drop redundant client failure helper wrappers

Call _report_connection_failure directly from connect() instead of
_http_failure and _connection_failure delegators.

* fix(mcp): propagate CancelledError after HTTP transport teardown

Re-raise cancellation from disconnect when no HTTP auth status is
recovered during context unwind, with a regression test.

* fix(mcp): preserve typed errors from MCPClient.disconnect

Re-raise MCPConnectionError and CancelledError from exit-stack teardown
instead of wrapping auth failures in RuntimeError, with a regression test.

Author: Vidit-Ostwal

lib/crewai/src/crewai/events/event_listener.py

@@ -919,6 +919,7 @@ def on_mcp_connection_failed(_: Any, event: MCPConnectionFailedEvent) -> None:
                 event.transport_type,
                 event.error,
                 event.error_type,
+                event.status_code,
             )
             self._telemetry.feature_usage_span("mcp:connection_failed")
 

lib/crewai/src/crewai/events/types/mcp_events.py

@@ -49,6 +49,7 @@ class MCPConnectionFailedEvent(MCPEvent):
     type: Literal["mcp_connection_failed"] = "mcp_connection_failed"
     error: str
     error_type: str | None = None  # "timeout", "authentication", "network", etc.
+    status_code: int | None = None
     started_at: datetime | None = None
     failed_at: datetime | None = None
 

lib/crewai/src/crewai/events/utils/console_formatter.py

@@ -1665,6 +1665,7 @@ def handle_mcp_connection_failed(
         transport_type: str | None = None,
         error: str = "",
         error_type: str | None = None,
+        status_code: int | None = None,
     ) -> None:
         """Handle MCP connection failed event."""
         if not self.verbose:
@@ -1683,6 +1684,10 @@ def handle_mcp_connection_failed(
             content.append("Transport: ", style="white")
             content.append(f"{transport_type}\n", style="red")
 
+        if status_code is not None:
+            content.append("HTTP Status: ", style="white")
+            content.append(f"{status_code}\n", style="red")
+
         if error_type:
             content.append("Error Type: ", style="white")
             content.append(f"{error_type}\n", style="red")

lib/crewai/src/crewai/mcp/client.py

@@ -27,6 +27,14 @@
     MCPToolExecutionFailedEvent,
     MCPToolExecutionStartedEvent,
 )
+from crewai.mcp.exceptions import (
+    MCPConnectionError,
+    error_for_status,
+    error_type_for_status,
+    find_http_status,
+    find_transport_failure,
+    tool_execution_error_type,
+)
 from crewai.mcp.transports.base import BaseTransport
 from crewai.mcp.transports.http import HTTPTransport
 from crewai.mcp.transports.sse import SSETransport
@@ -148,13 +156,18 @@ async def connect(self) -> Self:
             Self for method chaining.
 
         Raises:
-            ConnectionError: If connection fails.
+            MCPAuthenticationError: If the server refused the connection with
+                an authentication status.
+            MCPHTTPError: If the server refused the connection with any other
+                HTTP status.
+            MCPConnectionError: If the connection failed for any other reason.
             ImportError: If MCP SDK not available.
         """
         if self.connected:
             return self
 
-        server_name, server_url, transport_type = self._get_server_info()
+        server_info = self._get_server_info()
+        server_name, server_url, transport_type = server_info
         is_reconnect = self._was_connected
 
         started_at = datetime.now()
@@ -184,40 +197,13 @@ async def connect(self) -> Self:
 
             await self._exit_stack.enter_async_context(self._session)
 
-            # MCP protocol requires session.initialize() before any other request
-            try:
-                await asyncio.wait_for(
-                    self._session.initialize(),
-                    timeout=self.connect_timeout,
-                )
-            except asyncio.CancelledError:
-                # If initialization was cancelled (e.g., event loop closing),
-                # cleanup and re-raise - don't suppress cancellation
-                await self._cleanup_on_error()
-                raise
-            except BaseExceptionGroup as eg:
-                # Handle exception groups from anyio task groups
-                # Extract the actual meaningful error (not GeneratorExit)
-                actual_error = None
-                for exc in eg.exceptions:
-                    if isinstance(exc, Exception) and not isinstance(
-                        exc, GeneratorExit
-                    ):
-                        # Check if it's an HTTP error (like 401)
-                        error_msg = str(exc).lower()
-                        if "401" in error_msg or "unauthorized" in error_msg:
-                            actual_error = exc
-                            break
-                        if "cancel scope" not in error_msg and "task" not in error_msg:
-                            actual_error = exc
-                            break
-
-                await self._cleanup_on_error()
-                if actual_error:
-                    raise ConnectionError(
-                        f"Failed to connect to MCP server: {actual_error}"
-                    ) from actual_error
-                raise ConnectionError(f"Failed to connect to MCP server: {eg}") from eg
+            # MCP protocol requires session.initialize() before any other request.
+            # Failures propagate to the handlers below, which unwind the transport
+            # once and inspect what that unwinding reveals about the cause.
+            await asyncio.wait_for(
+                self._session.initialize(),
+                timeout=self.connect_timeout,
+            )
 
             self._initialized = True
             self._was_connected = True
@@ -253,7 +239,12 @@ async def connect(self) -> Self:
             )
             raise ImportError(error_msg) from e
         except asyncio.TimeoutError as e:
-            await self._cleanup_on_error()
+            cleanup_error = await self._cleanup_on_error()
+            status_code = find_http_status(e, cleanup_error)
+            if status_code is not None:
+                raise self._report_connection_failure(
+                    server_info, started_at, status_code=status_code
+                ) from e
             error_msg = f"MCP connection timed out after {self.connect_timeout} seconds. The server may be slow or unreachable."
             self._emit_connection_failed(
                 server_name,
@@ -263,10 +254,21 @@ async def connect(self) -> Self:
                 "timeout",
                 started_at,
             )
-            raise ConnectionError(error_msg) from e
-        except asyncio.CancelledError:
-            # Re-raise cancellation - don't suppress it
-            await self._cleanup_on_error()
+            raise MCPConnectionError(error_msg) from e
+        except asyncio.CancelledError as e:
+            # A failing transport cancels this coroutine, so cancellation alone
+            # says nothing. Unwinding the transport is what reveals the cause.
+            cleanup_error = await self._cleanup_on_error()
+            status_code = find_http_status(e, cleanup_error)
+            if status_code is not None:
+                raise self._report_connection_failure(
+                    server_info, started_at, status_code=status_code
+                ) from e
+            if (failure := find_transport_failure(cleanup_error)) is not None:
+                raise self._report_connection_failure(
+                    server_info, started_at, error=failure
+                ) from failure
+            # Nothing failed, so this is a real cancellation: never swallow it.
             self._emit_connection_failed(
                 server_name,
                 server_url,
@@ -276,54 +278,59 @@ async def connect(self) -> Self:
                 started_at,
             )
             raise
-        except BaseExceptionGroup as eg:
-            # Handle exception groups from anyio task groups at outer level
-            actual_error = None
-            for exc in eg.exceptions:
-                if isinstance(exc, Exception) and not isinstance(exc, GeneratorExit):
-                    error_msg = str(exc).lower()
-                    if "401" in error_msg or "unauthorized" in error_msg:
-                        actual_error = exc
-                        break
-                    if "cancel scope" not in error_msg and "task" not in error_msg:
-                        actual_error = exc
-                        break
-
-            await self._cleanup_on_error()
-            error_type = (
-                "authentication"
-                if actual_error
-                and (
-                    "401" in str(actual_error).lower()
-                    or "unauthorized" in str(actual_error).lower()
-                )
-                else "network"
-            )
-            error_msg = str(actual_error) if actual_error else str(eg)
-            self._emit_connection_failed(
-                server_name,
-                server_url,
-                transport_type,
-                error_msg,
-                error_type,
-                started_at,
-            )
-            if actual_error:
-                raise ConnectionError(
-                    f"Failed to connect to MCP server: {actual_error}"
-                ) from actual_error
-            raise ConnectionError(f"Failed to connect to MCP server: {eg}") from eg
-        except Exception as e:
-            await self._cleanup_on_error()
+        except (BaseExceptionGroup, Exception) as e:
+            cleanup_error = await self._cleanup_on_error()
+            status_code = find_http_status(e, cleanup_error)
+            if status_code is not None:
+                raise self._report_connection_failure(
+                    server_info, started_at, status_code=status_code
+                ) from e
+            failure = find_transport_failure(e, cleanup_error) or e
+            raise self._report_connection_failure(
+                server_info, started_at, error=failure
+            ) from e
+
+    def _report_connection_failure(
+        self,
+        server_info: tuple[str, str | None, str | None],
+        started_at: datetime,
+        *,
+        error: BaseException | None = None,
+        status_code: int | None = None,
+    ) -> MCPConnectionError:
+        """Build a connection failure, emit the event, and return it to raise."""
+        if status_code is not None:
+            failure = error_for_status(status_code)
+            error_msg = str(failure)
+            error_type = error_type_for_status(status_code) or "network"
+        elif error is not None and isinstance(error, MCPConnectionError):
+            failure = error
+            error_msg = str(error)
             error_type = (
-                "authentication"
-                if "401" in str(e).lower() or "unauthorized" in str(e).lower()
+                error_type_for_status(error.status_code) or "network"
+                if error.status_code is not None
                 else "network"
             )
-            self._emit_connection_failed(
-                server_name, server_url, transport_type, str(e), error_type, started_at
-            )
-            raise ConnectionError(f"Failed to connect to MCP server: {e}") from e
+            status_code = error.status_code
+        elif error is not None:
+            error_msg = f"Failed to connect to MCP server: {error}"
+            error_type = "network"
+            status_code = find_http_status(error)
+            failure = MCPConnectionError(error_msg, status_code=status_code)
+        else:
+            raise ValueError("Either error or status_code must be provided")
+
+        server_name, server_url, transport_type = server_info
+        self._emit_connection_failed(
+            server_name,
+            server_url,
+            transport_type,
+            error_msg,
+            error_type,
+            started_at,
+            status_code=status_code,
+        )
+        return failure
 
     def _emit_connection_failed(
         self,
@@ -333,6 +340,7 @@ def _emit_connection_failed(
         error: str,
         error_type: str,
         started_at: datetime,
+        status_code: int | None = None,
     ) -> None:
         """Emit connection failed event."""
         failed_at = datetime.now()
@@ -344,19 +352,31 @@ def _emit_connection_failed(
                 transport_type=transport_type,
                 error=error,
                 error_type=error_type,
+                status_code=status_code,
                 started_at=started_at,
                 failed_at=failed_at,
             ),
         )
 
-    async def _cleanup_on_error(self) -> None:
-        """Cleanup resources when an error occurs during connection."""
+    async def _cleanup_on_error(self) -> BaseException | None:
+        """Cleanup resources when an error occurs during connection.
+
+        Returns:
+            The exception raised while unwinding the transport, if any. The
+            transport reports the server's refusal here rather than to the
+            coroutine that was waiting, so the caller inspects this for an
+            HTTP status instead of treating it as a cleanup problem.
+        """
         try:
             await self._exit_stack.aclose()
-
-        except Exception as e:
-            # Best effort cleanup - ignore all other errors
-            raise RuntimeError(f"Error during MCP client cleanup: {e}") from e
+        except asyncio.CancelledError as e:
+            return e
+        except (Exception, BaseExceptionGroup) as e:
+            # Groups are caught explicitly because one holding only BaseExceptions
+            # is not an Exception, yet can still carry the server's refusal.
+            return e
+        else:
+            return None
         finally:
             self._session = None
             self._initialized = False
@@ -369,7 +389,11 @@ async def disconnect(self) -> None:
 
         try:
             await self._exit_stack.aclose()
-        except Exception as e:
+        except asyncio.CancelledError:
+            raise
+        except MCPConnectionError:
+            raise
+        except (Exception, BaseExceptionGroup) as e:
             raise RuntimeError(f"Error during MCP client disconnect: {e}") from e
         finally:
             self._session = None
@@ -520,12 +544,6 @@ async def call_tool_result(
             return tool_result
         except Exception as e:
             failed_at = datetime.now()
-            error_type = (
-                "timeout"
-                if isinstance(e, (asyncio.TimeoutError, ConnectionError))
-                and "timeout" in str(e).lower()
-                else "server_error"
-            )
             crewai_event_bus.emit(
                 self,
                 MCPToolExecutionFailedEvent(
@@ -535,7 +553,7 @@ async def call_tool_result(
                     tool_name=tool_name,
                     tool_args=cleaned_arguments,
                     error=str(e),
-                    error_type=error_type,
+                    error_type=tool_execution_error_type(e),
                     started_at=started_at,
                     failed_at=failed_at,
                 ),
@@ -717,9 +735,17 @@ async def _retry_operation(
                     raise ConnectionError(last_error) from e
 
             except Exception as e:
+                if isinstance(e, MCPConnectionError):
+                    raise
+
+                status_code = find_http_status(e)
+                if status_code is not None and (
+                    error_type_for_status(status_code) == "authentication"
+                ):
+                    raise error_for_status(status_code, detail=str(e)) from e
+
                 error_str = str(e).lower()
 
-                # Classify errors as retryable or non-retryable
                 if "authentication" in error_str or "unauthorized" in error_str:
                     raise ConnectionError(f"Authentication failed: {e}") from e