Merge pull request #126 from IBM/streamablehttp_servers_support

Streamablehttp servers support

Author: crivetimihai

mcpgateway/admin.py

@@ -755,6 +755,7 @@ async def admin_add_gateway(request: Request, db: Session = Depends(get_db), use
         name=form["name"],
         url=form["url"],
         description=form.get("description"),
+        transport=form.get("transport", "SSE"),
         auth_type=form.get("auth_type", ""),
         auth_username=form.get("auth_username", ""),
         auth_password=form.get("auth_password", ""),
@@ -769,12 +770,12 @@ async def admin_add_gateway(request: Request, db: Session = Depends(get_db), use
     except Exception as ex:
         if isinstance(ex, GatewayConnectionError):
             return RedirectResponse(f"{root_path}/admin#gateways", status_code=502)
-        elif isinstance(ex, ValueError):
+        if isinstance(ex, ValueError):
             return RedirectResponse(f"{root_path}/admin#gateways", status_code=400)
-        elif isinstance(ex, RuntimeError):
-            return RedirectResponse(f"{root_path}/admin#gateways", status_code=500)
-        else:
+        if isinstance(ex, RuntimeError):
             return RedirectResponse(f"{root_path}/admin#gateways", status_code=500)
+        
+        return RedirectResponse(f"{root_path}/admin#gateways", status_code=500)
 
 
 @admin_router.post("/gateways/{gateway_id}/edit")
@@ -806,6 +807,7 @@ async def admin_edit_gateway(
         name=form["name"],
         url=form["url"],
         description=form.get("description"),
+        transport=form.get("transport", "SSE"),
         auth_type=form.get("auth_type", None),
         auth_username=form.get("auth_username", None),
         auth_password=form.get("auth_password", None),

mcpgateway/db.py

@@ -933,6 +933,7 @@ class Gateway(Base):
     name: Mapped[str] = mapped_column(unique=True)
     url: Mapped[str]
     description: Mapped[Optional[str]]
+    transport: Mapped[str] = mapped_column(default="SSE")
     capabilities: Mapped[Dict[str, Any]] = mapped_column(JSON)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))

mcpgateway/schemas.py

@@ -271,15 +271,15 @@ class ToolCreate(BaseModelWithConfig):
       - Unique tool name
       - Valid endpoint URL
       - Valid JSON Schema for input validation
-      - Request type (HTTP method)
       - Integration type: 'MCP' for MCP-compliant tools or 'REST' for REST integrations
+      - Request type (For REST-GET,POST,PUT,DELETE and for MCP-SSE,STDIO,STREAMABLEHTTP)
       - Optional authentication credentials: BasicAuth or BearerTokenAuth or HeadersAuth.
     """
 
     name: str = Field(..., description="Unique name for the tool")
     url: Union[str, AnyHttpUrl] = Field(None, description="Tool endpoint URL")
     description: Optional[str] = Field(None, description="Tool description")
-    request_type: Literal["GET", "POST", "PUT", "DELETE", "SSE", "STDIO"] = Field("SSE", description="HTTP method to be used for invoking the tool")
+    request_type: Literal["GET", "POST", "PUT", "DELETE", "SSE", "STDIO", "STREAMABLEHTTP"] = Field("SSE", description="HTTP method to be used for invoking the tool")
     integration_type: Literal["MCP", "REST"] = Field("MCP", description="Tool integration type: 'MCP' for MCP-compliant tools, 'REST' for REST integrations")
     headers: Optional[Dict[str, str]] = Field(None, description="Additional headers to send when invoking the tool")
     input_schema: Optional[Dict[str, Any]] = Field(
@@ -340,7 +340,7 @@ class ToolUpdate(BaseModelWithConfig):
     name: Optional[str] = Field(None, description="Unique name for the tool")
     url: Optional[Union[str, AnyHttpUrl]] = Field(None, description="Tool endpoint URL")
     description: Optional[str] = Field(None, description="Tool description")
-    request_type: Optional[Literal["GET", "POST", "PUT", "DELETE", "SSE", "STDIO"]] = Field(None, description="HTTP method to be used for invoking the tool")
+    request_type: Optional[Literal["GET", "POST", "PUT", "DELETE", "SSE", "STDIO", "STREAMABLEHTTP"]] = Field(None, description="HTTP method to be used for invoking the tool")
     integration_type: Optional[Literal["MCP", "REST"]] = Field(None, description="Tool integration type")
     headers: Optional[Dict[str, str]] = Field(None, description="Additional headers to send when invoking the tool")
     input_schema: Optional[Dict[str, Any]] = Field(None, description="JSON Schema for validating tool parameters")
@@ -641,6 +641,7 @@ class GatewayCreate(BaseModelWithConfig):
     name: str = Field(..., description="Unique name for the gateway")
     url: Union[str, AnyHttpUrl] = Field(..., description="Gateway endpoint URL")
     description: Optional[str] = Field(None, description="Gateway description")
+    transport: str = Field(default="SSE", description="Transport used by MCP server: SSE or STREAMABLEHTTP")
 
     # Authorizations
     auth_type: Optional[str] = Field(None, description="Type of authentication: basic, bearer, headers, or none")
@@ -752,6 +753,7 @@ class GatewayUpdate(BaseModelWithConfig):
     name: Optional[str] = Field(None, description="Unique name for the gateway")
     url: Optional[Union[str, AnyHttpUrl]] = Field(None, description="Gateway endpoint URL")
     description: Optional[str] = Field(None, description="Gateway description")
+    transport: str = Field(default="SSE", description="Transport used by MCP server: SSE or STREAMABLEHTTP")
 
     name: Optional[str] = Field(None, description="Unique name for the prompt")
     # Authorizations
@@ -877,6 +879,7 @@ class GatewayRead(BaseModelWithConfig):
     name: str = Field(..., description="Unique name for the gateway")
     url: str = Field(..., description="Gateway endpoint URL")
     description: Optional[str] = Field(None, description="Gateway description")
+    transport: str = Field(default="SSE", description="Transport used by MCP server: SSE or STREAMABLEHTTP")
     capabilities: Dict[str, Any] = Field(default_factory=dict, description="Gateway capabilities")
     created_at: datetime = Field(default_factory=datetime.utcnow, description="Creation timestamp")
     updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update timestamp")

mcpgateway/services/gateway_service.py

@@ -24,6 +24,7 @@
 from filelock import FileLock, Timeout
 from mcp import ClientSession
 from mcp.client.sse import sse_client
+from mcp.client.streamable_http import streamablehttp_client
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -186,7 +187,7 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
             auth_type = getattr(gateway, "auth_type", None)
             auth_value = getattr(gateway, "auth_value", {})
 
-            capabilities, tools = await self._initialize_gateway(str(gateway.url), auth_value)
+            capabilities, tools = await self._initialize_gateway(str(gateway.url), auth_value, gateway.transport)
 
             all_names = [td.name for td in tools]
 
@@ -217,6 +218,7 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
                 name=gateway.name,
                 url=str(gateway.url),
                 description=gateway.description,
+                transport=gateway.transport,
                 capabilities=capabilities,
                 last_seen=datetime.now(timezone.utc),
                 auth_type=auth_type,
@@ -311,6 +313,8 @@ async def update_gateway(self, db: Session, gateway_id: int, gateway_update: Gat
                 gateway.url = str(gateway_update.url)
             if gateway_update.description is not None:
                 gateway.description = gateway_update.description
+            if gateway_update.transport is not None:
+                gateway.transport = gateway_update.transport
 
             if getattr(gateway, "auth_type", None) is not None:
                 gateway.auth_type = gateway_update.auth_type
@@ -322,7 +326,7 @@ async def update_gateway(self, db: Session, gateway_id: int, gateway_update: Gat
             # Try to reinitialize connection if URL changed
             if gateway_update.url is not None:
                 try:
-                    capabilities, _ = await self._initialize_gateway(gateway.url, gateway.auth_value)
+                    capabilities, _ = await self._initialize_gateway(gateway.url, gateway.auth_value, gateway.transport)
                     gateway.capabilities = capabilities
                     gateway.last_seen = datetime.utcnow()
 
@@ -399,7 +403,7 @@ async def toggle_gateway_status(self, db: Session, gateway_id: int, activate: bo
                     self._active_gateways.add(gateway.url)
                     # Try to initialize if activating
                     try:
-                        capabilities, tools = await self._initialize_gateway(gateway.url, gateway.auth_value)
+                        capabilities, tools = await self._initialize_gateway(gateway.url, gateway.auth_value, gateway.transport)
                         gateway.capabilities = capabilities.dict()
                         gateway.last_seen = datetime.utcnow()
                     except Exception as e:
@@ -571,14 +575,21 @@ async def check_health_of_gateways(self, gateways: List[DbGateway]) -> bool:
                     headers = decode_auth(auth_data)
 
                     # Perform the GET and raise on 4xx/5xx
-                    async with client.stream("GET", gateway.url, headers=headers) as response:
-                        # This will raise immediately if status is 4xx/5xx
-                        response.raise_for_status()
+                    if (gateway.transport).lower() == "sse":
+                        timeout = httpx.Timeout(settings.health_check_timeout)
+                        async with client.stream("GET", gateway.url, headers=headers, timeout=timeout) as response:
+                            # This will raise immediately if status is 4xx/5xx
+                            response.raise_for_status()
+                    elif (gateway.transport).lower() == "streamablehttp":
+                        async with streamablehttp_client(url=gateway.url, headers=headers, timeout=settings.health_check_timeout) as (read_stream, write_stream, get_session_id):
+                            async with ClientSession(read_stream, write_stream) as session:
+                                # Initialize the session
+                                response = await session.initialize()
 
                     # Mark successful check
                     gateway.last_seen = datetime.utcnow()
 
-                except Exception:
+                except Exception as e:
                     await self._handle_gateway_failure(gateway)
 
         # All gateways passed
@@ -629,7 +640,7 @@ async def subscribe_events(self) -> AsyncGenerator[Dict[str, Any], None]:
         finally:
             self._event_subscribers.remove(queue)
 
-    async def _initialize_gateway(self, url: str, authentication: Optional[Dict[str, str]] = None) -> Any:
+    async def _initialize_gateway(self, url: str, authentication: Optional[Dict[str, str]] = None, transport: str = "SSE") -> Any:
         """Initialize connection to a gateway and retrieve its capabilities.
 
         Args:
@@ -676,7 +687,45 @@ async def connect_to_sse_server(server_url: str, authentication: Optional[Dict[s
 
                 return capabilities, tools
 
-            capabilities, tools = await connect_to_sse_server(url, authentication)
+            async def connect_to_streamablehttp_server(server_url: str, authentication: Optional[Dict[str, str]] = None):
+                """
+                Connect to an MCP server running with Streamable HTTP transport
+
+                Args:
+                    server_url: URL to connect to the server
+                    authentication: Authentication headers for connection to URL
+
+                Returns:
+                    list, list: List of capabilities and tools
+                """
+                if authentication is None:
+                    authentication = {}
+                # Store the context managers so they stay alive
+                decoded_auth = decode_auth(authentication)
+
+                # Use async with for both streamablehttp_client and ClientSession
+                async with streamablehttp_client(url=server_url, headers=decoded_auth) as (read_stream, write_stream, get_session_id):
+                    async with ClientSession(read_stream, write_stream) as session:
+                        # Initialize the session
+                        response = await session.initialize()
+                        # if get_session_id:
+                        #     session_id = get_session_id()
+                        #     if session_id:
+                        #         print(f"Session ID: {session_id}")
+                        capabilities = response.capabilities.model_dump(by_alias=True, exclude_none=True)
+                        response = await session.list_tools()
+                        tools = response.tools
+                        tools = [tool.model_dump(by_alias=True, exclude_none=True) for tool in tools]
+                        tools = [ToolCreate.model_validate(tool) for tool in tools]
+                        for tool in tools:
+                            tool.request_type = "STREAMABLEHTTP"
+
+                return capabilities, tools
+
+            if transport.lower() == "sse":
+                capabilities, tools = await connect_to_sse_server(url, authentication)
+            elif transport.lower() == "streamablehttp":
+                capabilities, tools = await connect_to_streamablehttp_server(url, authentication)
 
             return capabilities, tools
         except Exception as e:

mcpgateway/services/tool_service.py

@@ -25,6 +25,7 @@
 import httpx
 from mcp import ClientSession
 from mcp.client.sse import sse_client
+from mcp.client.streamable_http import streamablehttp_client
 from sqlalchemy import delete, func, not_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
@@ -534,6 +535,7 @@ async def invoke_tool(self, db: Session, name: str, arguments: Dict[str, Any]) -
 
                 success = True
             elif tool.integration_type == "MCP":
+                transport = tool.request_type.lower()
                 gateway = db.execute(select(DbGateway).where(DbGateway.id == tool.gateway_id).where(DbGateway.is_active)).scalar_one_or_none()
                 if gateway.auth_type == "bearer":
                     headers = decode_auth(gateway.auth_value)
@@ -558,10 +560,31 @@ async def connect_to_sse_server(server_url: str) -> str:
                             tool_call_result = await session.call_tool(name, arguments)
                     return tool_call_result
 
+                async def connect_to_streamablehttp_server(server_url: str) -> str:
+                    """
+                    Connect to an MCP server running with Streamable HTTP transport
+
+                    Args:
+                        server_url (str): MCP Server URL
+
+                    Returns:
+                        str: Result of tool call
+                    """
+                    # Use async with directly to manage the context
+                    async with streamablehttp_client(url=server_url, headers=headers) as (read_stream, write_stream, get_session_id):
+                        async with ClientSession(read_stream, write_stream) as session:
+                            # Initialize the session
+                            await session.initialize()
+                            tool_call_result = await session.call_tool(name, arguments)
+                    return tool_call_result
+
                 tool_gateway_id = tool.gateway_id
                 tool_gateway = db.execute(select(DbGateway).where(DbGateway.id == tool_gateway_id).where(DbGateway.is_active)).scalar_one_or_none()
 
-                tool_call_result = await connect_to_sse_server(tool_gateway.url)
+                if transport == "sse":
+                    tool_call_result = await connect_to_sse_server(tool_gateway.url)
+                elif transport == "streamablehttp":
+                    tool_call_result = await connect_to_streamablehttp_server(tool_gateway.url)
                 content = tool_call_result.model_dump(by_alias=True).get("content", [])
 
                 success = True

mcpgateway/static/admin.js

@@ -318,7 +318,7 @@ document.addEventListener("DOMContentLoaded", function () {
   );
 
   const requestTypeMap = {
-    MCP: ["SSE", "STDIO"],
+    MCP: ["SSE", "STREAMABLE", "STDIO"],
     REST: ["GET", "POST", "PUT", "PATCH", "DELETE"],
   };
 
@@ -896,6 +896,10 @@ async function viewGateway(gatewayId) {
           <p><strong>Name:</strong> ${gateway.name}</p>
           <p><strong>URL:</strong> ${gateway.url}</p>
           <p><strong>Description:</strong> ${gateway.description || "N/A"}</p>
+          <p><strong>Transport:</strong> 
+            ${gateway.transport === "STREAMABLEHTTP" ? "Streamable HTTP" : 
+              gateway.transport === "SSE" ? "SSE" : "N/A"}
+          </p>
           <p><strong>Status:</strong>
             <span class="px-2 inline-flex text-xs leading-5 font-semibold rounded-full ${gateway.isActive ? "bg-green-100 text-green-800" : "bg-red-100 text-red-800"}">
               ${gateway.isActive ? "Active" : "Inactive"}
@@ -927,6 +931,7 @@ async function editGateway(gatewayId) {
     document.getElementById("edit-gateway-url").value = gateway.url;
     document.getElementById("edit-gateway-description").value =
       gateway.description || "";
+    document.getElementById("edit-gateway-transport").value = gateway.transport;
     openModal("gateway-edit-modal");
   } catch (error) {
     console.error("Error fetching gateway details:", error);
@@ -1753,7 +1758,7 @@ function closeModal(modalId, clearId=null) {
 }
 
 const integrationRequestMap = {
-  MCP: ["SSE", "STDIO"],
+  MCP: ["SSE", "STREAMABLE", "STDIO"],
   REST: ["GET", "POST", "PUT", "PATCH", "DELETE"],
 };
 

mcpgateway/templates/admin.html

@@ -1394,6 +1394,18 @@ <h3 class="text-lg font-bold mb-4">Add New Gateway</h3>
                   class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-indigo-500 focus:ring-indigo-500"
                 ></textarea>
               </div>
+              <div>
+                <label class="block text-sm font-medium text-gray-700"
+                  >Transport Type</label
+                >
+                <select
+                  name="transport"
+                  class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-indigo-500 focus:ring-indigo-500"
+                >
+                  <option value="SSE" selected>SSE</option>
+                  <option value="STREAMABLEHTTP">Streamable HTTP</option>
+                </select>
+              </div>
               <div>
                 <label class="block text-sm font-medium text-gray-700"
                   >Authentication Type</label
@@ -2259,6 +2271,19 @@ <h3 class="text-lg font-medium text-gray-900">Edit Gateway</h3>
                       class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-indigo-500 focus:ring-indigo-500"
                     ></textarea>
                   </div>
+                  <div>
+                    <label class="block text-sm font-medium text-gray-700"
+                      >Transport Type</label
+                    >
+                    <select
+                      name="transport"
+                      id="edit-gateway-transport"
+                      class="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-indigo-500 focus:ring-indigo-500"
+                    >
+                      <option value="SSE" selected>SSE</option>
+                      <option value="STREAMABLEHTTP">Streamable HTTP</option>
+                    </select>
+                  </div>
                   <div>
                     <label class="block text-sm font-medium text-gray-700"
                       >Authentication Type</label

mcpgateway/transports/streamablehttp_transport.py

@@ -25,8 +25,8 @@
 from typing import List, Union
 from uuid import uuid4
 
-from mcp import types
 from fastapi.security.utils import get_authorization_scheme_param
+from mcp import types
 from mcp.server.lowlevel import Server
 from mcp.server.streamable_http import (
     EventCallback,