migrate mcp transport to http streamable

fix tests

refactor mcp and mcp_projects

backwards compat with SSE transport

provide streamable http option for json mcp config

remove streamable_http mgmt and update tests

src/backend/base/langflow/api/utils/mcp/__init__.py

@@ -3,14 +3,15 @@
 from langflow.api.utils.mcp.config_utils import (
     auto_configure_starter_projects_mcp,
     get_composer_streamable_http_url,
-    get_project_sse_url,
     get_project_streamable_http_url,
+    get_project_sse_url,
     get_url_by_os,
 )
 
 __all__ = [
     "auto_configure_starter_projects_mcp",
     "get_composer_streamable_http_url",
+    "get_project_streamable_http_url",
     "get_project_sse_url",
     "get_project_streamable_http_url",
     "get_url_by_os",

src/backend/base/langflow/api/v1/mcp.py

@@ -56,20 +56,8 @@ async def handle_global_call_tool(name: str, arguments: dict) -> list[types.Text
     return await handle_call_tool(name, arguments, server)
 
 
-########################################################
-# The transports handle the full ASGI response.
-# FastAPI still expects the endpoint to return
-# a Response, while Starlette's middleware
-# stream validation panics when
-# a http.response.start message
-# is encountered twice within the same stream.
-# This class nullifies the redundant
-# response to end streams gracefully.
-########################################################
-class ResponseNoOp(Response):
-    async def __call__(self, scope, receive, send) -> None:  # noqa: ARG002
-        return
-
+sse = SseServerTransport("/api/v1/mcp/")
+streamable_http_manager = StreamableHTTPSessionManager(server)
 
 def find_validation_error(exc):
     """Searches for a pydantic.ValidationError in the exception chain."""
@@ -148,73 +136,6 @@ async def handle_messages(request: Request):
         raise HTTPException(status_code=500, detail=f"Internal server error: {e}") from e
 
 
-################################################################################
-# Streamable HTTP Transport
-################################################################################
-class StreamableHTTP:
-    def __init__(self):
-        self.session_manager: StreamableHTTPSessionManager | None = None
-        self._context_stack: AsyncExitStack | None = None
-        self._started = False
-        self._start_stop_lock = asyncio.Lock()
-
-    async def start(self, *, stateless: bool = True) -> None:
-        """Create and enter the Streamable HTTP session manager lifecycle."""
-        async with self._start_stop_lock:
-            if self._started:
-                await logger.adebug("Streamable HTTP session manager already running; skipping start")
-                return
-
-            manager = StreamableHTTPSessionManager(server, stateless=stateless)
-            stack = AsyncExitStack()
-            try:
-                await stack.enter_async_context(manager.run())
-            except Exception:
-                await stack.aclose()
-                raise
-
-            self.session_manager = manager
-            self._context_stack = stack
-            self._started = True
-            await logger.adebug("Streamable HTTP session manager started")
-
-    def get_manager(self) -> StreamableHTTPSessionManager:
-        """Fetch the active Streamable HTTP session manager or raise if it is unavailable."""
-        if not self._started or self.session_manager is None:
-            raise HTTPException(status_code=503, detail="MCP Streamable HTTP transport is not initialized")
-        return self.session_manager
-
-    async def stop(self) -> None:
-        """Close the Streamable HTTP session manager context."""
-        async with self._start_stop_lock:
-            if not self._started:
-                return
-
-            try:
-                if self._context_stack is not None:
-                    await self._context_stack.aclose()
-            finally:
-                self._context_stack = None
-                self.session_manager = None
-                self._started = False
-                await logger.adebug("Streamable HTTP session manager stopped")
-
-
-_streamable_http = StreamableHTTP()
-
-
-async def start_streamable_http_manager(stateless: bool = True) -> None:  # noqa: FBT001, FBT002
-    await _streamable_http.start(stateless=stateless)
-
-
-def get_streamable_http_manager() -> StreamableHTTPSessionManager:
-    return _streamable_http.get_manager()
-
-
-async def stop_streamable_http_manager() -> None:
-    await _streamable_http.stop()
-
-
 async def _dispatch_streamable_http(
     request: Request,
     current_user: CurrentActiveMCPUser,
@@ -229,8 +150,7 @@ async def _dispatch_streamable_http(
 
     context_token = current_user_ctx.set(current_user)
     try:
-        manager = get_streamable_http_manager()
-        await manager.handle_request(request.scope, request.receive, request._send)  # noqa: SLF001
+        await streamable_http_manager.handle_request(request.scope, request.receive, request._send)  # noqa: SLF001
     except HTTPException:
         raise
     except Exception as exc:
@@ -239,17 +159,11 @@ async def _dispatch_streamable_http(
     finally:
         current_user_ctx.reset(context_token)
 
-    return ResponseNoOp()
-
-
-streamable_http_route_config = {
-    "methods": ["GET", "POST", "DELETE"],
-    "response_class": ResponseNoOp,
-}
-
+    return Response()
 
-@router.api_route("/streamable", **streamable_http_route_config)
-@router.api_route("/streamable/", **streamable_http_route_config)
+streamable_http_methods = ["GET", "POST", "DELETE"]
+@router.api_route("/streamable", methods=streamable_http_methods)
+@router.api_route("/streamable/", methods=streamable_http_methods)
 async def handle_streamable_http(request: Request, current_user: CurrentActiveMCPUser):
     """Streamable HTTP endpoint for MCP clients that support the new transport."""
     return await _dispatch_streamable_http(request, current_user)