feat: Add configurable SSE keepalive events

- Add SSE_KEEPALIVE_ENABLED and SSE_KEEPALIVE_INTERVAL environment variables
- Update sse_transport.py to respect keepalive configuration settings
- Update translate.py to use configurable keepalive defaults
- Add comprehensive tests for keepalive configuration scenarios
- Update documentation with configuration options and troubleshooting
- Fixes issue #689 by allowing users to disable keepalive events for incompatible clients
- Supports infrastructure-specific intervals (30s default, 60s AWS ALB, 240s Azure)

Signed-off-by: Mihai Criveti <[email protected]>

Author: crivetimihai

.env.example

@@ -187,6 +187,16 @@ WEBSOCKET_PING_INTERVAL=30
 # SSE client retry timeout (milliseconds)
 SSE_RETRY_TIMEOUT=5000
 
+# Enable SSE keepalive events (true/false)
+# Set to false to disable keepalive events completely
+# Note: Disabling may cause timeouts with proxies/load balancers
+SSE_KEEPALIVE_ENABLED=true
+
+# SSE keepalive interval (seconds)
+# How often to send keepalive events during idle periods
+# Common values: 30 (default), 60 (AWS ALB), 120, 240 (Azure)
+SSE_KEEPALIVE_INTERVAL=30
+
 
 #####################################
 # Streamabe HTTP Transport Configuration

CLAUDE.md

@@ -215,3 +215,6 @@ pytest -m "not slow"
 
 make autoflake isort black pre-commit
 make doctest test smoketest lint-web flake8 pylint
+
+# Rules
+- When using git commit always add a -s to sign commits

README.md

@@ -126,7 +126,7 @@ It currently supports:
 
 * Federation across multiple MCP and REST services
 * Virtualization of legacy APIs as MCP-compliant tools and servers
-* Transport over HTTP, JSON-RPC, WebSocket, SSE, stdio and streamable-HTTP
+* Transport over HTTP, JSON-RPC, WebSocket, SSE (with configurable keepalive), stdio and streamable-HTTP
 * An Admin UI for real-time management and configuration
 * Built-in auth, observability, retries, and rate-limiting
 * Scalable deployments via Docker or PyPI, Redis-backed caching, and multi-cluster federation
