feat(mcp): add SSE transport fallback and keepalive for HTTP MCP servers

tests/tools/test_mcp_tool.py

@@ -921,8 +921,9 @@ def test_http_unavailable_raises(self):
 
         async def _test():
             with patch("tools.mcp_tool._MCP_HTTP_AVAILABLE", False):
-                with pytest.raises(ImportError, match="HTTP transport"):
-                    await server._run_http(config)
+                with patch("tools.mcp_tool._MCP_SSE_AVAILABLE", False):
+                    with pytest.raises(ImportError, match="HTTP transport"):
+                        await server._run_http(config)
 
         asyncio.run(_test())
 
@@ -2900,3 +2901,115 @@ async def fake_connect(name, config):
         assert mock_registry.get_toolset_for_tool("mcp_srv_do_thing") == "mcp-srv"
 
         _servers.pop("srv", None)
+
+
+# ---------------------------------------------------------------------------
+# SSE transport detection tests (based on work by @amiller in #5981)
+# ---------------------------------------------------------------------------
+
+class TestSSEConfig:
+    """Tests for SSE transport detection and handling."""
+
+    def test_is_sse_with_transport_config(self):
+        """Explicit transport: sse config is detected."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("sse-srv")
+        server._config = {"url": "https://example.com/mcp/sse", "transport": "sse"}
+        assert server._is_sse() is True
+
+    def test_is_sse_from_url_path(self):
+        """URL ending in /sse is auto-detected as SSE."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("auto-sse")
+        server._config = {"url": "https://example.com/mcp/sse"}
+        assert server._is_sse() is True
+
+    def test_is_sse_from_url_with_trailing_slash(self):
+        """URL ending in /sse/ (trailing slash) is still detected as SSE."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("trailing")
+        server._config = {"url": "https://example.com/mcp/sse/"}
+        assert server._is_sse() is True
+
+    def test_is_sse_from_url_with_query_params(self):
+        """URL with /sse path and query params is detected as SSE."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("query")
+        server._config = {"url": "https://example.com/mcp/sse?key=abc123"}
+        assert server._is_sse() is True
+
+    def test_not_sse_for_plain_http_url(self):
+        """Regular /mcp URL is NOT SSE."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("http-srv")
+        server._config = {"url": "https://example.com/mcp"}
+        assert server._is_sse() is False
+
+    def test_not_sse_for_stdio_config(self):
+        """Stdio config (no url) is not SSE."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("stdio-srv")
+        server._config = {"command": "npx", "args": []}
+        assert server._is_sse() is False
+
+    def test_sse_unavailable_raises(self):
+        """_run_sse raises ImportError when SSE client is not available."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("sse-srv")
+        config = {"url": "https://example.com/mcp/sse"}
+        async def _test():
+            with patch("tools.mcp_tool._MCP_SSE_AVAILABLE", False):
+                with pytest.raises(ImportError, match="SSE transport"):
+                    await server._run_sse(config)
+        asyncio.run(_test())
+
+    def test_sse_and_http_both_true_for_sse_url(self):
+        """An SSE URL satisfies both _is_http() and _is_sse()."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("sse-srv")
+        server._config = {"url": "https://example.com/mcp/sse"}
+        assert server._is_http() is True
+        assert server._is_sse() is True
+
+    def test_http_only_for_non_sse_url(self):
+        """A non-SSE HTTP URL satisfies _is_http() but not _is_sse()."""
+        from tools.mcp_tool import MCPServerTask
+        server = MCPServerTask("http-srv")
+        server._config = {"url": "https://example.com/mcp"}
+        assert server._is_http() is True
+        assert server._is_sse() is False
+
+
+class TestGetMcpStatusTransport:
+    """Tests for transport field in get_mcp_status()."""
+
+    def test_sse_transport_detected_in_status(self):
+        """SSE URL shows transport='sse' in status output."""
+        from tools.mcp_tool import get_mcp_status, _servers
+        config = {"mcp_servers": {"my-sse": {"url": "https://example.com/mcp/sse"}}}
+        with patch("hermes_cli.config.load_config", return_value=config), \
+             patch.dict(_servers, {}, clear=True):
+            status = get_mcp_status()
+        assert len(status) == 1
+        assert status[0]["name"] == "my-sse"
+        assert status[0]["transport"] == "sse"
+
+    def test_http_transport_in_status(self):
+        """Regular HTTP URL shows transport='http' in status output."""
+        from tools.mcp_tool import get_mcp_status, _servers
+        config = {"mcp_servers": {"my-http": {"url": "https://example.com/mcp"}}}
+        with patch("hermes_cli.config.load_config", return_value=config), \
+             patch.dict(_servers, {}, clear=True):
+            status = get_mcp_status()
+        assert len(status) == 1
+        assert status[0]["transport"] == "http"
+
+    def test_stdio_transport_in_status(self):
+        """Stdio server shows transport='stdio' in status output."""
+        from tools.mcp_tool import get_mcp_status, _servers
+        config = {"mcp_servers": {"my-stdio": {"command": "npx", "args": ["-y", "test"]}}}
+        with patch("hermes_cli.config.load_config", return_value=config), \
+             patch.dict(_servers, {}, clear=True):
+            status = get_mcp_status()
+        assert len(status) == 1
+        assert status[0]["transport"] == "stdio"

