feat: Improve OAuth error handling and frontend sync behavior

pyproject.toml

@@ -27,7 +27,7 @@ dependencies = [
     "pyarrow==19.0.0",
     "wikipedia==1.4.0",
     "qdrant-client==1.9.2",
-    "weaviate-client==4.10.2",
+    "weaviate-client>=4.10.2,<5.0.0",
     "faiss-cpu==1.9.0.post1",
     "types-cachetools>=5.5.0.20240820,<6.0.0",
     "pymongo==4.10.1",
@@ -37,7 +37,7 @@ dependencies = [
     'fastavro>=1.9.8,<2.0.0; python_version >= "3.13"',
     "redis>=5.2.1,<6.0.0",
     "metaphor-python==0.1.23",
-    'pywin32==307; sys_platform == "win32"',
+    'pywin32>=307,<400; sys_platform == "win32"',
     "langfuse==2.53.9",
     "metal_sdk==2.5.1",
     "MarkupSafe==3.0.2",
@@ -105,7 +105,8 @@ dependencies = [
     "arize-phoenix-otel>=0.6.1,<1.0.0",
     "openinference-instrumentation-langchain>=0.1.29,<0.1.52",
     # "crewai>=0.126.0",
-    "mcp>=1.10.1,<2.0.0",
+    "fastmcp==2.13.0",
+    "mcp>=1.17.0,<2.0.0",
     "uv==0.7.20",
     "scipy>=1.14.1,<1.16.2",
     "scrapegraph-py>=1.12.0,<2.0.0",

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

@@ -408,6 +408,9 @@ async def update_project_mcp_settings(
             should_start_composer = False
             should_stop_composer = False
 
+            # Store original auth settings in case we need to rollback
+            original_auth_settings = project.auth_settings
+
             # Update project-level auth settings with encryption
             if "auth_settings" in request.model_fields_set and request.auth_settings is not None:
                 auth_result = handle_auth_settings_update(
@@ -419,8 +422,6 @@ async def update_project_mcp_settings(
                 should_start_composer = auth_result["should_start_composer"]
                 should_stop_composer = auth_result["should_stop_composer"]
 
-            session.add(project)
-
             # Query flows in the project
             flows = (await session.exec(select(Flow).where(Flow.folder_id == project_id))).all()
             flows_to_update = {x.id: x for x in request.settings}
@@ -439,13 +440,17 @@ async def update_project_mcp_settings(
                     session.add(flow)
                     updated_flows.append(flow)
 
-            await session.commit()
-
             response: dict[str, Any] = {
                 "message": f"Updated MCP settings for {len(updated_flows)} flows and project auth settings"
             }
 
+            # Handle MCP Composer start/stop before committing auth settings
             if should_handle_mcp_composer:
+                # Get MCP Composer service once for all branches
+                mcp_composer_service: MCPComposerService = cast(
+                    MCPComposerService, get_service(ServiceType.MCP_COMPOSER_SERVICE)
+                )
+
                 if should_start_composer:
                     await logger.adebug(
                         f"Auth settings changed to OAuth for project {project.name} ({project_id}), "
@@ -457,23 +462,34 @@ async def update_project_mcp_settings(
                             auth_config = await _get_mcp_composer_auth_config(project)
                             await get_or_start_mcp_composer(auth_config, project.name, project_id)
                             composer_sse_url = await get_composer_sse_url(project)
+                            # Clear any previous error on success
+                            mcp_composer_service.clear_last_error(str(project_id))
                             response["result"] = {
                                 "project_id": str(project_id),
                                 "sse_url": composer_sse_url,
                                 "uses_composer": True,
                             }
                         except MCPComposerError as e:
+                            # Don't rollback auth settings - persist them so UI can show the error
+                            await logger.awarning(f"MCP Composer failed to start for project {project_id}: {e.message}")
+                            # Store the error message so it can be retrieved via composer-url endpoint
+                            mcp_composer_service.set_last_error(str(project_id), e.message)
                             response["result"] = {
                                 "project_id": str(project_id),
                                 "uses_composer": True,
                                 "error_message": e.message,
                             }
                         except Exception as e:
-                            # Unexpected errors
-                            await logger.aerror(f"Failed to get mcp composer URL for project {project_id}: {e}")
+                            # Rollback auth settings on unexpected errors
+                            await logger.aerror(
+                                f"Unexpected error starting MCP Composer for project {project_id}, "
+                                f"rolling back auth settings: {e}"
+                            )
+                            project.auth_settings = original_auth_settings
                             raise HTTPException(status_code=500, detail=str(e)) from e
                     else:
-                        # This shouldn't happen - we determined we should start composer but now we can't use it
+                        # OAuth is set but MCP Composer is disabled - save settings but return error
+                        # Don't rollback - keep the auth settings so they can be used when composer is enabled
                         await logger.aerror(
                             f"PATCH: OAuth set but MCP Composer is disabled in settings for project {project_id}"
                         )
@@ -487,10 +503,9 @@ async def update_project_mcp_settings(
                         f"Auth settings changed from OAuth for project {project.name} ({project_id}), "
                         "stopping MCP Composer"
                     )
-                    mcp_composer_service: MCPComposerService = cast(
-                        "MCPComposerService", get_service(ServiceType.MCP_COMPOSER_SERVICE)
-                    )
                     await mcp_composer_service.stop_project_composer(str(project_id))
+                    # Clear any error when user explicitly disables OAuth
+                    mcp_composer_service.clear_last_error(str(project_id))
 
                     # Provide the direct SSE URL since we're no longer using composer
                     sse_url = await get_project_sse_url(project_id)
@@ -503,6 +518,10 @@ async def update_project_mcp_settings(
                         "uses_composer": False,
                     }
 
+            # Only commit if composer started successfully (or wasn't needed)
+            session.add(project)
+            await session.commit()
+
             return response
 
     except Exception as e:
@@ -755,6 +774,20 @@ async def get_project_composer_url(
     try:
         project = await verify_project_access(project_id, current_user)
         if not should_use_mcp_composer(project):
+            # Check if there's a recent error from a failed OAuth attempt
+            mcp_composer_service: MCPComposerService = cast(
+                MCPComposerService, get_service(ServiceType.MCP_COMPOSER_SERVICE)
+            )
+            last_error = mcp_composer_service.get_last_error(str(project_id))
+
+            # If there's a recent error, return it even though OAuth is not currently active
+            # This happens when OAuth was attempted but rolled back due to an error
+            if last_error:
+                return {
+                    "project_id": str(project_id),
+                    "uses_composer": False,
+                    "error_message": last_error,
+                }
             return {
                 "project_id": str(project_id),
                 "uses_composer": False,
@@ -768,6 +801,11 @@ async def get_project_composer_url(
         try:
             await get_or_start_mcp_composer(auth_config, project.name, project_id)
             composer_sse_url = await get_composer_sse_url(project)
+            # Clear any previous error on success
+            mcp_composer_service: MCPComposerService = cast(
+                MCPComposerService, get_service(ServiceType.MCP_COMPOSER_SERVICE)
+            )
+            mcp_composer_service.clear_last_error(str(project_id))
             return {"project_id": str(project_id), "sse_url": composer_sse_url, "uses_composer": True}
         except MCPComposerError as e:
             return {"project_id": str(project_id), "uses_composer": True, "error_message": e.message}

src/backend/base/langflow/main.py

@@ -46,6 +46,10 @@
 # Ignore Pydantic deprecation warnings from Langchain
 warnings.filterwarnings("ignore", category=PydanticDeprecatedSince20)
 
+# Suppress ResourceWarning from anyio streams (SSE connections)
+warnings.filterwarnings("ignore", category=ResourceWarning, message=".*MemoryObjectReceiveStream.*")
+warnings.filterwarnings("ignore", category=ResourceWarning, message=".*MemoryObjectSendStream.*")
+
 _tasks: list[asyncio.Task] = []
 
 MAX_PORT = 65535

src/backend/base/pyproject.toml

@@ -32,11 +32,11 @@ dependencies = [
     "rich>=13.7.0,<14.0.0",
     "langchain-experimental>=0.3.4,<1.0.0",
     "sqlmodel==0.0.22",
-    "pydantic~=2.10.1",
+    "pydantic~=2.11.0",
     "pydantic-settings>=2.2.0,<3.0.0",
     "email-validator>=2.0.0",
     "typer>=0.13.0,<1.0.0",
-    "cachetools>=5.5.0,<6.0.0",
+    "cachetools>=6.0.0",
     "platformdirs>=4.2.0,<5.0.0",
     "python-multipart>=0.0.12,<1.0.0",
     "orjson==3.10.15",
@@ -80,7 +80,7 @@ dependencies = [
     "validators>=0.34.0,<1.0.0",
     "networkx>=3.4.2,<4.0.0",
     "json-repair>=0.30.3,<1.0.0",
-    "mcp~=1.10.1",
+    "mcp>=1.17.0,<2.0.0",
     "aiosqlite>=0.20.0,<1.0.0",
     "greenlet>=3.1.1,<4.0.0",
     "jsonquerylang>=1.1.1,<2.0.0",

src/frontend/src/controllers/API/queries/mcp/use-patch-flows-mcp.ts

@@ -45,7 +45,7 @@ export const usePatchFlowsMCP: useMutationFunctionType<
     PatchFlowMCPResponse,
     any,
     PatchFlowMCPRequest
-  > = mutate(["usePatchFlowsMCP"], patchFlowMCP, {
+  > = mutate(["usePatchFlowsMCP", params.project_id], patchFlowMCP, {
     onSuccess: (data, variables, context) => {
       const authSettings = (variables as unknown as PatchFlowMCPRequest)
         .auth_settings;
@@ -73,12 +73,12 @@ export const usePatchFlowsMCP: useMutationFunctionType<
       }
     },
     onSettled: () => {
-      // Use invalidateQueries instead of refetchQueries to avoid race conditions
-      // This marks the queries as stale but doesn't immediately refetch them
-      queryClient.invalidateQueries({ queryKey: ["useGetFlowsMCP"] });
+      // Invalidate only this specific project's queries to avoid affecting other projects
+      queryClient.invalidateQueries({
+        queryKey: ["useGetFlowsMCP", params.project_id],
+      });
     },
     ...options,
-    retry: 0,
   });
 
   return mutation;

src/frontend/src/controllers/API/queries/mcp/use-patch-install-mcp.ts

@@ -47,10 +47,8 @@ export const usePatchInstallMCP: useMutationFunctionType<
     PatchInstallMCPResponse,
     any,
     PatchInstallMCPBody
-  > = mutate(["usePatchInstallMCP"], patchInstallMCP, {
+  > = mutate(["usePatchInstallMCP", params.project_id], patchInstallMCP, {
     ...options,
-    retry: 0,
-
     onSuccess: (data, variables, context) => {
       queryClient.invalidateQueries({
         queryKey: ["useGetInstalledMCP", params.project_id],