Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions BUNDLE_API.md
Original file line number Diff line number Diff line change
Expand Up @@ -505,6 +505,16 @@ the deserialize half is covered by
exposes an immutable generation-tagged snapshot plus an eager catalog
validation hook for deployment readiness. Existing provider manifests are
unaffected and `BUNDLE_API_VERSION` remains `1`.
- **Provider identity resolution and policy snapshots (additive).**
`resolve_provider_id()` is the canonical resolver for registered names,
display names, aliases, stable IDs, and deterministic legacy fallbacks.
Model-provider policy services now expose cached synchronous `resolve()`,
async `aresolve()`, single-provider `is_allowed()`, and `invalidate()`
hooks while preserving the existing immutable snapshot and allow-all OSS
behavior. The process-local snapshot cache has a bounded TTL, and async
implementations can override `aget_allowed_provider_ids()` without
bypassing it. Existing synchronous policy subclasses, including subclasses
that did not call `super().__init__()`, remain source-compatible.
- **New typed error codes (additive): `provider-invalid`, `provider-skipped`.**
A malformed provider spec surfaces `provider-invalid`; a provider whose name
collides with a built-in or already-loaded provider surfaces
Expand Down
8 changes: 8 additions & 0 deletions src/backend/base/langflow/agentic/mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
Original file line number Diff line number Diff line change
@@ -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)
2 changes: 2 additions & 0 deletions src/backend/base/langflow/api/router.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
mcp_router,
memories_router,
model_options_router,
model_provider_policy_router,
models_router,
monitor_router,
openai_responses_router,
Expand Down Expand Up @@ -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)
Expand Down
2 changes: 2 additions & 0 deletions src/backend/base/langflow/api/v1/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -55,6 +56,7 @@
"mcp_router",
"memories_router",
"model_options_router",
"model_provider_policy_router",
"models_router",
"monitor_router",
"openai_responses_router",
Expand Down
102 changes: 102 additions & 0 deletions src/backend/base/langflow/api/v1/model_provider_policy.py
Original file line number Diff line number Diff line change
@@ -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",
]
18 changes: 18 additions & 0 deletions src/backend/base/langflow/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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 = []
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -63,6 +64,7 @@
"MemoryBaseWorkflowRun",
"MessageIngestionRecord",
"MessageTable",
"ModelProviderPolicy",
"SSOConfig",
"SSOUserProfile",
"SignalType",
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
from .model import MODEL_PROVIDER_POLICY_SINGLETON_ID, ModelProviderPolicy

__all__ = ["MODEL_PROVIDER_POLICY_SINGLETON_ID", "ModelProviderPolicy"]
Original file line number Diff line number Diff line change
@@ -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)
Loading
Loading