tools/mcp_tool.py

@@ -89,6 +89,7 @@
 
 _MCP_AVAILABLE = False
 _MCP_HTTP_AVAILABLE = False
+_MCP_SSE_AVAILABLE = False
 _MCP_SAMPLING_TYPES = False
 _MCP_NOTIFICATION_TYPES = False
 _MCP_MESSAGE_HANDLER_SUPPORTED = False
@@ -101,6 +102,11 @@
         _MCP_HTTP_AVAILABLE = True
     except ImportError:
         _MCP_HTTP_AVAILABLE = False
+    try:
+        from mcp.client.sse import sse_client
+        _MCP_SSE_AVAILABLE = True
+    except ImportError:
+        _MCP_SSE_AVAILABLE = False
     # Prefer the non-deprecated API (mcp >= 1.24.0); fall back to the
     # deprecated wrapper for older SDK versions.
     try:
@@ -851,12 +857,16 @@ async def _run_stdio(self, config: dict):
                 await self._shutdown_event.wait()
 
     async def _run_http(self, config: dict):
-        """Run the server using HTTP/StreamableHTTP transport."""
-        if not _MCP_HTTP_AVAILABLE:
+        """Run the server using HTTP transport.
+
+        Tries Streamable HTTP first; if that fails (e.g. server only
+        supports the legacy SSE protocol), falls back to SSE transport.
+        """
+        if not _MCP_HTTP_AVAILABLE and not _MCP_SSE_AVAILABLE:
             raise ImportError(
                 f"MCP server '{self.name}' requires HTTP transport but "
-                "mcp.client.streamable_http is not available. "
-                "Upgrade the mcp package to get HTTP support."
+                "neither mcp.client.streamable_http nor mcp.client.sse "
+                "is available. Upgrade the mcp package to get HTTP support."
             )
 
         url = config["url"]
@@ -876,6 +886,44 @@ async def _run_http(self, config: dict):
         if _MCP_NOTIFICATION_TYPES and _MCP_MESSAGE_HANDLER_SUPPORTED:
             sampling_kwargs["message_handler"] = self._make_message_handler()
 
+        # Try Streamable HTTP first
+        if _MCP_HTTP_AVAILABLE:
+            try:
+                await self._run_http_streamable(
+                    url, headers, connect_timeout, _oauth_auth, sampling_kwargs
+                )
+                return
+            except Exception as exc:
+                from mcp.shared.exceptions import McpError
+                if isinstance(exc, McpError) and "Session terminated" in str(exc):
+                    logger.info(
+                        "MCP server '%s': Streamable HTTP failed with "
+                        "'Session terminated' — attempting SSE fallback",
+                        self.name,
+                    )
+                elif _MCP_SSE_AVAILABLE:
+                    logger.info(
+                        "MCP server '%s': Streamable HTTP failed (%s) — "
+                        "attempting SSE fallback",
+                        self.name, exc,
+                    )
+                else:
+                    raise
+
+        # Fall back to SSE transport
+        if _MCP_SSE_AVAILABLE:
+            await self._run_http_sse(
+                url, headers, connect_timeout, _oauth_auth, sampling_kwargs
+            )
+        else:
+            raise ImportError(
+                f"MCP server '{self.name}' Streamable HTTP failed and "
+                "SSE transport (mcp.client.sse) is not available."
+            )
+
+    async def _run_http_streamable(self, url, headers, connect_timeout,
+                                   _oauth_auth, sampling_kwargs):
+        """Connect via Streamable HTTP transport."""
         if _MCP_NEW_HTTP:
             # New API (mcp >= 1.24.0): build an explicit httpx.AsyncClient
             # matching the SDK's own create_mcp_http_client defaults.
@@ -920,6 +968,111 @@ async def _run_http(self, config: dict):
                     self._ready.set()
                     await self._shutdown_event.wait()
 
