[Feat]: support streamable http mcp (#8864)

Co-authored-by: openhands <[email protected]>

Author: malhotra5

openhands/cli/main.py

@@ -263,14 +263,12 @@ def on_event(event: Event) -> None:
     # Add MCP tools to the agent
     if agent.config.enable_mcp:
         # Add OpenHands' MCP server by default
-        openhands_mcp_server, openhands_mcp_stdio_servers = (
+        _, openhands_mcp_stdio_servers = (
             OpenHandsMCPConfigImpl.create_default_mcp_server_config(
                 config.mcp_host, config, None
             )
         )
-        # FIXME: OpenHands' SSE server may not be running when CLI mode is started
-        # if openhands_mcp_server:
-        #     config.mcp.sse_servers.append(openhands_mcp_server)
+
         config.mcp.stdio_servers.extend(openhands_mcp_stdio_servers)
 
         await add_mcp_tools_to_agent(agent, runtime, memory, config)

openhands/core/config/mcp_config.py

@@ -54,6 +54,10 @@ def __eq__(self, other):
             and set(self.env.items()) == set(other.env.items())
         )
 
+class MCPSHTTPServerConfig(BaseModel):
+    url: str
+    api_key: str | None = None
+
 
 class MCPConfig(BaseModel):
     """Configuration for MCP (Message Control Protocol) settings.
@@ -65,11 +69,12 @@ class MCPConfig(BaseModel):
 
     sse_servers: list[MCPSSEServerConfig] = Field(default_factory=list)
     stdio_servers: list[MCPStdioServerConfig] = Field(default_factory=list)
+    shttp_servers: list[MCPSHTTPServerConfig] = Field(default_factory=list)
 
     model_config = {'extra': 'forbid'}
 
     @staticmethod
-    def _normalize_sse_servers(servers_data: list[dict | str]) -> list[dict]:
+    def _normalize_servers(servers_data: list[dict | str]) -> list[dict]:
         """Helper method to normalize SSE server configurations."""
         normalized = []
         for server in servers_data:
@@ -82,8 +87,13 @@ def _normalize_sse_servers(servers_data: list[dict | str]) -> list[dict]:
     @model_validator(mode='before')
     def convert_string_urls(cls, data):
         """Convert string URLs to MCPSSEServerConfig objects."""
-        if isinstance(data, dict) and 'sse_servers' in data:
-            data['sse_servers'] = cls._normalize_sse_servers(data['sse_servers'])
+        if isinstance(data, dict):
+            if 'sse_servers' in data:
+                data['sse_servers'] = cls._normalize_servers(data['sse_servers'])
+
+            if 'shttp_servers' in data:
+                data['shttp_servers'] = cls._normalize_servers(data['shttp_servers'])
+
         return data
 
     def validate_servers(self) -> None:
@@ -119,7 +129,7 @@ def from_toml_section(cls, data: dict) -> dict[str, 'MCPConfig']:
         try:
             # Convert all entries in sse_servers to MCPSSEServerConfig objects
             if 'sse_servers' in data:
-                data['sse_servers'] = cls._normalize_sse_servers(data['sse_servers'])
+                data['sse_servers'] = cls._normalize_servers(data['sse_servers'])
                 servers = []
                 for server in data['sse_servers']:
                     servers.append(MCPSSEServerConfig(**server))
@@ -132,6 +142,13 @@ def from_toml_section(cls, data: dict) -> dict[str, 'MCPConfig']:
                     servers.append(MCPStdioServerConfig(**server))
                 data['stdio_servers'] = servers
 
+            if 'shttp_servers' in data:
+                data['shttp_servers'] = cls._normalize_servers(data['shttp_servers'])
+                servers = []
+                for server in data['shttp_servers']:
+                    servers.append(MCPSHTTPServerConfig(**server))
+                data['shttp_servers'] = servers
+
             # Create SSE config if present
             mcp_config = MCPConfig.model_validate(data)
             mcp_config.validate_servers()
@@ -169,7 +186,7 @@ def add_search_engine(app_config: 'OpenHandsConfig') -> MCPStdioServerConfig | N
     @staticmethod
     def create_default_mcp_server_config(
         host: str, config: 'OpenHandsConfig', user_id: str | None = None
-    ) -> tuple[MCPSSEServerConfig, list[MCPStdioServerConfig]]:
+    ) -> tuple[MCPSHTTPServerConfig, list[MCPStdioServerConfig]]:
         """
         Create a default MCP server configuration.
 
@@ -179,12 +196,13 @@ def create_default_mcp_server_config(
         Returns:
             tuple[MCPSSEServerConfig, list[MCPStdioServerConfig]]: A tuple containing the default SSE server configuration and a list of MCP stdio server configurations
         """
-        sse_server = MCPSSEServerConfig(url=f'http://{host}/mcp/sse', api_key=None)
         stdio_servers = []
         search_engine_stdio_server = OpenHandsMCPConfig.add_search_engine(config)
         if search_engine_stdio_server:
             stdio_servers.append(search_engine_stdio_server)
-        return sse_server, stdio_servers
+
+        shttp_servers = MCPSHTTPServerConfig(url=f'http://{host}/mcp/mcp', api_key=None)
+        return shttp_servers, stdio_servers
 
 
 openhands_mcp_config_cls = os.environ.get(

openhands/core/main.py

@@ -134,14 +134,11 @@ async def run_controller(
     # Add MCP tools to the agent
     if agent.config.enable_mcp:
         # Add OpenHands' MCP server by default
-        openhands_mcp_server, openhands_mcp_stdio_servers = (
+        _, openhands_mcp_stdio_servers = (
             OpenHandsMCPConfigImpl.create_default_mcp_server_config(
                 config.mcp_host, config, None
             )
         )
-        # FIXME: OpenHands' SSE server may not be running when headless mode is started
-        # if openhands_mcp_server:
-        #     config.mcp.sse_servers.append(openhands_mcp_server)
         config.mcp.stdio_servers.extend(openhands_mcp_stdio_servers)
 
         await add_mcp_tools_to_agent(agent, runtime, memory, config)

openhands/mcp/client.py

@@ -1,9 +1,11 @@
 import asyncio
+import datetime
 from contextlib import AsyncExitStack
 from typing import Optional
 
 from mcp import ClientSession
 from mcp.client.sse import sse_client
+from mcp.client.streamable_http import streamablehttp_client
 from pydantic import BaseModel, Field
 
 from openhands.core.logger import openhands_logger as logger
@@ -58,14 +60,21 @@ async def connect_with_timeout():
                 if conversation_id:
                     headers['X-OpenHands-Conversation-ID'] = conversation_id
 
+                # Convert float timeout to datetime.timedelta for consistency
+                timeout_delta = datetime.timedelta(seconds=timeout)
+
                 streams_context = sse_client(
                     url=server_url,
                     headers=headers if headers else None,
                     timeout=timeout,
                 )
                 streams = await self.exit_stack.enter_async_context(streams_context)
+                # For SSE client, we only get read_stream and write_stream (2 values)
+                read_stream, write_stream = streams
                 self.session = await self.exit_stack.enter_async_context(
-                    ClientSession(*streams)
+                    ClientSession(
+                        read_stream, write_stream, read_timeout_seconds=timeout_delta
+                    )
                 )
                 await self._initialize_and_list_tools()
 
@@ -117,6 +126,77 @@ async def call_tool(self, tool_name: str, args: dict):
             raise RuntimeError('Client session is not available.')
         return await self.session.call_tool(name=tool_name, arguments=args)
 
+    async def connect_shttp(
+        self,
+        server_url: str,
+        api_key: str | None = None,
+        conversation_id: str | None = None,
+        timeout: float = 30.0,
+    ) -> None:
+        """Connect to an MCP server using StreamableHTTP transport.
+
+        Args:
+            server_url: The URL of the StreamableHTTP server to connect to.
+            api_key: Optional API key for authentication.
+            conversation_id: Optional conversation ID for session tracking.
+            timeout: Connection timeout in seconds. Default is 30 seconds.
+        """
+        if not server_url:
+            raise ValueError('Server URL is required.')
+        if self.session:
+            await self.disconnect()
+
+        try:
+            # Use asyncio.wait_for to enforce the timeout
+            async def connect_with_timeout():
+                headers = (
+                    {
+                        'Authorization': f'Bearer {api_key}',
+                        's': api_key,  # We need this for action execution server's MCP Router
+                        'X-Session-API-Key': api_key,  # We need this for Remote Runtime
+                    }
+                    if api_key
+                    else {}
+                )
+
+                if conversation_id:
+                    headers['X-OpenHands-Conversation-ID'] = conversation_id
+
+                # Convert float timeout to datetime.timedelta
+                timeout_delta = datetime.timedelta(seconds=timeout)
+                sse_read_timeout_delta = datetime.timedelta(
+                    seconds=timeout * 10
+                )  # 10x longer for read timeout
+
+                streams_context = streamablehttp_client(
+                    url=server_url,
+                    headers=headers if headers else None,
+                    timeout=timeout_delta,
+                    sse_read_timeout=sse_read_timeout_delta,
+                )
+                streams = await self.exit_stack.enter_async_context(streams_context)
+                # For StreamableHTTP client, we get read_stream, write_stream, and get_session_id (3 values)
+                read_stream, write_stream, _ = streams
+                self.session = await self.exit_stack.enter_async_context(
+                    ClientSession(
+                        read_stream, write_stream, read_timeout_seconds=timeout_delta
+                    )
+                )
+                await self._initialize_and_list_tools()
+
+            # Apply timeout to the entire connection process
+            await asyncio.wait_for(connect_with_timeout(), timeout=timeout)
+        except asyncio.TimeoutError:
+            logger.error(
+                f'Connection to {server_url} timed out after {timeout} seconds'
+            )
+            await self.disconnect()  # Clean up resources
+            raise  # Re-raise the TimeoutError
+        except Exception as e:
+            logger.error(f'Error connecting to {server_url}: {str(e)}')
+            await self.disconnect()  # Clean up resources
+            raise
+
     async def disconnect(self) -> None:
         """Disconnect from the MCP server and clean up resources."""
         if self.session:

openhands/mcp/utils.py

@@ -4,8 +4,10 @@
 if TYPE_CHECKING:
     from openhands.controller.agent import Agent
 
+
 from openhands.core.config.mcp_config import (
     MCPConfig,
+    MCPSHTTPServerConfig,
     MCPSSEServerConfig,
 )
 from openhands.core.config.openhands_config import OpenHandsConfig
@@ -48,7 +50,9 @@ def convert_mcp_clients_to_tools(mcp_clients: list[MCPClient] | None) -> list[di
 
 
 async def create_mcp_clients(
-    sse_servers: list[MCPSSEServerConfig], conversation_id: str | None = None
+    sse_servers: list[MCPSSEServerConfig],
+    shttp_servers: list[MCPSHTTPServerConfig],
+    conversation_id: str | None = None,
 ) -> list[MCPClient]:
     import sys
 
@@ -59,42 +63,60 @@ async def create_mcp_clients(
         )
         return []
 
-    mcp_clients: list[MCPClient] = []
-    # Initialize SSE connections
-    if sse_servers:
-        for server_url in sse_servers:
-            logger.info(
-                f'Initializing MCP agent for {server_url} with SSE connection...'
-            )
+    servers: list[MCPSSEServerConfig | MCPSHTTPServerConfig] = sse_servers.copy()
+    servers.extend(shttp_servers.copy())
 
-            client = MCPClient()
-            try:
+    if not servers:
+        return []
+
+    mcp_clients = []
+
+    for server in servers:
+        is_sse = isinstance(server, MCPSSEServerConfig)
+        connection_type = 'SSE' if is_sse else 'SHTTP'
+        logger.info(
+            f'Initializing MCP agent for {server} with {connection_type} connection...'
+        )
+        client = MCPClient()
+
+        try:
+            if is_sse:
                 await client.connect_sse(
-                    server_url.url,
-                    api_key=server_url.api_key,
+                    server.url,
+                    api_key=server.api_key,
                     conversation_id=conversation_id,
                 )
-                # Only add the client to the list after a successful connection
-                mcp_clients.append(client)
-                logger.info(f'Connected to MCP server {server_url} via SSE')
-            except Exception as e:
-                logger.error(
-                    f'Failed to connect to {server_url}: {str(e)}', exc_info=True
+            else:
+                await client.connect_shttp(
+                    server.url,
+                    api_key=server.api_key,
+                    conversation_id=conversation_id,
                 )
-                try:
-                    await client.disconnect()
-                except Exception as disconnect_error:
-                    logger.error(
-                        f'Error during disconnect after failed connection: {str(disconnect_error)}'
-                    )
 
+            # Only add the client to the list after a successful connection
+            mcp_clients.append(client)
+
+        except Exception as e:
+            logger.error(f'Failed to connect to {server}: {str(e)}', exc_info=True)
+            try:
+                await client.disconnect()
+            except Exception as disconnect_error:
+                logger.error(
+                    f'Error during disconnect after failed connection: {str(disconnect_error)}'
+                )
     return mcp_clients
 
 
-async def fetch_mcp_tools_from_config(mcp_config: MCPConfig) -> list[dict]:
+async def fetch_mcp_tools_from_config(
+    mcp_config: MCPConfig, conversation_id: str | None = None
+) -> list[dict]:
     """
     Retrieves the list of MCP tools from the MCP clients.
 
+    Args:
+        mcp_config: The MCP configuration
+        conversation_id: Optional conversation ID to associate with the MCP clients
+
     Returns:
         A list of tool dictionaries. Returns an empty list if no connections could be established.
     """
@@ -111,7 +133,7 @@ async def fetch_mcp_tools_from_config(mcp_config: MCPConfig) -> list[dict]:
         logger.debug(f'Creating MCP clients with config: {mcp_config}')
         # Create clients - this will fetch tools but not maintain active connections
         mcp_clients = await create_mcp_clients(
-            mcp_config.sse_servers,
+            mcp_config.sse_servers, mcp_config.shttp_servers, conversation_id
         )
 
         if not mcp_clients:

openhands/runtime/impl/action_execution/action_execution_client.py

@@ -464,7 +464,7 @@ async def call_tool_mcp(self, action: MCPAction) -> Observation:
         )
 
         # Create clients for this specific operation
-        mcp_clients = await create_mcp_clients(updated_mcp_config.sse_servers, self.sid)
+        mcp_clients = await create_mcp_clients(updated_mcp_config.sse_servers, updated_mcp_config.shttp_servers, self.sid)
 
         # Call the tool and return the result
         # No need for try/finally since disconnect() is now just resetting state

openhands/server/app.py

@@ -1,3 +1,4 @@
+import contextlib
 import warnings
 from contextlib import asynccontextmanager
 from typing import AsyncIterator
@@ -29,6 +30,20 @@
 from openhands.server.routes.trajectory import app as trajectory_router
 from openhands.server.shared import conversation_manager
 
+mcp_app = mcp_server.http_app(path='/mcp')
+
+
+def combine_lifespans(*lifespans):
+    # Create a combined lifespan to manage multiple session managers
+    @contextlib.asynccontextmanager
+    async def combined_lifespan(app):
+        async with contextlib.AsyncExitStack() as stack:
+            for lifespan in lifespans:
+                await stack.enter_async_context(lifespan(app))
+            yield
+
+    return combined_lifespan
+
 
 @asynccontextmanager
 async def _lifespan(app: FastAPI) -> AsyncIterator[None]:
@@ -40,8 +55,8 @@ async def _lifespan(app: FastAPI) -> AsyncIterator[None]:
     title='OpenHands',
     description='OpenHands: Code Less, Make More',
     version=__version__,
-    lifespan=_lifespan,
-    routes=[Mount(path='/mcp', app=mcp_server.sse_app())],
+    lifespan=combine_lifespans(_lifespan, mcp_app.lifespan),
+    routes=[Mount(path='/mcp', app=mcp_app)],
 )