feat(mcp): persist MCP servers in a database table (#13976)

* feat(mcp): persist MCP servers in a database table

MCP servers were stored in a per-user JSON file guarded by an in-process lock, so concurrent edits lose updates and the list diverges across workers/replicas. Store them in a new mcp_server table (one row per server, unique on (user_id, name)); get_server_list/get_server/update_server become per-row reads/upserts and the in-process lock is removed, so MCP config is safe at any worker/replica count. Secret values (env/headers) are encrypted at rest. Existing _mcp_servers_*.json files are imported by an idempotent startup backfill and a `langflow migrate-mcp` CLI; the file is kept for rollback. REST shape and function signatures are unchanged.

Fixes #13970

* fix(mcp): clear hashed cache-key variants when a server changes

The shared tool cache is keyed {server_name}:{hash of headers+timeout} (MCPComponent._mcp_servers_cache_key), so clearing only the bare server name left servers with custom headers/timeouts serving stale config after an update. Clear the bare name and all {server_name}: variants.

* fix(mcp): re-apply create/merge rules after IntegrityError refetch

The IntegrityError fallback in update_server reused the pre-race config, so a concurrent same-name create (check_existing) could overwrite the winning row and a concurrent PATCH (merge_existing) skipped the merge. Now re-apply the rules against the refetched row: check_existing raises 'already exists', merge_existing merges, otherwise overwrite. Adds a concurrency test.

* fix(mcp): order get_server_list by created_at to preserve insertion order

get_server_list had no ORDER BY, so with several rows the server list order was undefined. The legacy file (a dict) preserved insertion order and the UI relies on it (the starter project must stay first). Order by created_at to restore stable insertion order.

* fix(mcp): re-point migration onto current base head to resolve alembic multi-head

release-1.11.0 gained migration c3e7a1b9d2f4 after this branch opened. Our migration also chained off 4f0d2c9a8b7e, producing two alembic heads so 'alembic upgrade head' failed and the DB never initialized (cascading to all e2e/integration/docker jobs). Re-point down_revision from 4f0d2c9a8b7e to c3e7a1b9d2f4 so the chain is linear with a single head.

---------

Co-authored-by: Saad ur Rehman <saad.urrehman@cleura.com>

Author: thesaadmirza

src/backend/base/langflow/__main__.py

@@ -975,6 +975,33 @@ async def _create_superuser(username: str, password: str, auth_token: str | None
             typer.echo("Superuser creation failed.")
 
 
+@app.command(name="migrate-mcp")
+def migrate_mcp(
+    log_level: str = typer.Option("info", help="Logging level.", envvar="LANGFLOW_LOG_LEVEL"),
+    dry_run: bool = typer.Option(default=False, help="Report what would be imported without writing."),  # noqa: FBT001
+) -> None:
+    """Import existing per-user MCP config files (_mcp_servers_<id>.json) into the mcp_server table.
+
+    This also runs automatically on startup; use this command to run or preview it on demand.
+    It is idempotent and never deletes the legacy files.
+    """
+    configure(log_level=log_level)
+    asyncio.run(_migrate_mcp(dry_run=dry_run))
+
+
+async def _migrate_mcp(*, dry_run: bool) -> None:
+    from langflow.api.utils.mcp.backfill import backfill_mcp_servers_from_files
+
+    await initialize_services()
+    async with session_scope() as session:
+        summary = await backfill_mcp_servers_from_files(session, dry_run=dry_run)
+    verb = "would import" if dry_run else "imported"
+    typer.echo(
+        f"MCP migration complete: {verb} {summary['imported']} server(s) across "
+        f"{summary['users']} user(s) (skipped {summary['skipped']}, errors {summary['errors']})."
+    )
+
+
 # command to copy the langflow database from the cache to the current directory
 # because now the database is stored per installation
 @app.command()

src/backend/base/langflow/alembic/versions/247308ce2598_add_mcp_server_table.py

@@ -0,0 +1,51 @@
+"""add mcp_server table
+
+Revision ID: 247308ce2598
+Revises: c3e7a1b9d2f4
+Create Date: 2026-07-06 17:10:00.000000
+
+Phase: EXPAND
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+import sqlmodel
+from alembic import op
+from langflow.utils import migration
+
+# revision identifiers, used by Alembic.
+revision: str = "247308ce2598"  # pragma: allowlist secret
+down_revision: str | None = "c3e7a1b9d2f4"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if migration.table_exists("mcp_server", conn):
+        return
+
+    op.create_table(
+        "mcp_server",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("user_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("transport", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("config", sa.JSON(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), nullable=False),
+        sa.Column("version", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("user_id", "name", name="uq_mcp_server_name_user"),
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if not migration.table_exists("mcp_server", conn):
+        return
+
+    op.drop_table("mcp_server")

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

@@ -0,0 +1,80 @@
+"""Backfill MCP servers from the legacy per-user JSON file into the mcp_server table.
+
+Runs best-effort on startup and via ``langflow migrate-mcp``. Properties:
+
+- **Idempotent** — only inserts a server that isn't already a row for that user, so
+  it is safe to run on every boot and to re-run by hand.
+- **Multi-replica safe** — the ``(user_id, name)`` unique constraint is the backstop;
+  a concurrent insert from another replica raises ``IntegrityError`` which is caught.
+- **Backend-agnostic** — reads through the storage service, so it covers local disk
+  and S3/Ceph/MinIO identically.
+- **Non-destructive** — the legacy file is never deleted, so rolling back to a
+  file-based Langflow remains safe.
+"""
+
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import select
+
+from langflow.api.v2.mcp import _read_legacy_mcp_file
+from langflow.logging import logger
+from langflow.services.auth.mcp_encryption import encrypt_mcp_config
+from langflow.services.database.models import MCPServer
+from langflow.services.database.models.user.model import User
+from langflow.services.deps import get_settings_service, get_storage_service
+
+
+async def backfill_mcp_servers_from_files(session, *, dry_run: bool = False) -> dict[str, int]:
+    """Import each user's legacy ``_mcp_servers_<id>.json`` entries into ``mcp_server``.
+
+    Args:
+        session: An async DB session.
+        dry_run: If True, count what would be imported without writing.
+
+    Returns:
+        Summary counts: ``{"users", "imported", "skipped", "errors"}``.
+    """
+    storage_service = get_storage_service()
+    settings_service = get_settings_service()
+    summary = {"users": 0, "imported": 0, "skipped": 0, "errors": 0}
+
+    users = (await session.exec(select(User))).all()
+    for user in users:
+        summary["users"] += 1
+        try:
+            legacy = await _read_legacy_mcp_file(
+                user, session, storage_service, settings_service, create_if_missing=False
+            )
+        except Exception as e:  # noqa: BLE001
+            summary["errors"] += 1
+            await logger.awarning(f"MCP backfill: could not read legacy MCP file for user {user.id}: {e}")
+            continue
+
+        pending = 0
+        for name, config in (legacy.get("mcpServers") or {}).items():
+            already = (
+                await session.exec(select(MCPServer).where(MCPServer.user_id == user.id, MCPServer.name == name))
+            ).first()
+            if already is not None:
+                summary["skipped"] += 1
+                continue
+            summary["imported"] += 1
+            if dry_run:
+                continue
+            session.add(MCPServer(user_id=user.id, name=name, config=encrypt_mcp_config(config or {})))
+            pending += 1
+
+        if pending:
+            try:
+                await session.commit()
+            except IntegrityError:
+                # Another replica imported this user's servers first; safe to skip.
+                await session.rollback()
+
+    await logger.ainfo(
+        "MCP backfill complete: users=%s imported=%s skipped=%s errors=%s",
+        summary["users"],
+        summary["imported"],
+        summary["skipped"],
+        summary["errors"],
+    )
+    return summary