feat: Add robust startup retry logic and port conflict resolution (#10347)

* fix mcp timeout and settings

* ✨ (service.py): add new methods to read process output, ensure port availability, log startup error details, and normalize config values
🐛 (service.py): fix handling of process output, error messages, and port availability to improve error handling and logging during startup

* ✨ (service.py): add tracking of which project is using which port to prevent conflicts
🐛 (service.py): check if another project is using the same port before starting a new project to avoid conflicts
🐛 (service.py): handle releasing port and cleaning up port tracking when stopping a project to prevent memory leaks
🐛 (service.py): re-check port availability before each retry when starting a project to prevent race conditions
🐛 (service.py): register the port used by a project to prevent other projects from using the same port
🐛 (McpServerTab.tsx): invalidate MCP project data and composer URL queries to refresh auth settings and OAuth server info
🐛 (McpServerTab.tsx): clear waiting state if the auth type is not OAuth

* fix killing port on tab changes

* add not persist oauth on failure start

* add oath check on available port

* add improvements mcp service

* fix: suppress SSE streaming warnings and handle connection closed errors

- Add warning filters for ResourceWarning from anyio streams
- Handle SSE connection closed errors gracefully
- Prevent log spam from normal SSE lifecycle events

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

* add host and port to uvx command, safe redact value

* fix error messages state not been displayed

* fix mypy errors

* [autofix.ci] apply automated fixes

* add the correct auth test

---------

Co-authored-by: Claude <[email protected]>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: 3 people

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

@@ -385,6 +385,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(
@@ -396,8 +399,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}
@@ -416,13 +417,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}), "
@@ -434,26 +439,37 @@ 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
                         await logger.aerror(
                             f"PATCH: OAuth set but MCP Composer is disabled in settings for project {project_id}"
                         )
+                        # Don't rollback - keep the auth settings so they can be used when composer is enabled
                         response["result"] = {
                             "project_id": str(project_id),
                             "uses_composer": False,
@@ -464,10 +480,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)
@@ -480,6 +495,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:
@@ -731,7 +750,22 @@ async def get_project_composer_url(
     """
     try:
         project = await verify_project_access(project_id, current_user)
+
+        # 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 not should_use_mcp_composer(project):
+            # 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,

src/backend/base/langflow/main.py

@@ -47,6 +47,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
@@ -458,6 +462,17 @@ async def exception_handler(_request: Request, exc: Exception):
                 status_code=exc.status_code,
                 content={"message": str(exc.detail)},
             )
+
+        # Suppress known SSE streaming errors that are harmless
+        exc_str = str(exc)
+        if "Unexpected message" in exc_str and "http.response.start" in exc_str:
+            # This is a known issue with SSE connections being closed
+            await logger.adebug(f"SSE connection closed: {exc_str}")
+            return JSONResponse(
+                status_code=200,
+                content={"message": "Connection closed"},
+            )
+
         await logger.aerror(f"unhandled error: {exc}", exc_info=exc)
 
         await log_exception_to_telemetry(exc, "handler")