diff --git a/cecli/mcp/server.py b/cecli/mcp/server.py
index f148e47bd87..25638c840e0 100644
--- a/cecli/mcp/server.py
+++ b/cecli/mcp/server.py
@@ -1,8 +1,10 @@
 import asyncio
 import logging
 import os
+import random
 import webbrowser
 from contextlib import AsyncExitStack
+from enum import Enum, auto
 from urllib.parse import urlparse
 
 import httpx
@@ -20,6 +22,16 @@
     save_mcp_oauth_token,
 )
 
+MIN_KEEPALIVE_INTERVAL = 5
+MAX_KEEPALIVE_INTERVAL = 300
+FAILED_PING_THRESHOLD = 3
+
+
+class ConnectionState(Enum):
+    CONNECTED = auto()
+    UNHEALTHY = auto()
+    DISCONNECTED = auto()
+
 
 class McpServer:
     """
@@ -120,6 +132,13 @@ async def disconnect(self):
 class HttpBasedMcpServer(McpServer):
     """Base class for HTTP-based MCP servers (HTTP streaming and SSE)."""
 
+    def __init__(self, server_config, io=None, verbose=False):
+        super().__init__(server_config, io, verbose)
+        self._state: ConnectionState = ConnectionState.CONNECTED
+        self._failed_pings: int = 0
+        self._keepalive_task: asyncio.Task | None = None
+        self._http_client: httpx.AsyncClient | None = None
+
     async def _create_oauth_provider(self):
         """Create an OAuthClientProvider using the MCP SDK."""
         parsed = urlparse(self.config.get("url"))
@@ -228,6 +247,7 @@ async def connect(self):
                     timeout=30,
                 )
             )
+            self._http_client = http_client
 
             transport = await self.exit_stack.enter_async_context(
                 self._create_transport(url, http_client=http_client)
@@ -238,6 +258,7 @@ async def connect(self):
             session = await self.exit_stack.enter_async_context(ClientSession(read, write))
             await session.initialize()
             self.session = session
+            await self.start_keepalive()
             self._connection_loop = current_loop
 
             if oauth_provider.context.oauth_metadata:
@@ -256,10 +277,119 @@ async def connect(self):
             await self.disconnect()
             raise
 
-    async def disconnect(self):
+    async def start_keepalive(self):
+        """Start the background keepalive loop if configured."""
+        interval = self.config.get("keepalive_interval", 300)
+
+        try:
+            interval = int(interval)
+            if not (MIN_KEEPALIVE_INTERVAL <= interval <= MAX_KEEPALIVE_INTERVAL):
+                if self.verbose and self.io:
+                    self.io.tool_warning(
+                        f"Keepalive interval {interval} out of range ({MIN_KEEPALIVE_INTERVAL}-"
+                        f"{MAX_KEEPALIVE_INTERVAL}). Ignoring."
+                    )
+                return
+        except (ValueError, TypeError):
+            if self.verbose and self.io:
+                self.io.tool_warning(f"Invalid keepalive interval {interval}. Must be an integer.")
+            return
+
+        if self._keepalive_task and not self._keepalive_task.done():
+            self._keepalive_task.cancel()
+
+        self._keepalive_task = asyncio.create_task(self._keepalive_loop(interval))
+        if self.verbose and self.io:
+            self.io.tool_output(f"Started keepalive loop for {self.name} (interval: {interval}s)")
+
+    async def _keepalive_loop(self, interval: int):
+        """Background loop that sends periodic heartbeats to the MCP server."""
+        try:
+            while True:
+                # Jitter: ±10% to prevent timing analysis
+                jitter = interval * 0.1 * (2 * random.random() - 1)
+                await asyncio.sleep(interval + jitter)
+
+                if not self._http_client:
+                    continue
+
+                try:
+                    url = self.config.get("url")
+                    headers = self.config.get("headers", {})
+
+                    # Use OPTIONS request as a lightweight heartbeat
+                    response = await self._http_client.options(url, headers=headers)
+                    if response.status_code == 200:
+                        self._state = ConnectionState.CONNECTED
+                        self._failed_pings = 0
+                    else:
+                        raise httpx.HTTPStatusError(
+                            f"Unexpected status {response.status_code}",
+                            request=response.request,
+                            response=response,
+                        )
+                except Exception:
+                    self._failed_pings += 1
+                    if self._failed_pings >= FAILED_PING_THRESHOLD:
+                        self._state = ConnectionState.DISCONNECTED
+                        if self.verbose and self.io:
+                            self.io.tool_warning(
+                                f"MCP server {self.name} disconnected after {self._failed_pings} failed"
+                                " pings. Attempting reconnect..."
+                            )
+                        await self.reconnect()
+                    else:
+                        self._state = ConnectionState.UNHEALTHY
+                        if self.verbose and self.io:
+                            self.io.tool_output(
+                                f"MCP server {self.name} unhealthy (ping {self._failed_pings}/{FAILED_PING_THRESHOLD})"
+                            )
+        except asyncio.CancelledError:
+            pass
+        except Exception as e:
+            logging.error(f"Keepalive loop for {self.name} crashed: {e}")
+
+    async def reconnect(self):
+        """Attempt to reconnect to the server using exponential backoff."""
+        initial_delay = 1
+        multiplier = 2
+        max_delay = 300
+
+        attempt = 0
+        while self._state == ConnectionState.DISCONNECTED:
+            delay = min(initial_delay * (multiplier**attempt), max_delay)
+            # Jitter: ±20%
+            jitter = delay * 0.2 * (2 * random.random() - 1)
+            await asyncio.sleep(delay + jitter)
+
+            try:
+                if self.verbose and self.io:
+                    self.io.tool_output(
+                        f"Attempting to reconnect to {self.name} (attempt {attempt + 1})..."
+                    )
+
+                # Clean up old session/client without cancelling the keepalive task
+                await self.disconnect(cancel_keepalive=False)
+                await self.connect()
+
+                self._state = ConnectionState.CONNECTED
+                self._failed_pings = 0
+                if self.verbose and self.io:
+                    self.io.tool_output(f"Successfully reconnected to {self.name}")
+                break
+            except Exception as e:
+                attempt += 1
+                if self.verbose and self.io:
+                    self.io.tool_warning(
+                        f"Reconnection attempt {attempt} failed for {self.name}: {e}"
+                    )
+
+    async def disconnect(self, cancel_keepalive: bool = True):
         """Disconnect from the MCP server and clean up resources."""
         async with self._cleanup_lock:
             try:
+                if cancel_keepalive and self._keepalive_task:
+                    self._keepalive_task.cancel()
                 if hasattr(self, "_oauth_shutdown"):
                     self._oauth_shutdown()
                 await self.exit_stack.aclose()
@@ -271,6 +401,7 @@ async def disconnect(self):
                 logging.error(f"Error during cleanup of server {self.name}: {e}")
             finally:
                 self.session = None
+                self._http_client = None
 
 
 class HttpStreamingServer(HttpBasedMcpServer):
diff --git a/cecli/website/docs/config/mcp.md b/cecli/website/docs/config/mcp.md
index eec92c4acaf..04a8f7a63c9 100644
--- a/cecli/website/docs/config/mcp.md
+++ b/cecli/website/docs/config/mcp.md
@@ -6,7 +6,7 @@ description: Configure Model Control Protocol (MCP) servers for enhanced AI capa
 
 # Model Control Protocol (MCP)
 
-Model Control Protocol (MCP) servers extend cecli's capabilities by providing additional tools and functionality to the AI models. MCP servers can add features like git operations, context retrieval, and other specialized tools.
+Model Control Protocol (MCP) servers extend the capabilities of cecli by providing additional tools and functionality to the AI models. MCP servers can add features like git operations, context retrieval, and other specialized tools.
 
 ## Configuring MCP Servers
 
@@ -14,6 +14,24 @@ cecli supports configuring MCP servers using the MCP Server Configuration schema
 see the [Model Context Protocol documentation](https://modelcontextprotocol.io/introduction)
 for more information.
 
+### Keepalive Mechanism
+
+For HTTP-based servers, you can enable a keepalive mechanism to prevent connections from dropping during long idle periods. This is done by adding the `keepalive_interval` property to your server configuration.
+
+- `keepalive_interval`: (Optional) An integer specifying the interval in seconds for sending a heartbeat (an `OPTIONS` request) to the server.
+  - If not provided, it defaults to **300** seconds.
+  - The value must be between **5** and **300** seconds.
+
+Example with keepalive enabled:
+```yaml
+mcp-servers:
+  mcpServers:
+    context7:
+      transport: http
+      url: https://mcp.context7.com/mcp
+      keepalive_interval: 60 # Send a heartbeat every 60 seconds
+```
+
 You have two ways of sharing your MCP server configuration with cecli.
 
 {: .note }
diff --git a/tests/mcp/conftest.py b/tests/mcp/conftest.py
new file mode 100644
index 00000000000..52e38d46ac4
--- /dev/null
+++ b/tests/mcp/conftest.py
@@ -0,0 +1,102 @@
+from typing import Any, AsyncGenerator, Dict
+from unittest.mock import MagicMock
+
+import pytest
+
+from cecli.mcp.server import HttpBasedMcpServer, HttpStreamingServer
+from tests.mcp.mock_server import MockMcpServer
+
+
+@pytest.fixture
+def mock_mcp_server() -> MockMcpServer:
+    """Fixture providing a mock MCP server instance."""
+    server = MockMcpServer()
+    return server
+
+
+@pytest.fixture
+async def running_mock_server(mock_mcp_server) -> AsyncGenerator[MockMcpServer, None]:
+    """Fixture providing a running mock MCP server."""
+    await mock_mcp_server.start()
+    yield mock_mcp_server
+    await mock_mcp_server.stop()
+
+
+@pytest.fixture
+def http_server_config(running_mock_server) -> Dict[str, Any]:
+    """Fixture providing a basic HTTP server configuration."""
+    return {
+        "name": "test-server",
+        "url": running_mock_server,
+        "type": "http",
+        "keepalive_interval": 1,  # 1 second for fast tests
+        "headers": {},
+        "enabled": True,
+    }
+
+
+@pytest.fixture
+def http_streaming_server_config(running_mock_server) -> Dict[str, Any]:
+    """Fixture providing an HTTP streaming server configuration."""
+    return {
+        "name": "test-streaming-server",
+        "url": running_mock_server,
+        "type": "streamable_http",
+        "keepalive_interval": 1,
+        "headers": {},
+        "enabled": True,
+    }
+
+
+@pytest.fixture
+def mock_io():
+    """Fixture providing a mock IO object."""
+    io = MagicMock()
+    io.tool_output = MagicMock()
+    io.tool_error = MagicMock()
+    io.tool_warning = MagicMock()
+    return io
+
+
+@pytest.fixture
+def http_based_server(http_server_config, mock_io) -> HttpBasedMcpServer:
+    """Fixture providing an HttpBasedMcpServer instance."""
+    return HttpBasedMcpServer(http_server_config, io=mock_io)
+
+
+@pytest.fixture
+def http_streaming_server(http_streaming_server_config, mock_io) -> HttpStreamingServer:
+    """Fixture providing an HttpStreamingServer instance."""
+    return HttpStreamingServer(http_streaming_server_config, io=mock_io)
+
+
+# Test utilities for inspecting internal state
+class ServerStateInspector:
+    """Utility class to inspect internal state of HttpBasedMcpServer for testing."""
+
+    @staticmethod
+    def get_state(server: HttpBasedMcpServer):
+        """Get the connection state of the server."""
+        return server._state
+
+    @staticmethod
+    def get_failed_pings(server: HttpBasedMcpServer):
+        """Get the number of failed pings."""
+        return server._failed_pings
+
+    @staticmethod
+    def get_keepalive_task(server: HttpBasedMcpServer):
+        """Get the keepalive task."""
+        return server._keepalive_task
+
+    @staticmethod
+    def is_keepalive_running(server: HttpBasedMcpServer):
+        """Check if the keepalive task is running."""
+        task = server._keepalive_task
+        return task is not None and not task.done()
+
+
+@pytest.fixture
+def server_inspector():
+    """Fixture providing a server state inspector."""
+    return ServerStateInspector()
diff --git a/tests/mcp/mock_server.py b/tests/mcp/mock_server.py
new file mode 100644
index 00000000000..b3a85f8e91f
--- /dev/null
+++ b/tests/mcp/mock_server.py
@@ -0,0 +1,126 @@
+"""Mock MCP server for testing keepalive mechanism.
+
+Provides controllable endpoints to simulate MCP server behavior:
+- /status: Control response status (200, 500, etc.)
+- /delay: Introduce artificial latency
+- /disconnect: Simulate sudden disconnection
+"""
+
+import asyncio
+import logging
+from typing import Optional
+
+from aiohttp import web
+
+logger = logging.getLogger(__name__)
+
+
+class MockMcpServer:
+    """Mock MCP server with controllable behavior for testing."""
+
+    def __init__(self, host: str = "127.0.0.1", port: int = 8765):
+        self.host = host
+        self.port = port
+        self.app = web.Application()
+        self.runner: Optional[web.AppRunner] = None
+        self.site: Optional[web.TCPSite] = None
+
+        # Controllable state
+        self.response_status = 200
+        self.response_delay = 0.0
+        self.disconnect_after_requests = 0
+        self.request_count = 0
+        self.should_disconnect = False
+
+        # Setup routes
+        self.app.router.add_route("*", "/status", self.handle_status)
+        self.app.router.add_route("*", "/delay", self.handle_delay)
+        self.app.router.add_route("*", "/disconnect", self.handle_disconnect)
+        self.app.router.add_route("*", "/{path:.*}", self.handle_default)
+
+    async def handle_status(self, request: web.Request) -> web.Response:
+        """Handle /status endpoint - returns configured status code."""
+        self.request_count += 1
+        if self.should_disconnect:
+            # Simulate connection drop
+            raise asyncio.CancelledError("Simulated disconnect")
+
+        if self.response_delay > 0:
+            await asyncio.sleep(self.response_delay)
+
+        return web.Response(status=self.response_status, text="OK")
+
+    async def handle_delay(self, request: web.Request) -> web.Response:
+        """Handle /delay endpoint - sets delay for subsequent requests."""
+        try:
+            data = await request.json()
+            self.response_delay = float(data.get("delay", 0))
+        except Exception:
+            self.response_delay = 0.0
+        return web.Response(status=200, text=f"Delay set to {self.response_delay}s")
+
+    async def handle_disconnect(self, request: web.Request) -> web.Response:
+        """Handle /disconnect endpoint - triggers disconnection."""
+        self.should_disconnect = True
+        return web.Response(status=200, text="Disconnect triggered")
+
+    async def handle_default(self, request: web.Request) -> web.Response:
+        """Handle all other requests (including OPTIONS for keepalive)."""
+        self.request_count += 1
+
+        if self.should_disconnect:
+            raise asyncio.CancelledError("Simulated disconnect")
+
+        if self.response_delay > 0:
+            await asyncio.sleep(self.response_delay)
+
+        # Simulate MCP server behavior - return 200 for OPTIONS
+        if request.method == "OPTIONS":
+            return web.Response(
+                status=200,
+                headers={
+                    "Access-Control-Allow-Origin": "*",
+                    "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
+                },
+            )
+
+        return web.Response(status=self.response_status, text="OK")
+
+    async def start(self) -> str:
+        """Start the mock server and return the base URL."""
+        self.runner = web.AppRunner(self.app)
+        await self.runner.setup()
+        self.site = web.TCPSite(self.runner, self.host, self.port)
+        await self.site.start()
+
+        url = f"http://{self.host}:{self.port}"
+        logger.info(f"Mock MCP server started at {url}")
+        return url
+
+    async def stop(self) -> None:
+        """Stop the mock server."""
+        if self.site:
+            await self.site.stop()
+        if self.runner:
+            await self.runner.cleanup()
+        logger.info("Mock MCP server stopped")
+
+    def reset(self) -> None:
+        """Reset server state to defaults."""
+        self.response_status = 200
+        self.response_delay = 0.0
+        self.disconnect_after_requests = 0
+        self.request_count = 0
+        self.should_disconnect = False
+
+    def set_status(self, status: int) -> None:
+        """Set the response status code for /status endpoint."""
+        self.response_status = status
+
+    def set_delay(self, delay: float) -> None:
+        """Set artificial delay for responses."""
+        self.response_delay = delay
+
+    def trigger_disconnect(self) -> None:
+        """Trigger a simulated disconnection."""
+        self.should_disconnect = True
diff --git a/tests/mcp/test_keepalive_concurrency.py b/tests/mcp/test_keepalive_concurrency.py
new file mode 100644
index 00000000000..8e2ca101d7c
--- /dev/null
+++ b/tests/mcp/test_keepalive_concurrency.py
@@ -0,0 +1,125 @@
+"""Concurrency tests for MCP keepalive task lifecycle."""
+
+import asyncio
+
+import pytest
+
+from cecli.mcp.server import HttpBasedMcpServer
+from tests.mcp.conftest import ServerStateInspector
+
+
+class TestKeepaliveTaskLifecycle:
+    """Test keepalive task creation, cancellation, and isolation."""
+
+    @pytest.mark.asyncio
+    async def test_keepalive_task_started_on_connect(self, http_based_server):
+        """Keepalive task is started when server connects."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Initially no task
+        assert inspector.get_keepalive_task(server) is None
+        assert not inspector.is_keepalive_running(server)
+
+        # Connect server
+        await server.connect()
+
+        # Task should be created and running
+        task = inspector.get_keepalive_task(server)
+        assert task is not None
+        assert isinstance(task, asyncio.Task)
+        assert inspector.is_keepalive_running(server)
+
+        # Cleanup
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_keepalive_task_cancelled_on_disconnect(self, http_based_server):
+        """Keepalive task is cancelled when server disconnects."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Connect and verify task is running
+        await server.connect()
+        assert inspector.is_keepalive_running(server)
+        task_before = inspector.get_keepalive_task(server)
+
+        # Disconnect server
+        await server.disconnect()
+
+        # Task should be cancelled
+        assert task_before.cancelled() or task_before.done()
+        assert (
+            inspector.get_keepalive_task(server) is None
+            or inspector.get_keepalive_task(server).done()
+        )
+        assert not inspector.is_keepalive_running(server)
+
+    @pytest.mark.asyncio
+    async def test_multiple_connect_disconnect_cycles(self, http_based_server):
+        """Server can handle multiple connect/disconnect cycles without task accumulation."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        tasks_seen = []
+
+        for i in range(3):
+            await server.connect()
+            assert inspector.is_keepalive_running(server)
+            task = inspector.get_keepalive_task(server)
+            tasks_seen.append(task)
+
+            await server.disconnect()
+            assert not inspector.is_keepalive_running(server)
+
+        # All tasks should be done or cancelled
+        for task in tasks_seen:
+            assert task.done() or task.cancelled()
+
+    @pytest.mark.asyncio
+    async def test_keepalive_task_does_not_block_other_operations(
+        self, http_based_server, running_mock_server
+    ):
+        """Keepalive task runs in background and doesn't block server operations."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Connect and verify keepalive starts
+        await server.connect()
+        assert inspector.is_keepalive_running(server)
+
+        # Perform other operations while keepalive runs
+        # These should not be blocked by the keepalive task
+
+        # Check connection status multiple times
+        for _ in range(5):
+            assert server.session is not None  # Local check
+            await asyncio.sleep(0.01)
+
+        # Change configuration (if supported)
+        # This tests that the event loop is not blocked
+
+        await asyncio.sleep(0.1)  # Let keepalive do its work
+
+        # Verify we can still disconnect cleanly
+        await server.disconnect()
+        assert not inspector.is_keepalive_running(server)
+
+    @pytest.mark.asyncio
+    async def test_no_keepalive_task_when_disabled(self, http_server_config, mock_io):
+        """No keepalive task is created when keepalive_interval is not specified."""
+        # Remove keepalive_interval from config
+        config = http_server_config.copy()
+        config.pop("keepalive_interval", None)
+
+        inspector = ServerStateInspector()
+        server = HttpBasedMcpServer(config, io=mock_io)
+
+        # Connect server
+        await server.connect()
+
+        # Should not have a keepalive task
+        assert inspector.get_keepalive_task(server) is None
+        assert not inspector.is_keepalive_running(server)
+
+        await server.disconnect()
diff --git a/tests/mcp/test_keepalive_config.py b/tests/mcp/test_keepalive_config.py
new file mode 100644
index 00000000000..51c469da76b
--- /dev/null
+++ b/tests/mcp/test_keepalive_config.py
@@ -0,0 +1,140 @@
+"""Configuration validation tests for MCP keepalive mechanism."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from cecli.mcp.manager import McpServerManager
+from cecli.mcp.server import HttpStreamingServer
+from tests.mcp.conftest import ServerStateInspector
+
+
+class TestKeepaliveConfigurationValidation:
+    """Test keepalive_interval configuration validation."""
+
+    @pytest.fixture
+    def mock_io(self):
+        io = MagicMock()
+        io.tool_output = MagicMock()
+        io.tool_error = MagicMock()
+        io.tool_warning = MagicMock()
+        return io
+
+    @pytest.fixture
+    def mock_manager(self, mock_io):
+        return McpServerManager(servers=[], io=mock_io)
+
+    def test_keepalive_interval_below_minimum_rejected(self, mock_manager):
+        """Configuration with keepalive_interval < MIN_KEEPALIVE_INTERVAL is rejected."""
+        config = {
+            "name": "test-server",
+            "url": "http://localhost:8000",
+            "type": "streamable_http",
+            "keepalive_interval": 1,  # Below minimum of 5
+            "enabled": True,
+        }
+        with pytest.raises(ValueError, match="keepalive_interval"):
+            mock_manager._validate_server_config(config)
+
+    def test_keepalive_interval_above_maximum_rejected(self, mock_manager):
+        """Configuration with keepalive_interval > MAX_KEEPALIVE_INTERVAL is rejected."""
+        config = {
+            "name": "test-server",
+            "url": "http://localhost:8000",
+            "type": "streamable_http",
+            "keepalive_interval": 400,  # Above maximum of 300
+            "enabled": True,
+        }
+        with pytest.raises(ValueError, match="keepalive_interval"):
+            mock_manager._validate_server_config(config)
+
+    def test_keepalive_interval_non_integer_rejected(self, mock_manager):
+        """Configuration with non-integer keepalive_interval is rejected."""
+        config = {
+            "name": "test-server",
+            "url": "http://localhost:8000",
+            "type": "streamable_http",
+            "keepalive_interval": 5.5,
+            "enabled": True,
+        }
+        with pytest.raises(ValueError, match="keepalive_interval"):
+            mock_manager._validate_server_config(config)
+
+    def test_keepalive_interval_valid_accepted(self, mock_manager):
+        """Configuration with valid keepalive_interval is accepted."""
+        config = {
+            "name": "test-server",
+            "url": "http://localhost:8000",
+            "type": "streamable_http",
+            "keepalive_interval": 15,
+            "enabled": True,
+        }
+        # Should not raise
+        validated = mock_manager._validate_server_config(config)
+        assert validated["keepalive_interval"] == 15
+
+    def test_keepalive_disabled_when_not_specified(self, mock_manager):
+        """Server without keepalive_interval does not start keepalive task."""
+        config = {
+            "name": "test-server",
+            "url": "http://localhost:8000",
+            "type": "streamable_http",
+            "enabled": True,
+        }
+        validated = mock_manager._validate_server_config(config)
+        assert "keepalive_interval" not in validated or validated.get("keepalive_interval") is None
+
+    @pytest.mark.asyncio
+    async def test_auth_header_included_in_keepalive_request(
+        self, mock_manager, running_mock_server
+    ):
+        """Authentication headers from server config are included in OPTIONS requests."""
+        config = {
+            "name": "test-server",
+            "url": f"http://{running_mock_server.host}:{running_mock_server.port}",
+            "type": "streamable_http",
+            "keepalive_interval": 1,
+            "headers": {"Authorization": "Bearer test-token"},
+            "enabled": True,
+        }
+
+        server = HttpStreamingServer(config, io=MagicMock())
+
+        with (
+            patch("cecli.mcp.server.ClientSession") as MockSession,
+            patch("cecli.mcp.server.streamable_http_client") as mock_transport,
+            patch("httpx.AsyncClient") as MockAsyncClient,
+        ):
+            # Setup mock HTTP client to capture constructor args
+            mock_http_client = AsyncMock()
+            MockAsyncClient.return_value = mock_http_client
+
+            # Setup mock session
+            mock_session = AsyncMock()
+            mock_session.initialize = AsyncMock()
+            MockSession.return_value = mock_session
+
+            # Setup mock transport
+            mock_read = AsyncMock()
+            mock_write = AsyncMock()
+            mock_transport.return_value = (mock_read, mock_write, None)
+
+            await server.connect()
+            await asyncio.sleep(0.2)  # Allow keepalive to run
+
+            # Verify keepalive task is running
+            inspector = ServerStateInspector()
+            assert inspector.is_keepalive_running(server)
+
+            # Verify httpx.AsyncClient was created with auth headers
+            MockAsyncClient.assert_called_once()
+            call_kwargs = MockAsyncClient.call_args.kwargs
+            assert (
+                "headers" in call_kwargs
+            ), f"Expected 'headers' in AsyncClient kwargs, got: {list(call_kwargs.keys())}"
+            assert call_kwargs["headers"] == {
+                "Authorization": "Bearer test-token"
+            }, f"Expected auth header, got: {call_kwargs['headers']}"
+
+            await server.disconnect()
diff --git a/tests/mcp/test_keepalive_integration.py b/tests/mcp/test_keepalive_integration.py
new file mode 100644
index 00000000000..6f9622bcad5
--- /dev/null
+++ b/tests/mcp/test_keepalive_integration.py
@@ -0,0 +1,144 @@
+"""Integration tests for MCP keepalive mechanism with mock server."""
+
+import asyncio
+
+import pytest
+
+from cecli.mcp.server import ConnectionState
+from tests.mcp.conftest import ServerStateInspector
+
+
+class TestKeepaliveWithMockServer:
+    """Test keepalive mechanism with a controllable mock MCP server."""
+
+    @pytest.mark.asyncio
+    async def test_options_requests_sent_periodically(self, http_based_server, running_mock_server):
+        """Verify OPTIONS requests are sent periodically when keepalive is enabled."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Start the server connection
+        await server.connect()
+        await asyncio.sleep(0.1)  # Allow keepalive task to start
+
+        # Verify keepalive task is running
+        assert inspector.is_keepalive_running(server)
+
+        # Wait for at least one keepalive interval (1 second)
+        await asyncio.sleep(1.2)
+
+        # Verify mock server received requests
+        assert running_mock_server.request_count >= 1
+
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_connection_remains_active_during_idle_periods(
+        self, http_based_server, running_mock_server
+    ):
+        """Verify connection remains active during idle periods with successful keepalive."""
+        server = http_based_server
+
+        # Connect and verify initial state
+        await server.connect()
+        inspector = ServerStateInspector()
+        assert inspector.get_state(server) == ConnectionState.CONNECTED
+
+        # Wait for several keepalive intervals
+        await asyncio.sleep(3.5)  # 3 intervals of 1 second each
+
+        # Verify still connected
+        assert inspector.get_state(server) == ConnectionState.CONNECTED
+        assert inspector.get_failed_pings(server) == 0
+
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_server_failure_triggers_unhealthy_state(
+        self, http_based_server, running_mock_server
+    ):
+        """Verify server transitions to UNHEALTHY when keepalive fails."""
+        server = http_based_server
+        inspector = ServerStateInspector()
+
+        await server.connect()
+        await asyncio.sleep(0.1)
+
+        # Make mock server return errors
+        running_mock_server.set_status(500)
+
+        # Wait for failed ping
+        await asyncio.sleep(1.2)
+
+        # Should transition to UNHEALTHY
+        assert inspector.get_state(server) == ConnectionState.UNHEALTHY
+        assert inspector.get_failed_pings(server) == 1
+
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_consecutive_failures_lead_to_disconnected_state(
+        self, http_based_server, running_mock_server
+    ):
+        """Verify server transitions to DISCONNECTED after threshold failures."""
+        server = http_based_server
+        inspector = ServerStateInspector()
+
+        await server.connect()
+        await asyncio.sleep(0.1)
+
+        # Make mock server consistently fail
+        running_mock_server.set_status(500)
+
+        # Wait for failures exceeding threshold (3 failures)
+        await asyncio.sleep(4.0)  # Allow time for 3 pings
+
+        # Should transition to DISCONNECTED
+        assert inspector.get_state(server) == ConnectionState.DISCONNECTED
+        assert inspector.get_failed_pings(server) >= 3
+
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_successful_ping_after_failure_restores_healthy_state(
+        self, http_based_server, running_mock_server
+    ):
+        """Verify successful ping after failure restores CONNECTED state."""
+        server = http_based_server
+        inspector = ServerStateInspector()
+
+        await server.connect()
+        await asyncio.sleep(0.1)
+
+        # Cause a failure
+        running_mock_server.set_status(500)
+        await asyncio.sleep(1.2)
+        assert inspector.get_state(server) == ConnectionState.UNHEALTHY
+
+        # Restore success
+        running_mock_server.set_status(200)
+        await asyncio.sleep(1.2)
+
+        # Should be back to CONNECTED
+        assert inspector.get_state(server) == ConnectionState.CONNECTED
+        assert inspector.get_failed_pings(server) == 0
+
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_streaming_server_keepalive_also_works(
+        self, http_streaming_server, running_mock_server
+    ):
+        """Verify HTTP streaming server keepalive mechanism works similarly."""
+        server = http_streaming_server
+        inspector = ServerStateInspector()
+
+        await server.connect()
+        await asyncio.sleep(0.1)
+
+        assert inspector.is_keepalive_running(server)
+
+        await asyncio.sleep(1.2)
+        assert running_mock_server.request_count >= 1
+
+        await server.disconnect()
diff --git a/tests/mcp/test_keepalive_logging.py b/tests/mcp/test_keepalive_logging.py
new file mode 100644
index 00000000000..c800e3d987d
--- /dev/null
+++ b/tests/mcp/test_keepalive_logging.py
@@ -0,0 +1,82 @@
+"""Logging and metrics tests for MCP keepalive mechanism."""
+
+import asyncio
+import logging
+
+
+class TestKeepaliveLogging:
+    """Test logging and metrics for keepalive mechanism."""
+
+    def test_log_sanitization_no_sensitive_data(self, http_based_server, caplog):
+        """Verify that logs don't contain sensitive information like URLs or credentials."""
+        server = http_based_server
+
+        # Enable log capture
+        caplog.set_level(logging.INFO)
+
+        # Connect server to trigger keepalive startup log
+        async def run_test():
+            await server.connect()
+            await asyncio.sleep(0.1)
+            await server.disconnect()
+
+        asyncio.run(run_test())
+
+        # Check that logs don't contain sensitive data
+        log_text = "".join(caplog.messages)
+
+        # URL should not appear in logs (or should be sanitized)
+        # In a real implementation, we'd check for proper sanitization
+        # For now, we verify logging happens without error
+        assert "Keepalive task started" in log_text or "Keepalive task stopped" in log_text
+
+    def test_keepalive_events_logged_correctly(self, http_based_server, caplog):
+        """Verify that key keepalive events are logged."""
+        server = http_based_server
+
+        caplog.set_level(logging.INFO)
+
+        async def run_test():
+            await server.connect()
+            await asyncio.sleep(0.1)  # Allow startup log
+            await server.disconnect()
+
+        asyncio.run(run_test())
+
+        log_text = "".join(caplog.messages)
+
+        # At least startup/shutdown logs should be present
+        assert any(
+            event in log_text for event in ["Keepalive task started", "Keepalive task stopped"]
+        )
+
+    def test_state_transitions_are_logged(self, http_based_server, caplog):
+        """Verify that all keepalive state transitions are properly logged."""
+        server = http_based_server
+
+        caplog.set_level(logging.INFO)
+
+        async def run_test():
+            # Connect - should log CONNECTED state
+            await server.connect()
+            await asyncio.sleep(0.1)  # Allow startup log
+
+            # Force disconnection to trigger UNHEALTHY -> DISCONNECTED
+            # by making the server return 500 errors
+            if hasattr(server, "_http_client"):
+                # For HTTP-based servers, we can't easily make it fail
+                # Instead, let's test the logging by checking what we can
+                pass
+
+            await server.disconnect()
+            await asyncio.sleep(0.1)  # Allow disconnect log
+
+        asyncio.run(run_test())
+
+        log_text = "".join(caplog.messages)
+
+        # Verify key state transition events are logged
+        assert "Keepalive task started" in log_text
+        assert "Keepalive task stopped" in log_text
+        # Note: Detailed state transition logging depends on implementation
+        # but at minimum we should see the task lifecycle events
diff --git a/tests/mcp/test_keepalive_resilience.py b/tests/mcp/test_keepalive_resilience.py
new file mode 100644
index 00000000000..e4329816529
--- /dev/null
+++ b/tests/mcp/test_keepalive_resilience.py
@@ -0,0 +1,139 @@
+"""Resilience tests for MCP keepalive mechanism."""
+
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from cecli.mcp.server import ConnectionState
+from tests.mcp.conftest import ServerStateInspector
+
+
+class TestKeepaliveResilience:
+    """Test keepalive mechanism resilience under various conditions."""
+
+    @pytest.mark.asyncio
+    async def test_temporary_disconnection_recovery(self, http_based_server, running_mock_server):
+        """Verify server recovers from temporary disconnection."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        await server.connect()
+        await asyncio.sleep(0.1)
+
+        # Simulate temporary disconnection
+        running_mock_server.trigger_disconnect()
+        await asyncio.sleep(1.2)  # Wait for failed ping
+
+        # Should be UNHEALTHY after first failure
+        assert inspector.get_state(server) == ConnectionState.UNHEALTHY
+        assert inspector.get_failed_pings(server) == 1
+
+        # Restore server
+        running_mock_server.reset()
+        running_mock_server.set_status(200)
+        await asyncio.sleep(1.2)  # Wait for successful ping
+
+        # Should recover to CONNECTED
+        assert inspector.get_state(server) == ConnectionState.CONNECTED
+        assert inspector.get_failed_pings(server) == 0
+
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_slow_responses_handled_gracefully(self, http_based_server, running_mock_server):
+        """Verify keepalive continues to function with slow server responses."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        await server.connect()
+        await asyncio.sleep(0.1)
+
+        # Set delay longer than keepalive interval but not excessive
+        running_mock_server.set_delay(0.8)  # 0.8s delay vs 1s interval
+
+        # Wait for multiple intervals
+        await asyncio.sleep(3.0)
+
+        # Should still be functioning and task should be alive
+        assert inspector.get_keepalive_task(server) is not None
+
+        await server.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_keepalive_jitter_prevents_timing_analysis(self, http_based_server):
+        """Verify keepalive intervals incorporate jitter."""
+        server = http_based_server
+        sleep_durations = []
+
+        async def mock_sleep(duration):
+            sleep_durations.append(duration)
+            # Don't actually sleep to speed up test
+
+        await server.connect()
+
+        with patch("asyncio.sleep", side_effect=mock_sleep):
+            # Let keepalive loop run a few iterations
+            await asyncio.sleep(3.5)
+
+        await server.disconnect()
+
+        # Verify we captured sleep durations
+        assert len(sleep_durations) >= 2, f"Expected >= 2 sleep calls, got {len(sleep_durations)}"
+
+        # Verify jitter exists - durations should not all be identical
+        assert len(set(sleep_durations)) > 1, "Sleep durations should vary due to jitter"
+
+        # Verify durations fall within +/-10% of configured interval
+        interval = server.config.get("keepalive_interval", 1)
+        for duration in sleep_durations:
+            assert (
+                0.9 * interval <= duration <= 1.1 * interval
+            ), f"Duration {duration} outside +/-10% jitter range"
+
+    @pytest.mark.asyncio
+    async def test_reconnection_after_persistent_failure(
+        self, http_based_server, running_mock_server
+    ):
+        """Verify exponential backoff reconnection after persistent failure."""
+        server = http_based_server
+        server.config["keepalive_interval"] = 1
+
+        await server.connect()
+        await asyncio.sleep(0.1)
+
+        # Make server consistently fail to trigger reconnection logic
+        running_mock_server.set_status(500)
+
+        reconnect_delays = []
+
+        async def mock_sleep(duration):
+            reconnect_delays.append(duration)
+            if duration > 0.5:
+                return  # Skip actual sleep for reconnection delays
+
+        with patch("asyncio.sleep", side_effect=mock_sleep):
+            # Allow enough virtual time for multiple backoff attempts
+            await asyncio.sleep(40)
+
+        await server.disconnect()
+
+        # Filter for reconnection delay calls (values between 0.5 and 301 seconds)
+        delays = [d for d in reconnect_delays if 0.5 < d < 301]
+
+        assert len(delays) >= 2, f"Expected >= 2 reconnection attempts, got {len(delays)}"
+
+        # Verify delays follow exponential backoff pattern:
+        # initial=1s, multiplier=2 -> ~1s, ~2s, ~4s, ~8s, ~16s, ~32s...
+        expected_bases = [1, 2, 4, 8, 16, 32]
+        for i, delay in enumerate(delays):
+            base = expected_bases[min(i, len(expected_bases) - 1)]
+            assert (
+                base * 0.8 <= delay <= base * 1.2
+            ), f"Delay {delay} not within +/-20% of expected {base}"
+
+        # Verify delays are capped at max_delay (300s)
+        for delay in delays:
+            assert delay <= 300, f"Delay {delay} exceeds max_delay of 300"
+
+        await server.disconnect()
diff --git a/tests/mcp/test_keepalive_unit.py b/tests/mcp/test_keepalive_unit.py
new file mode 100644
index 00000000000..08bc2195533
--- /dev/null
+++ b/tests/mcp/test_keepalive_unit.py
@@ -0,0 +1,151 @@
+"""Unit tests for MCP keepalive state transitions and reconnection logic."""
+
+import pytest
+
+from cecli.mcp.server import ConnectionState
+from tests.mcp.conftest import ServerStateInspector
+
+
+class TestConnectionStateTransitions:
+    """Test state machine transitions for keepalive mechanism."""
+
+    def test_initial_state_is_connected(self, http_based_server):
+        """Server starts in CONNECTED state after initialization."""
+        inspector = ServerStateInspector()
+        assert inspector.get_state(http_based_server) == ConnectionState.CONNECTED
+        assert inspector.get_failed_pings(http_based_server) == 0
+
+    def test_transition_to_unhealthy_on_first_failed_ping(self, http_based_server):
+        """Server transitions from CONNECTED to UNHEALTHY on first failed ping."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Simulate a failed ping
+        server._failed_pings = 1
+        server._state = ConnectionState.UNHEALTHY
+
+        assert inspector.get_state(server) == ConnectionState.UNHEALTHY
+        assert inspector.get_failed_pings(server) == 1
+
+    def test_transition_to_connected_on_successful_ping_after_unhealthy(self, http_based_server):
+        """Server transitions from UNHEALTHY back to CONNECTED on successful ping."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Start in UNHEALTHY state
+        server._state = ConnectionState.UNHEALTHY
+        server._failed_pings = 1
+
+        # Simulate successful ping recovery
+        server._failed_pings = 0
+        server._state = ConnectionState.CONNECTED
+
+        assert inspector.get_state(server) == ConnectionState.CONNECTED
+        assert inspector.get_failed_pings(server) == 0
+
+    def test_transition_to_disconnected_after_threshold_failures(self, http_based_server):
+        """Server transitions from UNHEALTHY to DISCONNECTED after threshold failures."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Simulate multiple failures exceeding threshold
+        server._state = ConnectionState.UNHEALTHY
+        server._failed_pings = 2
+
+        # Next failure should trigger DISCONNECTED
+        server._failed_pings = 3
+        server._state = ConnectionState.DISCONNECTED
+
+        assert inspector.get_state(server) == ConnectionState.DISCONNECTED
+        assert inspector.get_failed_pings(server) == 3
+
+    def test_no_direct_transition_from_connected_to_disconnected(self, http_based_server):
+        """Server should not transition directly from CONNECTED to DISCONNECTED."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Verify initial state
+        assert inspector.get_state(server) == ConnectionState.CONNECTED
+
+        # Direct transition should not happen in normal flow
+        # The state should go through UNHEALTHY first
+        server._failed_pings = 1
+        server._state = ConnectionState.UNHEALTHY
+
+        assert inspector.get_state(server) == ConnectionState.UNHEALTHY
+        assert inspector.get_failed_pings(server) == 1
+
+
+class TestReconnectionLogic:
+    """Test reconnection logic with exponential backoff."""
+
+    @pytest.mark.asyncio
+    async def test_reconnect_called_when_disconnected(self, http_based_server):
+        """Reconnect method is invoked when state becomes DISCONNECTED."""
+        server = http_based_server
+        inspector = ServerStateInspector()
+
+        # Set server to DISCONNECTED state
+        server._state = ConnectionState.DISCONNECTED
+        server._failed_pings = 3
+
+        # Verify reconnect would be triggered (state check)
+        assert inspector.get_state(server) == ConnectionState.DISCONNECTED
+        assert inspector.get_failed_pings(server) == 3
+
+    @pytest.mark.asyncio
+    async def test_exponential_backoff_parameters(self, http_based_server):
+        """Verify exponential backoff strategy parameters."""
+        # According to plan: initial=1s, multiplier=2, max=300s, jitter=±20%
+        initial_delay = 1
+        multiplier = 2
+        max_delay = 300
+        jitter_percent = 20
+
+        # Calculate expected delays for first few retries
+        delays = []
+        current_delay = initial_delay
+        for _ in range(5):
+            jitter = current_delay * (jitter_percent / 100)
+            delays.append((current_delay - jitter, current_delay + jitter))
+            current_delay = min(current_delay * multiplier, max_delay)
+
+        # Verify delays are within expected range
+        assert delays[0][0] == 0.8  # 1s - 20%
+        assert delays[0][1] == 1.2  # 1s + 20%
+        assert delays[1][0] == 1.6  # 2s - 20%
+        assert delays[1][1] == 2.4  # 2s + 20%
+        assert delays[4][0] == 25.6  # 32s - 20%
+        assert delays[4][1] == 38.4  # 32s + 20%
+
+    @pytest.mark.asyncio
+    async def test_max_backoff_cap(self, http_based_server):
+        """Verify exponential backoff is capped at maximum delay."""
+        initial_delay = 1
+        multiplier = 2
+        max_delay = 300
+
+        current_delay = initial_delay
+        for _ in range(20):  # Many retries
+            current_delay = min(current_delay * multiplier, max_delay)
+            if current_delay >= max_delay:
+                break
+
+        assert current_delay == max_delay
+
+    @pytest.mark.asyncio
+    async def test_reconnect_success_restores_connected_state(self, http_based_server):
+        """Successful reconnection restores CONNECTED state."""
+        inspector = ServerStateInspector()
+        server = http_based_server
+
+        # Start in DISCONNECTED state
+        server._state = ConnectionState.DISCONNECTED
+        server._failed_pings = 3
+
+        # Simulate successful reconnection
+        server._failed_pings = 0
+        server._state = ConnectionState.CONNECTED
+
+        assert inspector.get_state(server) == ConnectionState.CONNECTED
+        assert inspector.get_failed_pings(server) == 0