graceful termination of mcp projects /streamable endpoint

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

@@ -338,7 +338,7 @@ async def _dispatch_project_streamable_http(
         current_project_ctx.reset(project_token)
         current_user_ctx.reset(user_token)
 
-    return Response()
+    return ResponseNoOp(status_code=200)
 
 
 @router.get("/{project_id}/sse", response_class=HTMLResponse)
@@ -434,11 +434,14 @@ async def handle_project_messages(
 
 
 # Streamable HTTP transport
-streamable_http_methods = ["GET", "POST", "DELETE"]
+streamable_http_route_config = {
+    "methods": ["GET", "POST", "DELETE"],
+    "response_class": ResponseNoOp,
+}
 
 
-@router.api_route("/{project_id}/streamable", methods=streamable_http_methods)
-@router.api_route("/{project_id}/streamable/", methods=streamable_http_methods)
+@router.api_route("/{project_id}/streamable", **streamable_http_route_config)
+@router.api_route("/{project_id}/streamable/", **streamable_http_route_config)
 async def handle_project_streamable_http(
     project_id: UUID,
     request: Request,
@@ -1030,12 +1033,7 @@ def _args_reference_urls(args: Sequence[Any] | None, urls: list[str]) -> bool:
     """Check whether the given args list references any of the provided URLs."""
     if not args or not urls:
         return False
-    args_strings = [arg for arg in args if isinstance(arg, str)]
-    if not args_strings:
-        return False
-    args_set = set(args_strings)
-    last_arg = args_strings[-1]
-    return any((url == last_arg) or (url in args_set) for url in urls)
+    return bool({arg for arg in args if isinstance(arg, str)}.intersection(urls))
 
 
 def config_contains_server_url(config_data: dict, urls: Sequence[str] | str) -> bool:
@@ -1249,8 +1247,8 @@ async def _run_session_manager(self, *, task_status: TaskStatus[None] = anyio.TA
         """Own the lifecycle of the project's Streamable HTTP session manager."""
         try:
             async with self.session_manager.run():
-                self._manager_started = True
-                task_status.started()
+                self._manager_started = True # set flag before unblocking task (ensures waiting requests proceed)
+                task_status.started() # unblock
                 await anyio.sleep_forever()
         except anyio.get_cancelled_exc_class():
             await logger.adebug(f"Streamable HTTP manager cancelled for project {self.project_id}")
@@ -1313,11 +1311,11 @@ async def stop(self) -> None:
         async with self._start_stop_lock:
             if not self._started:
                 return
-            try:
-                if self._task_group is not None:
-                    self._task_group.cancel_scope.cancel()
-                if self._tg_task is not None:
-                    await self._tg_task
+            try: # https://anyio.readthedocs.io/en/stable/cancellation.html, https://docs.python.org/3/library/asyncio-task.html#asyncio.Task.cancel
+                self._task_group.cancel_scope.cancel() # type: ignore[union-attr]
+                await self._tg_task  # type: ignore[misc]
+            except Exception as e:  # noqa: BLE001
+                await logger.aexception(f"Failed to stop project task group: {e}")
             finally:
                 self._cleanup()
                 await logger.adebug("Project MCP task group stopped")

src/backend/tests/unit/api/v1/test_mcp_projects.py

@@ -10,6 +10,7 @@
 from httpx import AsyncClient
 from langflow.api.v1.mcp_projects import (
     ProjectMCPServer,
+    _args_reference_urls,
     get_project_mcp_server,
     init_mcp_servers,
     project_mcp_servers,
@@ -28,6 +29,26 @@
 pytestmark = pytest.mark.asyncio
 
 
+@pytest.mark.parametrize(
+    ("args", "urls", "expected"),
+    [
+        (None, ["https://langflow.local/sse"], False),
+        ([], ["https://langflow.local/sse"], False),
+        ([123, {"url": "foo"}], ["https://langflow.local/sse"], False),
+        (["https://langflow.local/sse", 42], ["https://langflow.local/sse"], True),
+        (["alpha", "beta"], [], False),
+    ],
+)
+def test_args_reference_urls_filters_strings_only(args, urls, expected):
+    assert _args_reference_urls(args, urls) is expected
+
+
+def test_args_reference_urls_matches_non_last_string_argument():
+    args = ["match-me", "other"]
+    urls = ["match-me"]
+    assert _args_reference_urls(args, urls) is True
+
+
 @pytest.fixture
 def mock_project(active_user):
     """Fixture to provide a mock project linked to the active user."""
@@ -82,8 +103,11 @@ def mock_streamable_http_manager():
         async_cm = AsyncContextManagerMock()
         manager_instance.run = MagicMock(return_value=async_cm)
 
-        # Mock handle_request as an async method (this is what gets called, not handle_post_message)
-        manager_instance.handle_request = AsyncMock()
+        async def _fake_handle_request(scope, receive, send): # noqa: ARG001
+            await send({"type": "http.response.start", "status": 200, "headers": []})
+            await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+        manager_instance.handle_request = AsyncMock(side_effect=_fake_handle_request)
 
         # Make the class constructor return our mocked instance
         mock_class.return_value = manager_instance