+    async def _run_sse(self, config: dict):
+        """Connect via SSE transport directly (explicit detection).
+
+        Used when the server is explicitly configured with ``transport: sse``
+        or the URL path ends with ``/sse``.  Bypasses Streamable HTTP entirely
+        and connects straight to the SSE endpoint with keepalive support.
+
+        Based on work by @amiller in #5981.
+        """
+        if not _MCP_SSE_AVAILABLE:
+            raise ImportError(
+                f"MCP server '{self.name}' requires SSE transport but "
+                "mcp.client.sse is not available. Upgrade the mcp package "
+                "to get SSE support."
+            )
+
+        url = config["url"]
+        headers = dict(config.get("headers") or {})
+        connect_timeout = config.get("connect_timeout", _DEFAULT_CONNECT_TIMEOUT)
+
+        # OAuth 2.1 PKCE
+        _oauth_auth = None
+        if self._auth_type == "oauth":
+            try:
+                from tools.mcp_oauth import build_oauth_auth
+                _oauth_auth = build_oauth_auth(self.name, url)
+            except Exception as exc:
+                logger.warning("MCP OAuth setup failed for '%s': %s", self.name, exc)
+
+        sampling_kwargs = self._sampling.session_kwargs() if self._sampling else {}
+        if _MCP_NOTIFICATION_TYPES and _MCP_MESSAGE_HANDLER_SUPPORTED:
+            sampling_kwargs["message_handler"] = self._make_message_handler()
+
+        await self._run_http_sse(
+            url, headers, connect_timeout, _oauth_auth, sampling_kwargs
+        )
+
+    async def _sse_keepalive(self, session, interval: float = 60.0):
+        """Send periodic pings to keep the SSE connection alive.
+
+        Some SSE MCP servers (e.g. Supermemory on Cloudflare Workers) close
+        idle connections after a few minutes of silence.  This coroutine sends
+        a lightweight MCP ping every *interval* seconds so the server sees the
+        client as active.  On failure the loop exits silently — the main
+        ``_run_http_sse`` coroutine will notice the dead session on the next
+        tool call or when the SSE stream raises.
+        """
+        try:
+            while True:
+                await asyncio.sleep(interval)
+                try:
+                    await session.send_ping()
+                except Exception:
+                    logger.debug(
+                        "MCP server '%s': SSE keepalive ping failed",
+                        self.name,
+                    )
+                    break
+        except asyncio.CancelledError:
+            pass
+
+    async def _run_http_sse(self, url, headers, connect_timeout,
+                            _oauth_auth, sampling_kwargs):
+        """Connect via legacy SSE transport (GET-based connect + POST messages).
+
+        Some MCP servers (e.g. Supermemory V4 on Cloudflare Workers) use the
+        SSE protocol where the client opens a GET SSE stream to receive an
+        ``endpoint`` event, then POSTs messages to that endpoint URL.
+
+        A background keepalive task sends periodic pings to prevent the server
+        from closing idle SSE connections.
+        """
+        from mcp.client.sse import sse_client
+
+        sse_kwargs: dict = {
+            "timeout": float(connect_timeout),
+            "sse_read_timeout": 300.0,
+        }
+        if headers:
+            sse_kwargs["headers"] = headers
+        if _oauth_auth is not None:
+            sse_kwargs["auth"] = _oauth_auth
+
+        async with sse_client(url, **sse_kwargs) as (
+            read_stream, write_stream,
+        ):
+            async with ClientSession(read_stream, write_stream, **sampling_kwargs) as session:
+                await session.initialize()
+                self.session = session
+                await self._discover_tools()
+                self._ready.set()
+
+                # Start keepalive to prevent idle disconnect
+                keepalive = asyncio.ensure_future(
+                    self._sse_keepalive(session)
+                )
+                try:
+                    await self._shutdown_event.wait()
+                finally:
+                    keepalive.cancel()
+                    try:
+                        await keepalive
+                    except (asyncio.CancelledError, Exception):
+                        pass
+
     async def _discover_tools(self):
         """Discover tools from the connected session."""
         if self.session is None:
@@ -962,7 +1115,12 @@ async def run(self, config: dict):
         while True:
             try:
                 if self._is_http():
-                    await self._run_http(config)
+                    # If explicitly SSE (config or URL path), go straight to SSE.
+                    # Otherwise try Streamable HTTP with SSE fallback.
+                    if self._is_sse():
+                        await self._run_sse(config)
+                    else:
+                        await self._run_http(config)
                 else:
                     await self._run_stdio(config)
                 # Normal exit (shutdown requested) -- break out
@@ -1875,7 +2033,13 @@ def get_mcp_status() -> List[dict]:
         active_servers = dict(_servers)
 
     for name, cfg in configured.items():
-        transport = "http" if "url" in cfg else "stdio"
+        # Determine transport type: sse, http, or stdio
+        if "url" in cfg:
+            _tmp = MCPServerTask(name)
+            _tmp._config = cfg
+            transport = "sse" if _tmp._is_sse() else "http"
+        else:
+            transport = "stdio"
         server = active_servers.get(name)
         if server and server.session is not None:
             entry = {