diff --git a/lib/crewai-tools/tests/adapters/mcp_adapter_test.py b/lib/crewai-tools/tests/adapters/mcp_adapter_test.py
index 6f67878dd..15c517b36 100644
--- a/lib/crewai-tools/tests/adapters/mcp_adapter_test.py
+++ b/lib/crewai-tools/tests/adapters/mcp_adapter_test.py
@@ -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)
diff --git a/lib/crewai/pyproject.toml b/lib/crewai/pyproject.toml
index 1549c35cd..fe5770093 100644
--- a/lib/crewai/pyproject.toml
+++ b/lib/crewai/pyproject.toml
@@ -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",
diff --git a/lib/crewai/src/crewai/mcp/_utils.py b/lib/crewai/src/crewai/mcp/_utils.py
new file mode 100644
index 000000000..ea4f82ef7
--- /dev/null
+++ b/lib/crewai/src/crewai/mcp/_utils.py
@@ -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"]
diff --git a/lib/crewai/src/crewai/mcp/client.py b/lib/crewai/src/crewai/mcp/client.py
index adf1afb8c..cbf93289d 100644
--- a/lib/crewai/src/crewai/mcp/client.py
+++ b/lib/crewai/src/crewai/mcp/client.py
@@ -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
diff --git a/lib/crewai/src/crewai/mcp/transports/base.py b/lib/crewai/src/crewai/mcp/transports/base.py
index d590d3e31..85e7db838 100644
--- a/lib/crewai/src/crewai/mcp/transports/base.py
+++ b/lib/crewai/src/crewai/mcp/transports/base.py
@@ -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
diff --git a/lib/crewai/src/crewai/mcp/transports/http.py b/lib/crewai/src/crewai/mcp/transports/http.py
index 7b05ee78d..c1137b6ce 100644
--- a/lib/crewai/src/crewai/mcp/transports/http.py
+++ b/lib/crewai/src/crewai/mcp/transports/http.py
@@ -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:
diff --git a/lib/crewai/src/crewai/mcp/transports/sse.py b/lib/crewai/src/crewai/mcp/transports/sse.py
index c2184e7d0..b7abcf3ab 100644
--- a/lib/crewai/src/crewai/mcp/transports/sse.py
+++ b/lib/crewai/src/crewai/mcp/transports/sse.py
@@ -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)
 
diff --git a/lib/crewai/src/crewai/mcp/transports/stdio.py b/lib/crewai/src/crewai/mcp/transports/stdio.py
index 8701ca668..c90424212 100644
--- a/lib/crewai/src/crewai/mcp/transports/stdio.py
+++ b/lib/crewai/src/crewai/mcp/transports/stdio.py
@@ -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
 
diff --git a/lib/crewai/tests/mcp/test_client.py b/lib/crewai/tests/mcp/test_client.py
new file mode 100644
index 000000000..2e7df9573
--- /dev/null
+++ b/lib/crewai/tests/mcp/test_client.py
@@ -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"
diff --git a/lib/crewai/tests/mcp/test_tool_resolver_native.py b/lib/crewai/tests/mcp/test_tool_resolver_native.py
index 7fe2ed5be..bce5c0bb3 100644
--- a/lib/crewai/tests/mcp/test_tool_resolver_native.py
+++ b/lib/crewai/tests/mcp/test_tool_resolver_native.py
@@ -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)
\ No newline at end of file
+            resolver._resolve_native(http_config)
diff --git a/uv.lock b/uv.lock
index 491e5e797..7d539a3f6 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1325,6 +1325,7 @@ dependencies = [
     { name = "aiofiles" },
     { name = "aiosqlite" },
     { name = "appdirs" },
+    { name = "async-timeout", marker = "python_full_version < '3.11'" },
     { name = "cel-python" },
     { name = "chromadb" },
     { name = "click" },
@@ -1428,6 +1429,7 @@ requires-dist = [
     { name = "aiosqlite", specifier = "~=0.21.0" },
     { name = "anthropic", marker = "extra == 'anthropic'", specifier = "~=0.73.0" },
     { name = "appdirs", specifier = "~=1.4.4" },
+    { name = "async-timeout", marker = "python_full_version < '3.11'", specifier = ">=4,<6" },
     { name = "azure-ai-inference", marker = "extra == 'azure-ai-inference'", specifier = "~=1.0.0b9" },
     { name = "azure-identity", marker = "extra == 'azure-ai-inference'", specifier = ">=1.17.0,<2" },
     { name = "boto3", marker = "extra == 'aws'", specifier = "~=1.43.46" },