feat(models): add approved provider policy store

Author: erichare

src/backend/base/langflow/agentic/mcp/server.py

@@ -64,6 +64,14 @@ async def _ensure_services() -> None:
         from langflow.services.utils import initialize_services
 
         await initialize_services()
+        from langflow.services.task.model_provider_policy_refresh import (
+            model_provider_policy_refresh_worker,
+        )
+
+        # Standalone stdio servers do not enter the FastAPI lifespan, but they
+        # are long-lived model consumers and must converge after an admin
+        # changes the install-wide provider ceiling.
+        await model_provider_policy_refresh_worker.start()
         _services_initialized = True
 
 

src/backend/base/langflow/alembic/versions/e8f1a2b3c4d5_add_model_provider_policy.py

@@ -0,0 +1,86 @@
+"""Add the install-wide model-provider policy store.
+
+Phase: EXPAND
+Revision ID: e8f1a2b3c4d5
+Revises: b7d5f9a3c2e4
+Create Date: 2026-07-29 00:00:00.000000
+
+The singleton starts with an empty approved-provider list. Empty deliberately
+means unrestricted, preserving existing behavior until an administrator saves
+a narrower policy. Its monotonically increasing version lets every worker
+notice committed changes without relying on process-local invalidation.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+from langflow.utils import migration
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+revision: str = "e8f1a2b3c4d5"  # pragma: allowlist secret
+down_revision: str | None = "b7d5f9a3c2e4"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+TABLE_NAME = "model_provider_policy"
+SINGLETON_ID = 1
+REQUIRED_COLUMNS = frozenset({"id", "approved_provider_ids", "version"})
+
+
+def _policy_table() -> sa.TableClause:
+    return sa.table(
+        TABLE_NAME,
+        sa.column("id", sa.Integer()),
+        sa.column("approved_provider_ids", sa.JSON()),
+        sa.column("version", sa.Integer()),
+    )
+
+
+def _seed_singleton(conn: sa.Connection) -> None:
+    table = _policy_table()
+    exists = conn.execute(sa.select(table.c.id).where(table.c.id == SINGLETON_ID)).first()
+    if exists is None:
+        conn.execute(
+            table.insert().values(
+                id=SINGLETON_ID,
+                approved_provider_ids=[],
+                version=0,
+            )
+        )
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if migration.table_exists(TABLE_NAME, conn):
+        missing_columns = {
+            column for column in REQUIRED_COLUMNS if not migration.column_exists(TABLE_NAME, column, conn)
+        }
+        if missing_columns:
+            missing = ", ".join(sorted(missing_columns))
+            msg = f"Existing {TABLE_NAME!r} table is missing required columns: {missing}"
+            raise RuntimeError(msg)
+        _seed_singleton(conn)
+        return
+
+    op.create_table(
+        TABLE_NAME,
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("approved_provider_ids", sa.JSON(), nullable=False),
+        sa.Column("version", sa.Integer(), nullable=False),
+        sa.CheckConstraint(f"id = {SINGLETON_ID}", name=op.f("ck_model_provider_policy_singleton")),
+        sa.CheckConstraint("version >= 0", name=op.f("ck_model_provider_policy_version")),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_model_provider_policy")),
+    )
+    _seed_singleton(conn)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if not migration.table_exists(TABLE_NAME, conn):
+        return
+    op.drop_table(TABLE_NAME)

src/backend/base/langflow/api/router.py

@@ -28,6 +28,7 @@
     mcp_router,
     memories_router,
     model_options_router,
+    model_provider_policy_router,
     models_router,
     monitor_router,
     openai_responses_router,
@@ -93,6 +94,7 @@ def include_deployment_router(target_router: APIRouter) -> None:
 router_v1.include_router(openai_responses_router)
 router_v1.include_router(models_router)
 router_v1.include_router(model_options_router)
+router_v1.include_router(model_provider_policy_router)
 router_v1.include_router(authz_shares_router)
 router_v1.include_router(authz_audit_router)
 router_v1.include_router(authz_roles_router)

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

@@ -20,6 +20,7 @@
 from langflow.api.v1.mcp_projects import router as mcp_projects_router
 from langflow.api.v1.memories import router as memories_router
 from langflow.api.v1.model_options import router as model_options_router
+from langflow.api.v1.model_provider_policy import router as model_provider_policy_router
 from langflow.api.v1.models import router as models_router
 from langflow.api.v1.monitor import router as monitor_router
 from langflow.api.v1.openai_responses import router as openai_responses_router
@@ -55,6 +56,7 @@
     "mcp_router",
     "memories_router",
     "model_options_router",
+    "model_provider_policy_router",
     "models_router",
     "monitor_router",
     "openai_responses_router",

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

@@ -0,0 +1,102 @@
+"""Superuser administration for the install-wide approved-provider policy."""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+from lfx.base.models.provider_registry import get_registry_snapshot
+from pydantic import BaseModel, Field, StringConstraints, field_validator
+
+from langflow.api.utils import DbSession, DbSessionReadOnly
+from langflow.services.auth.utils import get_current_active_superuser
+from langflow.services.database.models.user.model import User
+from langflow.services.model_provider_policy import (
+    apply_model_provider_policy_state,
+    get_model_provider_policy_state,
+    replace_model_provider_policy_state,
+)
+
+router = APIRouter(prefix="/model-provider-policy", tags=["Model Provider Policy"])
+
+ProviderId = Annotated[str, StringConstraints(pattern=r"^[a-z0-9][a-z0-9._-]*$", max_length=255)]
+
+
+class RegisteredModelProviderRead(BaseModel):
+    """Stable provider identity exposed to the administrative picker."""
+
+    provider_id: str
+    display_name: str
+    provider: str
+
+
+class ModelProviderPolicyWrite(BaseModel):
+    """Complete replacement for the install-wide approved-provider set."""
+
+    approved_provider_ids: Annotated[list[ProviderId], Field(max_length=1000)]
+
+    @field_validator("approved_provider_ids")
+    @classmethod
+    def deduplicate_provider_ids(cls, provider_ids: list[str]) -> list[str]:
+        return sorted(set(provider_ids))
+
+
+class ModelProviderPolicyRead(BaseModel):
+    """Current deployment ceiling plus every provider registered in this process."""
+
+    approved_provider_ids: list[str]
+    registered_providers: list[RegisteredModelProviderRead]
+
+
+def _build_policy_response(approved_provider_ids: set[str] | frozenset[str]) -> ModelProviderPolicyRead:
+    snapshot = get_registry_snapshot()
+    registered_providers = [
+        RegisteredModelProviderRead(
+            provider_id=provider_id,
+            display_name=descriptor.display_name or descriptor.name,
+            provider=descriptor.name,
+        )
+        for provider_id, descriptor in snapshot.descriptors_by_id.items()
+    ]
+    registered_providers.sort(key=lambda descriptor: (descriptor.display_name.casefold(), descriptor.provider_id))
+    return ModelProviderPolicyRead(
+        approved_provider_ids=sorted(approved_provider_ids),
+        registered_providers=registered_providers,
+    )
+
+
+@router.get("", response_model=ModelProviderPolicyRead)
+@router.get("/", response_model=ModelProviderPolicyRead)
+async def read_model_provider_policy(
+    _admin: Annotated[User, Depends(get_current_active_superuser)],
+    session: DbSessionReadOnly,
+) -> ModelProviderPolicyRead:
+    """Read the global provider policy. An empty approved list is unrestricted."""
+    state = await get_model_provider_policy_state(session)
+    return _build_policy_response(state.approved_provider_ids)
+
+
+@router.post("", response_model=ModelProviderPolicyRead)
+@router.post("/", response_model=ModelProviderPolicyRead)
+@router.put("", response_model=ModelProviderPolicyRead)
+@router.put("/", response_model=ModelProviderPolicyRead)
+async def replace_model_provider_policy(
+    payload: ModelProviderPolicyWrite,
+    _admin: Annotated[User, Depends(get_current_active_superuser)],
+    session: DbSession,
+) -> ModelProviderPolicyRead:
+    """Atomically replace the global provider policy and invalidate snapshots."""
+    state = await replace_model_provider_policy_state(session, payload.approved_provider_ids)
+
+    # Never publish uncommitted policy to the runtime. A failed commit leaves
+    # the previous in-memory ceiling and cached decisions intact.
+    apply_model_provider_policy_state(state)
+    return _build_policy_response(state.approved_provider_ids)
+
+
+__all__ = [
+    "ModelProviderPolicyRead",
+    "ModelProviderPolicyWrite",
+    "RegisteredModelProviderRead",
+    "router",
+]

src/backend/base/langflow/main.py

@@ -247,6 +247,16 @@ async def lifespan(_app: FastAPI):
             except Exception as exc:  # noqa: BLE001 — never block startup on cleanup scheduling
                 await logger.awarning(f"Failed to start authz audit-log cleanup worker: {exc}")
 
+            # Keep the default OSS provider ceiling coherent across backend
+            # worker processes after an administrator commits a replacement.
+            # This worker is part of policy enforcement, so a scheduling failure
+            # must fail startup rather than leave a worker stale indefinitely.
+            from langflow.services.task.model_provider_policy_refresh import (
+                model_provider_policy_refresh_worker,
+            )
+
+            await model_provider_policy_refresh_worker.start()
+
             current_time = asyncio.get_event_loop().time()
             await logger.adebug("Setting up LLM caching")
             setup_llm_caching()
@@ -621,6 +631,14 @@ async def refresh_models_dev_periodically() -> None:
                         await audit_log_cleanup_worker.stop()
                     except Exception as e:  # noqa: BLE001
                         await logger.aerror(f"Failed to stop authz audit-log cleanup worker: {e}")
+                    try:
+                        from langflow.services.task.model_provider_policy_refresh import (
+                            model_provider_policy_refresh_worker,
+                        )
+
+                        await model_provider_policy_refresh_worker.stop()
+                    except Exception as e:  # noqa: BLE001
+                        await logger.aerror(f"Failed to stop model-provider policy refresh worker: {e}")
 
                     # Cancel background tasks
                     tasks_to_cancel = []

src/backend/base/langflow/services/database/models/__init__.py

@@ -25,6 +25,7 @@
 from .mcp_server import MCPServer
 from .memory_base import MemoryBase, MemoryBaseSession, MemoryBaseWorkflowRun, MessageIngestionRecord
 from .message import MessageTable
+from .model_provider_policy import ModelProviderPolicy
 from .traces.model import SpanTable, TraceTable
 from .transactions import TransactionTable
 from .user import User
@@ -63,6 +64,7 @@
     "MemoryBaseWorkflowRun",
     "MessageIngestionRecord",
     "MessageTable",
+    "ModelProviderPolicy",
     "SSOConfig",
     "SSOUserProfile",
     "SignalType",

src/backend/base/langflow/services/database/models/model_provider_policy/__init__.py

@@ -0,0 +1,3 @@
+from .model import MODEL_PROVIDER_POLICY_SINGLETON_ID, ModelProviderPolicy
+
+__all__ = ["MODEL_PROVIDER_POLICY_SINGLETON_ID", "ModelProviderPolicy"]

src/backend/base/langflow/services/database/models/model_provider_policy/model.py

@@ -0,0 +1,21 @@
+import sqlalchemy as sa
+from sqlmodel import Field, SQLModel
+
+MODEL_PROVIDER_POLICY_SINGLETON_ID = 1
+
+
+class ModelProviderPolicy(SQLModel, table=True):  # type: ignore[call-arg]
+    """Versioned singleton containing the install-wide provider ceiling."""
+
+    __tablename__ = "model_provider_policy"
+    __table_args__ = (
+        sa.CheckConstraint(
+            f"id = {MODEL_PROVIDER_POLICY_SINGLETON_ID}",
+            name="ck_model_provider_policy_singleton",
+        ),
+        sa.CheckConstraint("version >= 0", name="ck_model_provider_policy_version"),
+    )
+
+    id: int = Field(default=MODEL_PROVIDER_POLICY_SINGLETON_ID, primary_key=True)
+    approved_provider_ids: list[str] = Field(default_factory=list, sa_column=sa.Column(sa.JSON, nullable=False))
+    version: int = Field(default=0, nullable=False, ge=0)