feat: Support the Streamable HTTP transport for MCP (#6615)

<!-- Thank you for your contribution! Please review
https://microsoft.github.io/autogen/docs/Contribute before opening a
pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?

MCP Python-sdk has started to support a new transport protocol named
`Streamble HTTP` since
[v1.8.0](https://github.com/modelcontextprotocol/python-sdk/releases/tag/v1.8.0)
last month. I heard it supersedes the SSE transport. Therefore, AutoGen
have to support it as soon as possible.

## Related issue number

#6517

## Checks

- [x] I've included any doc changes needed for
<https://microsoft.github.io/autogen/>. See
<https://github.com/microsoft/autogen/blob/main/CONTRIBUTING.md> to
build and test documentation locally.
- [x] I've added tests (if relevant) corresponding to the changes
introduced in this PR.
- [x] I've made sure all auto checks have passed.

---------

Co-authored-by: Victor Dibia <victordibia@microsoft.com>
Co-authored-by: Victor Dibia <victor.dibia@gmail.com>

Author: withsmilo

python/packages/autogen-ext/pyproject.toml

@@ -144,7 +144,7 @@ semantic-kernel-all = [
 
 rich = ["rich>=13.9.4"]
 
-mcp = ["mcp>=1.6.0"]
+mcp = ["mcp>=1.8.1"]
 canvas = [
     "unidiff>=0.7.5",
 ]

python/packages/autogen-ext/src/autogen_ext/tools/mcp/__init__.py

@@ -1,9 +1,10 @@
 from ._actor import McpSessionActor
-from ._config import McpServerParams, SseServerParams, StdioServerParams
+from ._config import McpServerParams, SseServerParams, StdioServerParams, StreamableHttpServerParams
 from ._factory import mcp_server_tools
 from ._session import create_mcp_server_session
 from ._sse import SseMcpToolAdapter
 from ._stdio import StdioMcpToolAdapter
+from ._streamable_http import StreamableHttpMcpToolAdapter
 from ._workbench import McpWorkbench
 
 __all__ = [
@@ -13,6 +14,8 @@
     "StdioServerParams",
     "SseMcpToolAdapter",
     "SseServerParams",
+    "StreamableHttpMcpToolAdapter",
+    "StreamableHttpServerParams",
     "McpServerParams",
     "mcp_server_tools",
     "McpWorkbench",

python/packages/autogen-ext/src/autogen_ext/tools/mcp/_config.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from typing import Any, Literal
 
 from mcp import StdioServerParameters
@@ -18,10 +19,24 @@ class SseServerParams(BaseModel):
 
     type: Literal["SseServerParams"] = "SseServerParams"
 
-    url: str
-    headers: dict[str, Any] | None = None
-    timeout: float = 5
-    sse_read_timeout: float = 60 * 5
+    url: str  # The SSE endpoint URL.
+    headers: dict[str, Any] | None = None  # Optional headers to include in requests.
+    timeout: float = 5  # HTTP timeout for regular operations.
+    sse_read_timeout: float = 60 * 5  # Timeout for SSE read operations.
 
 
-McpServerParams = Annotated[StdioServerParams | SseServerParams, Field(discriminator="type")]
+class StreamableHttpServerParams(BaseModel):
+    """Parameters for connecting to an MCP server over Streamable HTTP."""
+
+    type: Literal["StreamableHttpServerParams"] = "StreamableHttpServerParams"
+
+    url: str  # The endpoint URL.
+    headers: dict[str, Any] | None = None  # Optional headers to include in requests.
+    timeout: timedelta = timedelta(seconds=30)  # HTTP timeout for regular operations.
+    sse_read_timeout: timedelta = timedelta(seconds=60 * 5)  # Timeout for SSE read operations.
+    terminate_on_close: bool = True
+
+
+McpServerParams = Annotated[
+    StdioServerParams | SseServerParams | StreamableHttpServerParams, Field(discriminator="type")
+]

python/packages/autogen-ext/src/autogen_ext/tools/mcp/_factory.py

@@ -1,15 +1,16 @@
 from mcp import ClientSession
 
-from ._config import McpServerParams, SseServerParams, StdioServerParams
+from ._config import McpServerParams, SseServerParams, StdioServerParams, StreamableHttpServerParams
 from ._session import create_mcp_server_session
 from ._sse import SseMcpToolAdapter
 from ._stdio import StdioMcpToolAdapter
+from ._streamable_http import StreamableHttpMcpToolAdapter
 
 
 async def mcp_server_tools(
     server_params: McpServerParams,
     session: ClientSession | None = None,
-) -> list[StdioMcpToolAdapter | SseMcpToolAdapter]:
+) -> list[StdioMcpToolAdapter | SseMcpToolAdapter | StreamableHttpMcpToolAdapter]:
     """Creates a list of MCP tool adapters that can be used with AutoGen agents.
 
     This factory function connects to an MCP server and returns adapters for all available tools.
@@ -26,14 +27,14 @@ async def mcp_server_tools(
     Args:
         server_params (McpServerParams): Connection parameters for the MCP server.
             Can be either StdioServerParams for command-line tools or
-            SseServerParams for HTTP/SSE services.
+            SseServerParams and StreamableHttpServerParams for HTTP/SSE services.
         session (ClientSession | None): Optional existing session to use. This is used
             when you want to reuse an existing connection to the MCP server. The session
             will be reused when creating the MCP tool adapters.
 
     Returns:
-        list[StdioMcpToolAdapter | SseMcpToolAdapter]: A list of tool adapters ready to use
-            with AutoGen agents.
+        list[StdioMcpToolAdapter | SseMcpToolAdapter | StreamableHttpMcpToolAdapter]:
+            A list of tool adapters ready to use with AutoGen agents.
 
     Examples:
 
@@ -200,4 +201,9 @@ async def main() -> None:
         return [StdioMcpToolAdapter(server_params=server_params, tool=tool, session=session) for tool in tools.tools]
     elif isinstance(server_params, SseServerParams):
         return [SseMcpToolAdapter(server_params=server_params, tool=tool, session=session) for tool in tools.tools]
+    elif isinstance(server_params, StreamableHttpServerParams):
+        return [
+            StreamableHttpMcpToolAdapter(server_params=server_params, tool=tool, session=session)
+            for tool in tools.tools
+        ]
     raise ValueError(f"Unsupported server params type: {type(server_params)}")

python/packages/autogen-ext/src/autogen_ext/tools/mcp/_session.py

@@ -5,8 +5,9 @@
 from mcp import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.stdio import stdio_client
+from mcp.client.streamable_http import streamablehttp_client
 
-from ._config import McpServerParams, SseServerParams, StdioServerParams
+from ._config import McpServerParams, SseServerParams, StdioServerParams, StreamableHttpServerParams
 
 
 @asynccontextmanager
@@ -24,5 +25,22 @@ async def create_mcp_server_session(
                 yield session
     elif isinstance(server_params, SseServerParams):
         async with sse_client(**server_params.model_dump(exclude={"type"})) as (read, write):
-            async with ClientSession(read_stream=read, write_stream=write) as session:
+            async with ClientSession(
+                read_stream=read,
+                write_stream=write,
+                read_timeout_seconds=timedelta(seconds=server_params.sse_read_timeout),
+            ) as session:
+                yield session
+    elif isinstance(server_params, StreamableHttpServerParams):
+        async with streamablehttp_client(**server_params.model_dump(exclude={"type"})) as (
+            read,
+            write,
+            session_id_callback,  # type: ignore[assignment, unused-variable]
+        ):
+            # TODO: Handle session_id_callback if needed
+            async with ClientSession(
+                read_stream=read,
+                write_stream=write,
+                read_timeout_seconds=server_params.sse_read_timeout,
+            ) as session:
                 yield session

python/packages/autogen-ext/src/autogen_ext/tools/mcp/_streamable_http.py

@@ -0,0 +1,121 @@
+from autogen_core import Component
+from mcp import ClientSession, Tool
+from pydantic import BaseModel
+from typing_extensions import Self
+
+from ._base import McpToolAdapter
+from ._config import StreamableHttpServerParams
+
+
+class StreamableHttpMcpToolAdapterConfig(BaseModel):
+    """Configuration for the MCP tool adapter."""
+
+    server_params: StreamableHttpServerParams
+    tool: Tool
+
+
+class StreamableHttpMcpToolAdapter(
+    McpToolAdapter[StreamableHttpServerParams],
+    Component[StreamableHttpMcpToolAdapterConfig],
+):
+    """
+    Allows you to wrap an MCP tool running over Streamable HTTP and make it available to AutoGen.
+
+    This adapter enables using MCP-compatible tools that communicate over Streamable HTTP
+    with AutoGen agents. Common use cases include integrating with remote MCP services,
+    cloud-based tools, and web APIs that implement the Model Context Protocol (MCP).
+
+    .. note::
+
+        To use this class, you need to install `mcp` extra for the `autogen-ext` package.
+
+        .. code-block:: bash
+
+            pip install -U "autogen-ext[mcp]"
+
+
+    Args:
+        server_params (StreamableHttpServerParams): Parameters for the MCP server connection,
+            including URL, headers, and timeouts.
+        tool (Tool): The MCP tool to wrap.
+        session (ClientSession, optional): The MCP client session to use. If not provided,
+            it will create a new session. This is useful for testing or when you want to
+            manage the session lifecycle yourself.
+
+    Examples:
+        Use a remote translation service that implements MCP over Streamable HTTP to
+        create tools that allow AutoGen agents to perform translations:
+
+        .. code-block:: python
+
+            import asyncio
+            from datetime import timedelta
+            from autogen_ext.models.openai import OpenAIChatCompletionClient
+            from autogen_ext.tools.mcp import StreamableHttpMcpToolAdapter, StreamableHttpServerParams
+            from autogen_agentchat.agents import AssistantAgent
+            from autogen_agentchat.ui import Console
+            from autogen_core import CancellationToken
+
+
+            async def main() -> None:
+                # Create server params for the remote MCP service
+                server_params = StreamableHttpServerParams(
+                    url="https://api.example.com/mcp",
+                    headers={"Authorization": "Bearer your-api-key", "Content-Type": "application/json"},
+                    timeout=timedelta(seconds=30),
+                    sse_read_timeout=timedelta(seconds=60 * 5),
+                    terminate_on_close=True,
+                )
+
+                # Get the translation tool from the server
+                adapter = await StreamableHttpMcpToolAdapter.from_server_params(server_params, "translate")
+
+                # Create an agent that can use the translation tool
+                model_client = OpenAIChatCompletionClient(model="gpt-4")
+                agent = AssistantAgent(
+                    name="translator",
+                    model_client=model_client,
+                    tools=[adapter],
+                    system_message="You are a helpful translation assistant.",
+                )
+
+                # Let the agent translate some text
+                await Console(
+                    agent.run_stream(task="Translate 'Hello, how are you?' to Spanish", cancellation_token=CancellationToken())
+                )
+
+
+            if __name__ == "__main__":
+                asyncio.run(main())
+
+    """
+
+    component_config_schema = StreamableHttpMcpToolAdapterConfig
+    component_provider_override = "autogen_ext.tools.mcp.StreamableHttpMcpToolAdapter"
+
+    def __init__(
+        self, server_params: StreamableHttpServerParams, tool: Tool, session: ClientSession | None = None
+    ) -> None:
+        super().__init__(server_params=server_params, tool=tool, session=session)
+
+    def _to_config(self) -> StreamableHttpMcpToolAdapterConfig:
+        """
+        Convert the adapter to its configuration representation.
+
+        Returns:
+            StreamableHttpMcpToolAdapterConfig: The configuration of the adapter.
+        """
+        return StreamableHttpMcpToolAdapterConfig(server_params=self._server_params, tool=self._tool)
+
+    @classmethod
+    def _from_config(cls, config: StreamableHttpMcpToolAdapterConfig) -> Self:
+        """
+        Create an instance of StreamableHttpMcpToolAdapter from its configuration.
+
+        Args:
+            config (StreamableHttpMcpToolAdapterConfig): The configuration of the adapter.
+
+        Returns:
+            StreamableHttpMcpToolAdapter: An instance of StreamableHttpMcpToolAdapter.
+        """
+        return cls(server_params=config.server_params, tool=config.tool)

python/packages/autogen-ext/src/autogen_ext/tools/mcp/_workbench.py

@@ -17,7 +17,7 @@
 from typing_extensions import Self
 
 from ._actor import McpSessionActor
-from ._config import McpServerParams, SseServerParams, StdioServerParams
+from ._config import McpServerParams, SseServerParams, StdioServerParams, StreamableHttpServerParams
 
 
 class McpWorkbenchConfig(BaseModel):
@@ -252,7 +252,7 @@ async def start(self) -> None:
             )
             return  # Already initialized, no need to start again
 
-        if isinstance(self._server_params, (StdioServerParams, SseServerParams)):
+        if isinstance(self._server_params, (StdioServerParams, SseServerParams, StreamableHttpServerParams)):
             self._actor = McpSessionActor(self._server_params)
             await self._actor.initialize()
             self._actor_loop = asyncio.get_event_loop()