andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-08-10 08:21:54 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Fix MCP connection timeouts and disable tool call retries by default

Browse Source
This commit is contained in:
lorenzejay
2026-07-30 14:18:22 -07:00
parent f15844b219
commit 2f6591d590
11 changed files with 249 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
lib/crewai-tools/tests/adapters/mcp_adapter_test.py
View File

@@ -1,3 +1,4 @@
import sys
from textwrap import dedent
from unittest.mock import MagicMock, patch
@@ -60,7 +61,7 @@ def echo_sse_server(echo_server_sse_script):
# Start the SSE server process with its own process group
process = subprocess.Popen(
["python", "-c", echo_server_sse_script],
[sys.executable, "-c", echo_server_sse_script],
)
# Give the server a moment to start up
@@ -76,7 +77,7 @@ def echo_sse_server(echo_server_sse_script):
def test_context_manager_syntax(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
with MCPServerAdapter(serverparams) as tools:
assert isinstance(tools, ToolCollection)
@@ -99,7 +100,7 @@ def test_context_manager_syntax_sse(echo_sse_server):
def test_try_finally_syntax(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
try:
mcp_server_adapter = MCPServerAdapter(serverparams)
@@ -129,7 +130,7 @@ def test_try_finally_syntax_sse(echo_sse_server):
def test_context_manager_with_filtered_tools(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
# Only select the echo_tool
with MCPServerAdapter(serverparams, "echo_tool") as tools:
@@ -159,7 +160,7 @@ def test_context_manager_sse_with_filtered_tools(echo_sse_server):
def test_try_finally_with_filtered_tools(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
try:
# Select both tools but in reverse order
@@ -176,7 +177,7 @@ def test_try_finally_with_filtered_tools(echo_server_script):
def test_filter_with_nonexistent_tool(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
# Include a tool that doesn't exist
with MCPServerAdapter(serverparams, "echo_tool", "nonexistent_tool") as tools:
@@ -187,7 +188,7 @@ def test_filter_with_nonexistent_tool(echo_server_script):
def test_filter_with_only_nonexistent_tools(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
# All requested tools don't exist
with MCPServerAdapter(serverparams, "nonexistent1", "nonexistent2") as tools:
@@ -198,7 +199,7 @@ def test_filter_with_only_nonexistent_tools(echo_server_script):
def test_connect_timeout_parameter(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
with MCPServerAdapter(serverparams, connect_timeout=60) as tools:
assert isinstance(tools, ToolCollection)
@@ -210,7 +211,7 @@ def test_connect_timeout_parameter(echo_server_script):
def test_connect_timeout_with_filtered_tools(echo_server_script):
serverparams = StdioServerParameters(
command="uv", args=["run", "python", "-c", echo_server_script]
command=sys.executable, args=["-c", echo_server_script]
)
with MCPServerAdapter(serverparams, "echo_tool", connect_timeout=45) as tools:
assert isinstance(tools, ToolCollection)

1
lib/crewai/pyproject.toml
View File

@@ -10,6 +10,7 @@ requires-python = ">=3.10, <3.14"
dependencies = [
"crewai-core==1.15.8",
"crewai-cli==1.15.8",
"async-timeout>=4,<6; python_version < '3.11'",
# Core Dependencies
"pydantic>=2.11.9,<2.13",
"openai>=2.30.0,<3",

12
lib/crewai/src/crewai/mcp/_utils.py Normal file
View File

@@ -0,0 +1,12 @@
"""Internal compatibility helpers for MCP async lifecycle management."""
import sys
if sys.version_info >= (3, 11):
from asyncio import timeout as async_timeout
else:
from async_timeout import timeout as async_timeout
__all__ = ["async_timeout"]

61
lib/crewai/src/crewai/mcp/client.py
View File

@@ -26,6 +26,7 @@ from crewai.events.types.mcp_events import (
MCPToolExecutionFailedEvent,
MCPToolExecutionStartedEvent,
)
from crewai.mcp._utils import async_timeout
from crewai.mcp.transports.base import BaseTransport
from crewai.mcp.transports.http import HTTPTransport
from crewai.mcp.transports.sse import SSETransport
@@ -76,6 +77,7 @@ class MCPClient:
max_retries: int = MCP_MAX_RETRIES,
cache_tools_list: bool = False,
logger: logging.Logger | None = None,
retry_tool_calls: bool = False,
) -> None:
"""Initialize MCP client.
@@ -87,13 +89,20 @@ class MCPClient:
max_retries: Maximum retry attempts for operations.
cache_tools_list: Whether to cache tool list results.
logger: Optional logger instance.
retry_tool_calls: Whether to retry failed tool calls. Disabled by
default because replaying a tool can duplicate external side
effects when the server committed work before the response was
lost.
"""
self.transport = transport
self.connect_timeout = connect_timeout
self.transport.connect_timeout = connect_timeout
self.execution_timeout = execution_timeout
self.discovery_timeout = discovery_timeout
self.max_retries = max_retries
self.cache_tools_list = cache_tools_list
self.retry_tool_calls = retry_tool_calls
self._logger = logger or logging.getLogger(__name__)
self._session: Any = None
self._initialized = False
self._exit_stack = AsyncExitStack()
@@ -167,24 +176,26 @@ class MCPClient:
try:
from mcp import ClientSession
# Use AsyncExitStack to manage transport and session contexts together
# This ensures they're in the same async scope and prevents cancel scope errors
# Always enter transport context via exit stack (it handles already-connected state)
await self._exit_stack.enter_async_context(self.transport)
self._session = ClientSession(
self.transport.read_stream,
self.transport.write_stream,
)
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,
)
# Each transport bounds its own context entry so a timeout can
# unwind partially-created SDK resources before returning.
connect_started = asyncio.get_running_loop().time()
await self._exit_stack.enter_async_context(self.transport)
elapsed = asyncio.get_running_loop().time() - connect_started
remaining_timeout = max(self.connect_timeout - elapsed, 0)
# Keep session startup in this task. anyio cancel scopes must
# be exited by the task that entered them.
async with async_timeout(remaining_timeout):
self._session = ClientSession(
self.transport.read_stream,
self.transport.write_stream,
)
await self._exit_stack.enter_async_context(self._session)
# MCP requires initialize() before any other request.
await self._session.initialize()
except asyncio.CancelledError:
# If initialization was cancelled (e.g., event loop closing),
# cleanup and re-raise - don't suppress cancellation
@@ -350,8 +361,9 @@ class MCPClient:
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
# Do not replace the connection failure with a secondary cleanup
# failure. Preserve the root cause and leave cleanup observable.
self._logger.warning("Error during MCP client cleanup: %s", e)
finally:
self._session = None
self._initialized = False
@@ -453,10 +465,13 @@ class MCPClient:
)
try:
tool_result: _MCPToolResult = await self._retry_operation(
lambda: self._call_tool_impl(tool_name, cleaned_arguments),
timeout=self.execution_timeout,
)
if self.retry_tool_calls:
tool_result: _MCPToolResult = await self._retry_operation(
lambda: self._call_tool_impl(tool_name, cleaned_arguments),
timeout=self.execution_timeout,
)
else:
tool_result = await self._call_tool_impl(tool_name, cleaned_arguments)
finished_at = datetime.now()
execution_duration_ms = (finished_at - started_at).total_seconds() * 1000

4
lib/crewai/src/crewai/mcp/transports/base.py
View File

@@ -30,12 +30,14 @@ class BaseTransport(ABC):
with MCP servers.
"""
def __init__(self, **kwargs: Any) -> None:
def __init__(self, connect_timeout: float = 30.0, **kwargs: Any) -> None:
"""Initialize the transport.
Args:
connect_timeout: Maximum seconds allowed for transport startup.
**kwargs: Transport-specific configuration options.
"""
self.connect_timeout = connect_timeout
self._read_stream: MCPReadStream | None = None
self._write_stream: MCPWriteStream | None = None
self._connected = False

13
lib/crewai/src/crewai/mcp/transports/http.py
View File

@@ -12,6 +12,7 @@ if sys.version_info >= (3, 11):
else:
from exceptiongroup import BaseExceptionGroup
from crewai.mcp._utils import async_timeout
from crewai.mcp.transports.base import BaseTransport, TransportType
@@ -81,13 +82,15 @@ class HTTPTransport(BaseTransport):
)
try:
read, write, _ = await asyncio.wait_for(
self._transport_context.__aenter__(), timeout=30.0
)
except asyncio.TimeoutError as e:
# Enter and exit the SDK's anyio cancel scope in the same task.
# asyncio.wait_for() runs __aenter__ in a child task and can
# later trigger "cancel scope in a different task" failures.
async with async_timeout(self.connect_timeout):
read, write, _ = await self._transport_context.__aenter__()
except TimeoutError as e:
self._transport_context = None
raise ConnectionError(
"Transport context entry timed out after 30 seconds. "
f"Transport context entry timed out after {self.connect_timeout} seconds. "
"Server may be slow or unreachable."
) from e
except Exception as e:

4
lib/crewai/src/crewai/mcp/transports/sse.py
View File

@@ -4,6 +4,7 @@ from typing import Any
from typing_extensions import Self
from crewai.mcp._utils import async_timeout
from crewai.mcp.transports.base import BaseTransport, TransportType
@@ -68,7 +69,8 @@ class SSETransport(BaseTransport):
headers=self.headers if self.headers else None,
)
read, write = await self._transport_context.__aenter__()
async with async_timeout(self.connect_timeout):
read, write = await self._transport_context.__aenter__()
self._set_streams(read=read, write=write)

4
lib/crewai/src/crewai/mcp/transports/stdio.py
View File

@@ -7,6 +7,7 @@ from typing import Any
from typing_extensions import Self
from crewai.mcp._utils import async_timeout
from crewai.mcp.transports.base import BaseTransport, TransportType
@@ -97,7 +98,8 @@ class StdioTransport(BaseTransport):
self._transport_context = stdio_client(server_params)
try:
read, write = await self._transport_context.__aenter__()
async with async_timeout(self.connect_timeout):
read, write = await self._transport_context.__aenter__()
except Exception as e:
import traceback

163
lib/crewai/tests/mcp/test_client.py Normal file
View File

@@ -0,0 +1,163 @@
"""MCP client connection lifecycle and retry-safety tests."""
from __future__ import annotations
import asyncio
from contextlib import AbstractAsyncContextManager
import logging
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from crewai.events.event_bus import crewai_event_bus
from crewai.mcp.client import MCPClient
from crewai.mcp.transports.base import BaseTransport, TransportType
from crewai.mcp.transports.http import HTTPTransport
class ConnectedTransport(BaseTransport):
"""Minimal connected transport for client operation tests."""
@property
def transport_type(self) -> TransportType:
return TransportType.HTTP
async def connect(self) -> ConnectedTransport:
self._connected = True
return self
async def disconnect(self) -> None:
self._connected = False
async def __aenter__(self) -> ConnectedTransport:
return await self.connect()
async def __aexit__(self, *_args: Any) -> None:
await self.disconnect()
class FailingSession:
"""Session that models a lost response after a side effect committed."""
def __init__(self) -> None:
self.calls = 0
async def call_tool(self, _name: str, _arguments: dict[str, Any]) -> None:
self.calls += 1
raise ConnectionError("connection dropped after commit")
@pytest.mark.asyncio
async def test_connect_timeout_bounds_transport_startup():
"""connect_timeout must include transport context entry."""
class HangingContext(AbstractAsyncContextManager):
async def __aenter__(self):
await asyncio.Event().wait()
async def __aexit__(self, *_args: Any):
return None
client = MCPClient(
HTTPTransport("https://mcp.example.com"),
connect_timeout=1,
)
with (
patch(
"mcp.client.streamable_http.streamablehttp_client",
return_value=HangingContext(),
),
patch.object(crewai_event_bus, "emit"),
pytest.raises(ConnectionError, match="timed out after 1 seconds"),
):
await asyncio.wait_for(client.connect(), timeout=1.25)
@pytest.mark.asyncio
async def test_http_transport_context_enters_and_exits_in_same_task():
"""AnyIO transport cancel scopes must not cross asyncio tasks."""
entered_task = None
exited_task = None
class RecordingContext(AbstractAsyncContextManager):
async def __aenter__(self):
nonlocal entered_task
entered_task = asyncio.current_task()
return MagicMock(), MagicMock(), None
async def __aexit__(self, *_args: Any):
nonlocal exited_task
exited_task = asyncio.current_task()
transport = HTTPTransport("https://mcp.example.com")
with patch(
"mcp.client.streamable_http.streamablehttp_client",
return_value=RecordingContext(),
):
await transport.connect()
await transport.disconnect()
assert entered_task is exited_task
@pytest.mark.asyncio
async def test_tool_call_is_not_retried_by_default():
"""A lost response must not blindly replay a potentially mutating call."""
transport = ConnectedTransport()
await transport.connect()
client = MCPClient(transport, max_retries=3)
session = FailingSession()
client._session = session
client._initialized = True
with (
patch.object(crewai_event_bus, "emit"),
pytest.raises(ConnectionError, match="connection dropped after commit"),
):
await client.call_tool("mutating_tool")
assert session.calls == 1
@pytest.mark.asyncio
async def test_tool_call_retries_require_explicit_opt_in():
"""Callers can explicitly accept replay risk for idempotent tools."""
transport = ConnectedTransport()
await transport.connect()
client = MCPClient(
transport,
max_retries=3,
retry_tool_calls=True,
)
session = FailingSession()
client._session = session
client._initialized = True
with (
patch.object(crewai_event_bus, "emit"),
patch("crewai.mcp.client.asyncio.sleep", new=AsyncMock()),
pytest.raises(ConnectionError, match="failed after 3 attempts"),
):
await client.call_tool("idempotent_tool")
assert session.calls == 3
@pytest.mark.asyncio
async def test_cleanup_failure_is_logged_without_masking_connection_error():
"""Best-effort cleanup must preserve the original connection exception."""
logger = MagicMock(spec=logging.Logger)
client = MCPClient(ConnectedTransport(), logger=logger)
async def fail_cleanup() -> None:
raise RuntimeError("cleanup failed")
client._exit_stack.push_async_callback(fail_cleanup)
await client._cleanup_on_error()
logger.warning.assert_called_once()
log_template, log_error = logger.warning.call_args.args
assert log_template == "Error during MCP client cleanup: %s"
assert str(log_error) == "cleanup failed"

8
lib/crewai/tests/mcp/test_tool_resolver_native.py
View File

@@ -93,7 +93,11 @@ class TestResolveNativeRuntimeError:
def test_unmatched_runtime_error_is_wrapped_not_swallowed(
self, mock_asyncio_run, resolver, http_config
):
mock_asyncio_run.side_effect = RuntimeError("some other failure")
def fail_after_closing(coroutine):
coroutine.close()
raise RuntimeError("some other failure")
mock_asyncio_run.side_effect = fail_after_closing
with pytest.raises(RuntimeError, match="Failed to get native MCP tools"):
resolver._resolve_native(http_config)
resolver._resolve_native(http_config)