@@ -1029,9 +1029,13 @@ You can get started by copying the provided [.env.example](.env.example) to `.en
 | `TRANSPORT_TYPE`          | Enabled transports                 | `all`   | `http`,`ws`,`sse`,`stdio`,`all` |
 | `WEBSOCKET_PING_INTERVAL` | WebSocket ping (secs)              | `30`    | int > 0                         |
 | `SSE_RETRY_TIMEOUT`       | SSE retry timeout (ms)             | `5000`  | int > 0                         |
+| `SSE_KEEPALIVE_ENABLED`   | Enable SSE keepalive events        | `true`  | bool                            |
+| `SSE_KEEPALIVE_INTERVAL`  | SSE keepalive interval (secs)      | `30`    | int > 0                         |
 | `USE_STATEFUL_SESSIONS`   | streamable http config             | `false` | bool                            |
 | `JSON_RESPONSE_ENABLED`   | json/sse streams (streamable http) | `true`  | bool                            |
 
+> **💡 SSE Keepalive Events**: The gateway sends periodic keepalive events to prevent connection timeouts with proxies and load balancers. Disable with `SSE_KEEPALIVE_ENABLED=false` if your client doesn't handle unknown event types. Common intervals: 30s (default), 60s (AWS ALB), 240s (Azure).
+
 ### Federation
 
 | Setting                    | Description            | Default | Options    |
@@ -1112,7 +1116,13 @@ MCP Gateway uses Alembic for database migrations. Common commands:
 
 #### Troubleshooting
 
-If you see "No 'script_location' key found", ensure you're running from the project root directory.
+**Common Issues:**
+
+- **"No 'script_location' key found"**: Ensure you're running from the project root directory.
+
+- **"Unknown SSE event: keepalive" warnings**: Some MCP clients don't recognize keepalive events. These warnings are harmless and don't affect functionality. To disable: `SSE_KEEPALIVE_ENABLED=false`
+
+- **Connection timeouts with proxies/load balancers**: If experiencing timeouts, adjust keepalive interval to match your infrastructure: `SSE_KEEPALIVE_INTERVAL=60` (AWS ALB) or `240` (Azure).
 
 ### Development
 

mcpgateway/config.py

@@ -197,6 +197,8 @@ def _parse_allowed_origins(cls, v):
     transport_type: str = "all"  # http, ws, sse, all
     websocket_ping_interval: int = 30  # seconds
     sse_retry_timeout: int = 5000  # milliseconds
+    sse_keepalive_enabled: bool = True  # Enable SSE keepalive events
+    sse_keepalive_interval: int = 30  # seconds between keepalive events
 
     # Federation
     federation_enabled: bool = True

mcpgateway/translate.py

@@ -78,7 +78,20 @@
     httpx = None  # type: ignore[assignment]
 
 LOGGER = logging.getLogger("mcpgateway.translate")
-KEEP_ALIVE_INTERVAL = 30  # seconds - matches the reference implementation
+
+# Import settings for default keepalive interval
+try:
+    # First-Party
+    from mcpgateway.config import settings
+
+    DEFAULT_KEEP_ALIVE_INTERVAL = settings.sse_keepalive_interval
+    DEFAULT_KEEPALIVE_ENABLED = settings.sse_keepalive_enabled
+except ImportError:
+    # Fallback if config not available
+    DEFAULT_KEEP_ALIVE_INTERVAL = 30
+    DEFAULT_KEEPALIVE_ENABLED = True
+
+KEEP_ALIVE_INTERVAL = DEFAULT_KEEP_ALIVE_INTERVAL  # seconds - from config or fallback to 30
 __all__ = ["main"]  # for console-script entry-point
 
 
@@ -571,23 +584,26 @@ async def event_gen() -> AsyncIterator[Dict[str, Any]]:
                 "retry": int(keep_alive * 1000),
             }
 
-            # 2️⃣ Immediate keepalive so clients know the stream is alive
-            yield {"event": "keepalive", "data": "{}", "retry": keep_alive * 1000}
+            # 2️⃣ Immediate keepalive so clients know the stream is alive (if enabled in config)
+            if DEFAULT_KEEPALIVE_ENABLED:
+                yield {"event": "keepalive", "data": "{}", "retry": keep_alive * 1000}
 
             try:
                 while True:
                     if await request.is_disconnected():
                         break
 
                     try:
-                        msg = await asyncio.wait_for(queue.get(), keep_alive)
+                        timeout = keep_alive if DEFAULT_KEEPALIVE_ENABLED else None
+                        msg = await asyncio.wait_for(queue.get(), timeout)
                         yield {"event": "message", "data": msg.rstrip()}
                     except asyncio.TimeoutError:
-                        yield {
-                            "event": "keepalive",
-                            "data": "{}",
-                            "retry": keep_alive * 1000,
-                        }
+                        if DEFAULT_KEEPALIVE_ENABLED:
+                            yield {
+                                "event": "keepalive",
+                                "data": "{}",
+                                "retry": keep_alive * 1000,
+                            }
             finally:
                 pubsub.unsubscribe(queue)
 

mcpgateway/transports/sse_transport.py

@@ -346,20 +346,22 @@ async def event_generator():
                 "retry": settings.sse_retry_timeout,
             }
 
-            # Send keepalive immediately to help establish connection
-            yield {
-                "event": "keepalive",
-                "data": "{}",
-                "retry": settings.sse_retry_timeout,
-            }
+            # Send keepalive immediately to help establish connection (if enabled)
+            if settings.sse_keepalive_enabled:
+                yield {
+                    "event": "keepalive",
+                    "data": "{}",
+                    "retry": settings.sse_retry_timeout,
+                }
 
             try:
                 while not self._client_gone.is_set():
                     try:
                         # Wait for messages with a timeout for keepalives
+                        timeout = settings.sse_keepalive_interval if settings.sse_keepalive_enabled else None
                         message = await asyncio.wait_for(
                             self._message_queue.get(),
-                            timeout=30.0,  # 30 second timeout for keepalives (some tools require more timeout for execution)
+                            timeout=timeout,  # Configurable timeout for keepalives (some tools require more timeout for execution)
                         )
 
                         data = json.dumps(message, default=lambda obj: (obj.strftime("%Y-%m-%d %H:%M:%S") if isinstance(obj, datetime) else TypeError("Type not serializable")))
@@ -373,12 +375,13 @@ async def event_generator():
                             "retry": settings.sse_retry_timeout,
                         }
                     except asyncio.TimeoutError:
-                        # Send keepalive on timeout
-                        yield {
-                            "event": "keepalive",
-                            "data": "{}",
-                            "retry": settings.sse_retry_timeout,
-                        }
+                        # Send keepalive on timeout (if enabled)
+                        if settings.sse_keepalive_enabled:
+                            yield {
+                                "event": "keepalive",
+                                "data": "{}",
+                                "retry": settings.sse_retry_timeout,
+                            }
                     except Exception as e:
                         logger.error(f"Error processing SSE message: {e}")
                         yield {

tests/unit/mcpgateway/transports/test_sse_transport.py

@@ -228,3 +228,75 @@ async def test_event_generator(self, sse_transport, mock_request):
 
         # Cancel the generator to clean up
         sse_transport._client_gone.set()
+
+    @pytest.mark.asyncio
+    async def test_keepalive_disabled(self, sse_transport, mock_request):
+        """Test SSE response when keepalive is disabled."""
+        with patch("mcpgateway.transports.sse_transport.settings") as mock_settings:
+            mock_settings.sse_keepalive_enabled = False
+            mock_settings.sse_keepalive_interval = 30
+            mock_settings.sse_retry_timeout = 5000
+
+            await sse_transport.connect()
+            response = await sse_transport.create_sse_response(mock_request)
+            generator = response.body_iterator
+
+            # First event should be endpoint
+            event = await generator.__anext__()
+            assert event["event"] == "endpoint"
+
+            # No immediate keepalive should be sent
+            # Queue a test message
+            test_message = {"jsonrpc": "2.0", "result": "test", "id": 1}
+            await sse_transport._message_queue.put(test_message)
+
+            # Next event should be the message (no keepalive)
+            event = await generator.__anext__()
+            assert event["event"] == "message"
+
+            sse_transport._client_gone.set()
+
+    @pytest.mark.asyncio
+    async def test_keepalive_custom_interval(self, sse_transport, mock_request):
+        """Test SSE response with custom keepalive interval."""
+        with patch("mcpgateway.transports.sse_transport.settings") as mock_settings:
+            mock_settings.sse_keepalive_enabled = True
+            mock_settings.sse_keepalive_interval = 60  # Custom interval
+            mock_settings.sse_retry_timeout = 5000
+
+            await sse_transport.connect()
+            response = await sse_transport.create_sse_response(mock_request)
+            generator = response.body_iterator
+
+            # First event should be endpoint
+            event = await generator.__anext__()
+            assert event["event"] == "endpoint"
+
+            # Second event should be immediate keepalive
+            event = await generator.__anext__()
+            assert event["event"] == "keepalive"
+            assert event["data"] == "{}"
+
+            sse_transport._client_gone.set()
+
+    @pytest.mark.asyncio
+    async def test_keepalive_timeout_behavior(self, sse_transport, mock_request):
+        """Test timeout behavior respects keepalive settings."""
+        with patch("mcpgateway.transports.sse_transport.settings") as mock_settings:
+            mock_settings.sse_keepalive_enabled = True
+            mock_settings.sse_keepalive_interval = 1  # 1 second for quick test
+            mock_settings.sse_retry_timeout = 5000
+
+            await sse_transport.connect()
+            response = await sse_transport.create_sse_response(mock_request)
+            generator = response.body_iterator
+
+            # Skip endpoint and initial keepalive
+            await generator.__anext__()  # endpoint
+            await generator.__anext__()  # initial keepalive
+
+            # Wait for timeout keepalive (should happen after 1 second)
+            event = await asyncio.wait_for(generator.__anext__(), timeout=2.0)
+            assert event["event"] == "keepalive"
+
+            sse_transport._client_gone.set()