From d0d25475c482a4ae048d3187974304742e71e299 Mon Sep 17 00:00:00 2001 From: lorenzejay Date: Thu, 30 Jul 2026 20:47:33 -0700 Subject: [PATCH] Preserve MCP transport timeouts and cleanup failures --- lib/crewai/src/crewai/mcp/client.py | 4 + lib/crewai/src/crewai/mcp/transports/http.py | 28 +++- lib/crewai/src/crewai/mcp/transports/sse.py | 15 ++- lib/crewai/src/crewai/mcp/transports/stdio.py | 13 +- lib/crewai/tests/mcp/test_client.py | 122 +++++++++++++++++- 5 files changed, 171 insertions(+), 11 deletions(-) diff --git a/lib/crewai/src/crewai/mcp/client.py b/lib/crewai/src/crewai/mcp/client.py index 6199f2bd5..2fc156506 100644 --- a/lib/crewai/src/crewai/mcp/client.py +++ b/lib/crewai/src/crewai/mcp/client.py @@ -77,6 +77,10 @@ def _find_meaningful_exception(error: BaseException | None) -> Exception | None: def _connection_error_type(error: BaseException) -> str: """Classify an MCP connection error for emitted events.""" message = str(error).lower() + if isinstance(error, (TimeoutError, asyncio.TimeoutError)) or any( + marker in message for marker in ("timed out", "timeout") + ): + return "timeout" if "401" in message or "unauthorized" in message: return "authentication" if "certificate verify failed" in message or "ssl" in message: diff --git a/lib/crewai/src/crewai/mcp/transports/http.py b/lib/crewai/src/crewai/mcp/transports/http.py index ac710500d..86dae5a41 100644 --- a/lib/crewai/src/crewai/mcp/transports/http.py +++ b/lib/crewai/src/crewai/mcp/transports/http.py @@ -1,6 +1,7 @@ """HTTP and Streamable HTTP transport for MCP servers.""" import asyncio +import logging from typing import Any from typing_extensions import Self @@ -9,6 +10,9 @@ from crewai.mcp._utils import async_timeout from crewai.mcp.transports.base import BaseTransport, TransportType +logger = logging.getLogger(__name__) + + class HTTPTransport(BaseTransport): """HTTP/Streamable HTTP transport for connecting to remote MCP servers. @@ -84,7 +88,7 @@ class HTTPTransport(BaseTransport): # it is distinct from the built-in TimeoutError. except (TimeoutError, asyncio.TimeoutError) as e: self._transport_context = None - raise ConnectionError( + raise asyncio.TimeoutError( f"Transport context entry timed out after {self.connect_timeout} seconds. " "Server may be slow or unreachable." ) from e @@ -98,6 +102,10 @@ class HTTPTransport(BaseTransport): raise ImportError( "MCP library not available. Please install with: pip install mcp" ) from e + except (TimeoutError, asyncio.TimeoutError): + self._clear_streams() + self._transport_context = None + raise except Exception as e: self._clear_streams() if self._transport_context is not None: @@ -109,6 +117,8 @@ class HTTPTransport(BaseTransport): exc_type: type[BaseException] | None = None, exc_val: BaseException | None = None, exc_tb: Any = None, + *, + suppress_errors: bool = False, ) -> None: """Close the SDK context with the exception that triggered unwinding.""" if not self._connected: @@ -118,11 +128,16 @@ class HTTPTransport(BaseTransport): transport_context = self._transport_context self._transport_context = None if transport_context is not None: - await transport_context.__aexit__(exc_type, exc_val, exc_tb) + try: + await transport_context.__aexit__(exc_type, exc_val, exc_tb) + except Exception as e: + if not suppress_errors: + raise + logger.warning("Error during HTTP transport disconnect: %s", e) async def disconnect(self) -> None: """Close HTTP connection.""" - await self._disconnect() + await self._disconnect(suppress_errors=True) async def __aenter__(self) -> Self: """Async context manager entry.""" @@ -135,4 +150,9 @@ class HTTPTransport(BaseTransport): exc_tb: Any, ) -> None: """Async context manager exit.""" - await self._disconnect(exc_type, exc_val, exc_tb) + await self._disconnect( + exc_type, + exc_val, + exc_tb, + suppress_errors=exc_type is None, + ) diff --git a/lib/crewai/src/crewai/mcp/transports/sse.py b/lib/crewai/src/crewai/mcp/transports/sse.py index b7abcf3ab..8663ba6d3 100644 --- a/lib/crewai/src/crewai/mcp/transports/sse.py +++ b/lib/crewai/src/crewai/mcp/transports/sse.py @@ -1,5 +1,6 @@ """Server-Sent Events (SSE) transport for MCP servers.""" +import asyncio from typing import Any from typing_extensions import Self @@ -80,8 +81,16 @@ class SSETransport(BaseTransport): raise ImportError( "MCP library not available. Please install with: pip install mcp" ) from e + except (TimeoutError, asyncio.TimeoutError) as e: + self._clear_streams() + self._transport_context = None + raise asyncio.TimeoutError( + f"SSE transport context entry timed out after {self.connect_timeout} seconds. " + "Server may be slow or unreachable." + ) from e except Exception as e: self._clear_streams() + self._transport_context = None raise ConnectionError(f"Failed to connect to SSE MCP server: {e}") from e async def disconnect(self) -> None: @@ -91,8 +100,10 @@ class SSETransport(BaseTransport): try: self._clear_streams() - if self._transport_context is not None: - await self._transport_context.__aexit__(None, None, None) + transport_context = self._transport_context + self._transport_context = None + if transport_context is not None: + await transport_context.__aexit__(None, None, None) except Exception as e: import logging diff --git a/lib/crewai/src/crewai/mcp/transports/stdio.py b/lib/crewai/src/crewai/mcp/transports/stdio.py index c90424212..0242163aa 100644 --- a/lib/crewai/src/crewai/mcp/transports/stdio.py +++ b/lib/crewai/src/crewai/mcp/transports/stdio.py @@ -100,10 +100,13 @@ class StdioTransport(BaseTransport): try: async with async_timeout(self.connect_timeout): read, write = await self._transport_context.__aenter__() + except (TimeoutError, asyncio.TimeoutError) as e: + self._transport_context = None + raise asyncio.TimeoutError( + f"Stdio transport context entry timed out after {self.connect_timeout} seconds. " + "Server may be slow or unreachable." + ) from e except Exception as e: - import traceback - - traceback.print_exc() self._transport_context = None raise ConnectionError( f"Failed to enter stdio transport context: {e}" @@ -117,6 +120,10 @@ class StdioTransport(BaseTransport): raise ImportError( "MCP library not available. Please install with: pip install mcp" ) from e + except (TimeoutError, asyncio.TimeoutError): + self._clear_streams() + self._transport_context = None + raise except Exception as e: self._clear_streams() if self._transport_context is not None: diff --git a/lib/crewai/tests/mcp/test_client.py b/lib/crewai/tests/mcp/test_client.py index e72fc2dfa..672b4d3b8 100644 --- a/lib/crewai/tests/mcp/test_client.py +++ b/lib/crewai/tests/mcp/test_client.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -from contextlib import AbstractAsyncContextManager +from contextlib import AbstractAsyncContextManager, ExitStack import logging import sys from typing import Any @@ -16,6 +16,8 @@ from crewai.events.types.mcp_events import MCPConnectionFailedEvent from crewai.mcp.client import MCPClient from crewai.mcp.transports.base import BaseTransport, TransportType from crewai.mcp.transports.http import HTTPTransport +from crewai.mcp.transports.sse import SSETransport +from crewai.mcp.transports.stdio import StdioTransport if sys.version_info >= (3, 11): @@ -53,6 +55,13 @@ class LifecycleTransport(ConnectedTransport): return self +class WrappedTimeoutTransport(ConnectedTransport): + """Transport that exposes a timeout through a generic connection error.""" + + async def connect(self) -> WrappedTimeoutTransport: + raise ConnectionError("Transport context entry timed out after 1 seconds") + + class FailingSession: """Session that models a lost response after a side effect committed.""" @@ -75,6 +84,7 @@ async def test_connect_timeout_bounds_transport_startup(): async def __aexit__(self, *_args: Any): return None + emitted_events = [] client = MCPClient( HTTPTransport("https://mcp.example.com"), connect_timeout=1, @@ -85,11 +95,119 @@ async def test_connect_timeout_bounds_transport_startup(): "mcp.client.streamable_http.streamablehttp_client", return_value=HangingContext(), ), - patch.object(crewai_event_bus, "emit"), + patch.object( + crewai_event_bus, + "emit", + side_effect=lambda _source, event: emitted_events.append(event), + ), pytest.raises(ConnectionError, match="timed out after 1 seconds"), ): await asyncio.wait_for(client.connect(), timeout=1.25) + failed_event = next( + event for event in emitted_events if isinstance(event, MCPConnectionFailedEvent) + ) + assert failed_event.error_type == "timeout" + + +@pytest.mark.asyncio +async def test_wrapped_transport_timeout_event_is_classified_as_timeout(): + """Timeout text preserved in ConnectionError must remain observable.""" + emitted_events = [] + client = MCPClient(WrappedTimeoutTransport()) + + with ( + patch.object( + crewai_event_bus, + "emit", + side_effect=lambda _source, event: emitted_events.append(event), + ), + pytest.raises(ConnectionError, match="timed out after 1 seconds"), + ): + await client.connect() + + failed_event = next( + event for event in emitted_events if isinstance(event, MCPConnectionFailedEvent) + ) + assert failed_event.error_type == "timeout" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("transport", "client_path"), + [ + ( + SSETransport("https://mcp.example.com/sse", connect_timeout=0.01), + "mcp.client.sse.sse_client", + ), + ( + StdioTransport("python", connect_timeout=0.01), + "mcp.client.stdio.stdio_client", + ), + ], +) +async def test_transport_startup_timeout_is_preserved( + transport: BaseTransport, + client_path: str, +): + """SSE and stdio startup timeouts must remain timeout exceptions.""" + + class HangingContext(AbstractAsyncContextManager): + def __init__(self) -> None: + self.cancelled = False + + async def __aenter__(self): + try: + await asyncio.Event().wait() + finally: + self.cancelled = True + + async def __aexit__(self, *_args: Any): + return None + + context = HangingContext() + patches = [patch(client_path, return_value=context)] + if isinstance(transport, StdioTransport): + patches.extend( + [ + patch("mcp.StdioServerParameters"), + patch("mcp.client.stdio.get_default_environment", return_value={}), + ] + ) + + with ExitStack() as stack: + for context_patch in patches: + stack.enter_context(context_patch) + with pytest.raises(asyncio.TimeoutError, match="timed out"): + await transport.connect() + + assert transport.connected is False + assert transport._transport_context is None + assert context.cancelled is True + + +@pytest.mark.asyncio +async def test_http_disconnect_failure_is_best_effort(caplog: pytest.LogCaptureFixture): + """Normal HTTP cleanup failures must not replace a successful result.""" + + class FailingExitContext(AbstractAsyncContextManager): + async def __aenter__(self): + return MagicMock(), MagicMock(), None + + async def __aexit__(self, *_args: Any): + raise RuntimeError("cleanup failed") + + transport = HTTPTransport("https://mcp.example.com") + with patch( + "mcp.client.streamable_http.streamablehttp_client", + return_value=FailingExitContext(), + ): + await transport.connect() + await transport.disconnect() + + assert transport.connected is False + assert "Error during HTTP transport disconnect: cleanup failed" in caplog.text + @pytest.mark.asyncio async def test_http_transport_context_enters_and_exits_in_same_task():