chore(prototypes): multitenant lfx capability boundary POC (RFC)

Working POC of the multitenant capability boundary applied to lfx's
existing service injection. Not for merge; lives under prototypes/.

- CapabilityVariableService implements VariableServiceProtocol and is
  registered via @register_service(ServiceType.VARIABLE_SERVICE),
  replacing lfx's default at runtime.
- The real Basic Prompting starter project runs through lfx in a
  worker subprocess whose env has been scrubbed of DB and provider
  credentials.
- A short-lived run token gates variable resolution. Granted scope:
  the flow completes against real OpenAI. Empty scope: the shim
  returns PermissionError and lfx surfaces a ComponentBuildError.

Also establishes prototypes/ as a lint-free sandbox in the root ruff
config and pre-commit hooks (ruff, detect-secrets), matching the
existing pattern for docs/ python examples.

.pre-commit-config.yaml

@@ -19,14 +19,14 @@ repos:
         language: system
         types_or: [python, pyi]
         args: [--fix]
-        exclude: ^docs/.*/API-Reference/python-examples/
+        exclude: (^docs/.*/API-Reference/python-examples/|^prototypes/)
       - id: ruff-format
         name: ruff format
         entry: uv run ruff format
         language: system
         types_or: [python, pyi]
         args: [--config, pyproject.toml]
-        exclude: ^docs/.*/API-Reference/python-examples/
+        exclude: (^docs/.*/API-Reference/python-examples/|^prototypes/)
       - id: validate-migrations
         name: Validate Alembic Migrations (Expand-Contract)
         entry: python src/backend/base/langflow/alembic/migration_validator.py
@@ -47,7 +47,7 @@ repos:
     hooks:
       - id: detect-secrets
         args: ["--baseline", ".secrets.baseline"]
-        exclude: '(^docs/|^SECURITY\.md$|src/lfx/src/lfx/_assets/component_index\.json$|src/lfx/README\.md$)'
+        exclude: '(^docs/|^SECURITY\.md$|src/lfx/src/lfx/_assets/component_index\.json$|src/lfx/README\.md$|^prototypes/)'
   - repo: local
     hooks:
       - id: local-biome-check

prototypes/mt-custom-components/README.md

@@ -0,0 +1,115 @@
+# Multitenant lfx Prototype
+
+A working POC of the multitenant capability boundary applied to the real
+lfx codebase. Loads the actual `Basic Prompting` starter project, runs it
+through real lfx, and proves the worker process can complete the flow with
+**no DB credentials and no provider keys in its env** — only a short-lived
+run token.
+
+Design doc: `~/Projects/ideas/2026-05-14-langflow-multitenant-custom-components.md`.
+
+## What it shows
+
+- lfx's existing `VariableServiceProtocol` is the right injection point. A
+  capability-backed `CapabilityVariableService` is registered via
+  `@register_service(ServiceType.VARIABLE_SERVICE)` and quietly replaces
+  lfx's default.
+- The actual `src/backend/base/langflow/initial_setup/starter_projects/
+  Basic Prompting.json` loads and runs through `lfx.load.aload_flow_from_json`
+  → `graph.async_start`.
+- The lfx process runs as a **subprocess worker** with `LANGFLOW_DATABASE_URL`,
+  `PG*`, `OPENAI_API_KEY` and friends scrubbed from its env. It refuses to
+  boot if any of them are present.
+- The LanguageModel component's api_key (`load_from_db: true` →
+  `value: "OPENAI_API_KEY"`) is fetched through the capability shim →
+  Runtime API → returns the seeded value. Verified at the end by checking
+  the API's event log for the `variable_read`.
+- With a scope-less token, the capability shim returns `PermissionError`,
+  lfx surfaces it as `ComponentBuildError: OpenAI API key is required`,
+  the flow fails closed.
+
+## Run
+
+```bash
+cd prototypes/mt-custom-components
+
+# Happy path: scoped token, flow completes against real OpenAI
+bash scripts/run.sh
+
+# Boundary refusal: empty-scope token, flow fails closed
+bash scripts/run.sh --deny
+```
+
+The orchestrator inherits your shell's `OPENAI_API_KEY` and seeds it into
+the Runtime API. The worker subprocess never sees the key in its env. If
+you don't have a real key set, the orchestrator seeds the sentinel
+`"MOCK-not-a-real-key"` and the lfx run still proceeds (it just fails at
+the actual OpenAI HTTP call — the boundary verification still passes
+because the variable read was logged).
+
+## Layout
+
+```
+runtime_api/                       Control-plane stand-in
+  main.py                          FastAPI: variables / memory / artifacts / events
+  auth.py                          JWT mint/verify and scope check (HS256, prototype)
+  store.py                         In-memory tenant-keyed store
+
+capability_variable_service.py     The shim: implements VariableServiceProtocol,
+                                   registered via @register_service. Every
+                                   get_variable() call goes to the Runtime API
+                                   carrying the worker's run token.
+
+flows/basic_prompting.json         Real Langflow starter project, with the
+                                   LanguageModel api_key field set to
+                                   value="OPENAI_API_KEY", load_from_db=True so
+                                   the variable service is actually consulted.
+
+scripts/
+  orchestrator.py                  Control plane: starts the Runtime API,
+                                   seeds the variable, mints the run token,
+                                   builds a scrubbed env, spawns the worker.
+                                   Never imports lfx.
+  worker.py                        Worker subprocess: refuses to boot under
+                                   forbidden env, registers the capability
+                                   shim, loads + runs the flow, emits a
+                                   single JSON outcome line.
+  run.sh                           Convenience wrapper.
+```
+
+## Where it differs from production
+
+This is a POC. The shape lines up with the design doc; the production
+hardening does not:
+
+- **Token signing.** HS256 with a shared secret. Real version wants
+  asymmetric signing (control plane signs, workers only verify) and
+  mutual TLS on the Runtime API.
+- **Runtime API store.** In-memory dict keyed by `(tenant_id, name)`. Real
+  version is whatever the control plane's source-of-truth is.
+- **Subprocess vs container.** The "worker tier" here is a Python
+  subprocess with a scrubbed env. Real multitenant workers belong in
+  containers / Firecracker / a managed microVM; the shape of the
+  capability boundary doesn't change.
+- **One worker per flow.** Per-vertex isolation (a separate worker for
+  each component) is the design's stronger story — useful when first-party
+  components are trusted but user custom components in the same flow are
+  not. Out of scope here.
+- **Capability surface.** Only `variables:read:*` is exercised because
+  Basic Prompting only needs the api_key. Memory / files / artifacts /
+  events endpoints exist on the Runtime API and have the same shape.
+
+## One real finding worth flagging
+
+`src/lfx/src/lfx/base/models/unified_models/credentials.py:32` checks
+`os.environ.get(var_name)` **before** consulting the variable service. In
+this prototype that means the worker env *must* be scrubbed of provider
+keys — otherwise the env value silently bypasses the capability boundary.
+Real integration should either invert that order in multitenant mode, or
+make the variable service the sole resolver. The orchestrator here scrubs
+defensively for the same reason.
+
+Separately, `credentials.py:100` catches `(ValueError, Exception)` broadly
+when the variable service raises. Our shim's `PermissionError` (scope
+denied) gets flattened to "API key is required" — same outcome, lossier
+message. Worth letting typed denials bubble.

prototypes/mt-custom-components/capability_variable_service.py

@@ -0,0 +1,94 @@
+"""Capability-backed VariableService for lfx.
+
+Replaces lfx's default in-memory / env-fallback variable service. Every
+`get_variable(name, ...)` call goes to the prototype's Runtime API with a
+run token. The token's scopes determine what is reachable. The worker (or
+the lfx process registering this service) never holds the DB.
+
+Registration is via lfx's @register_service decorator, applied when this
+module is imported.
+"""
+
+from __future__ import annotations
+
+import os
+
+import httpx
+from lfx.log.logger import logger
+from lfx.services.base import Service
+from lfx.services.registry import register_service
+from lfx.services.schema import ServiceType
+
+
+class _RuntimeAPIConfig:
+    """Read RUNTIME_API_URL + RUN_TOKEN once at instantiation time.
+
+    The token is supplied by the orchestrator that boots lfx. In Phase A,
+    that orchestrator is `scripts/run_basic_prompting_lfx.py`. In a real
+    Stepflow-shaped deployment, the worker process would receive these via
+    env from the control plane.
+    """
+
+    def __init__(self) -> None:
+        self.base_url = os.environ["MT_RUNTIME_API_URL"].rstrip("/")
+        self.token = os.environ["MT_RUN_TOKEN"]
+
+
+@register_service(ServiceType.VARIABLE_SERVICE)
+class CapabilityVariableService(Service):
+    name = "variable_service"
+
+    def __init__(self) -> None:
+        super().__init__()
+        try:
+            self._config: _RuntimeAPIConfig | None = _RuntimeAPIConfig()
+        except KeyError:
+            # Tests / non-prototype usage: fall back to None and let calls error
+            # if anyone actually hits the service without configuration.
+            self._config = None
+        self._client = httpx.AsyncClient(timeout=10.0)
+        self.set_ready()
+        logger.debug("CapabilityVariableService registered")
+
+    async def get_variable(self, name: str, **kwargs) -> str | None:  # noqa: ARG002
+        """Fetch the variable through the Runtime API.
+
+        Signature absorbs `user_id`, `field`, `session` kwargs from the
+        langflow call path. We don't use them here — capability scoping
+        happens through the token, not through the kwargs.
+        """
+        if self._config is None:
+            msg = (
+                "CapabilityVariableService called without MT_RUNTIME_API_URL "
+                "and MT_RUN_TOKEN set. The prototype orchestrator is "
+                "responsible for setting these before lfx starts."
+            )
+            raise RuntimeError(msg)
+        url = f"{self._config.base_url}/runtime/variables/{name}"
+        r = await self._client.get(url, headers={"Authorization": f"Bearer {self._config.token}"})
+        if r.status_code == 200:
+            logger.debug(f"variable '{name}' resolved through capability service")
+            return r.json()["value"]
+        if r.status_code in (401, 403):
+            # Surface as a clear, actionable error. Langflow would otherwise
+            # mask this as "variable not found".
+            msg = f"capability denied for variables:read:{name}: {r.text}"
+            raise PermissionError(msg)
+        if r.status_code == 404:
+            return None
+        r.raise_for_status()
+        return None
+
+    def set_variable(self, name: str, value: str, **kwargs) -> None:
+        # The capability boundary doesn't expose variable writes to workers
+        # in this prototype. Authoring writes happen in the control plane.
+        msg = "set_variable is not supported in the capability-backed tier"
+        raise NotImplementedError(msg)
+
+    async def get_all_decrypted_variables(self, user_id, session) -> dict[str, str]:  # noqa: ARG002
+        # Not implemented: enumerating all of a tenant's variables in the
+        # worker tier is exactly the kind of broad read the design refuses.
+        return {}
+
+    async def teardown(self) -> None:
+        await self._client.aclose()

prototypes/mt-custom-components/pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "mt-lfx-prototype"
+version = "0.0.1"
+description = "Multitenant lfx prototype: capability-backed services + isolated worker subprocess"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi>=0.115",
+    "uvicorn[standard]>=0.30",
+    "pyjwt>=2.9",
+    "httpx>=0.27",
+    "pydantic>=2.8",
+    "lfx",
+    # Used by the Basic Prompting LanguageModel component.
+    "langchain-openai>=0.2",
+]
+
+[tool.uv.sources]
+lfx = { path = "../../src/lfx", editable = true }
+
+[tool.uv]
+package = false

prototypes/mt-custom-components/runtime_api/auth.py

@@ -0,0 +1,72 @@
+"""Run-token mint/verify and scope matching.
+
+Prototype simplification: HS256 with a shared secret. Production design calls
+for asymmetric signing plus mutual TLS between worker and Runtime API.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from dataclasses import dataclass
+
+import jwt
+
+ALGO = "HS256"
+SECRET = os.environ.get("RUNTIME_API_SECRET", "dev-secret-change-me")
+
+
+@dataclass(frozen=True)
+class RunClaims:
+    tenant_id: str
+    user_id: str
+    flow_id: str
+    run_id: str
+    component_id: str
+    scopes: tuple[str, ...]
+    exp: int
+
+
+def mint(
+    *,
+    tenant_id: str,
+    user_id: str,
+    flow_id: str,
+    run_id: str,
+    component_id: str,
+    scopes: list[str],
+    ttl_seconds: int = 300,
+) -> str:
+    payload = {
+        "tenant_id": tenant_id,
+        "user_id": user_id,
+        "flow_id": flow_id,
+        "run_id": run_id,
+        "component_id": component_id,
+        "scopes": scopes,
+        "exp": int(time.time()) + ttl_seconds,
+    }
+    return jwt.encode(payload, SECRET, algorithm=ALGO)
+
+
+def verify(token: str) -> RunClaims:
+    payload = jwt.decode(token, SECRET, algorithms=[ALGO])
+    return RunClaims(
+        tenant_id=payload["tenant_id"],
+        user_id=payload["user_id"],
+        flow_id=payload["flow_id"],
+        run_id=payload["run_id"],
+        component_id=payload["component_id"],
+        scopes=tuple(payload["scopes"]),
+        exp=payload["exp"],
+    )
+
+
+def has_scope(claims: RunClaims, required: str) -> bool:
+    """Exact-match scope check.
+
+    Scopes look like 'variables:read:openai_api_key' or
+    'memory:write:session:abc'. Prototype keeps matching literal; a real
+    implementation would support hierarchical or category scopes.
+    """
+    return required in claims.scopes

prototypes/mt-custom-components/runtime_api/main.py

@@ -0,0 +1,171 @@
+"""Runtime API.
+
+What this proves:
+- The worker never receives DB credentials. It receives a short-lived run
+  token. All DB-backed behavior goes through these endpoints.
+- Scope claims on the token determine what is reachable. No claim, no access.
+- Tenant isolation is enforced by the control plane, not by user code.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import Depends, FastAPI, Header, HTTPException
+from pydantic import BaseModel
+
+from . import auth
+from .store import store
+
+app = FastAPI(title="Langflow Runtime API (prototype)")
+
+
+def get_claims(authorization: str = Header(...)) -> auth.RunClaims:
+    if not authorization.lower().startswith("bearer "):
+        raise HTTPException(401, "missing bearer token")
+    token = authorization.split(" ", 1)[1].strip()
+    try:
+        return auth.verify(token)
+    except Exception as exc:
+        raise HTTPException(401, f"invalid token: {exc}") from exc
+
+
+def require_scope(claims: auth.RunClaims, scope: str) -> None:
+    if not auth.has_scope(claims, scope):
+        raise HTTPException(403, f"missing scope: {scope}")
+
+
+# Admin helpers. Real control plane mints tokens internally; we expose this for
+# the demo so a shell script can drive end-to-end flows.
+
+
+class MintRequest(BaseModel):
+    tenant_id: str
+    user_id: str
+    flow_id: str
+    run_id: str
+    component_id: str
+    scopes: list[str]
+    ttl_seconds: int = 300
+
+
+class MintResponse(BaseModel):
+    token: str
+
+
+@app.post("/admin/mint-token", response_model=MintResponse)
+def mint_token(req: MintRequest) -> MintResponse:
+    token = auth.mint(
+        tenant_id=req.tenant_id,
+        user_id=req.user_id,
+        flow_id=req.flow_id,
+        run_id=req.run_id,
+        component_id=req.component_id,
+        scopes=req.scopes,
+        ttl_seconds=req.ttl_seconds,
+    )
+    return MintResponse(token=token)
+
+
+class SeedVariableRequest(BaseModel):
+    tenant_id: str
+    name: str
+    value: str
+
+
+@app.post("/admin/seed-variable")
+def seed_variable(req: SeedVariableRequest) -> dict[str, str]:
+    store.seed_variable(req.tenant_id, req.name, req.value)
+    return {"status": "ok"}
+
+
+@app.get("/admin/artifacts/{tenant_id}/{run_id}")
+def list_artifacts(tenant_id: str, run_id: str) -> list[dict[str, Any]]:
+    return store.read_artifacts(tenant_id, run_id)
+
+
+@app.get("/admin/events")
+def list_events() -> list[dict[str, Any]]:
+    return store.read_events()
+
+
+# Runtime capability endpoints. These are the only DB-backed surface the
+# worker can reach.
+
+
+@app.get("/runtime/variables/{name}")
+def get_variable(name: str, claims: auth.RunClaims = Depends(get_claims)) -> dict[str, str]:
+    require_scope(claims, f"variables:read:{name}")
+    value = store.get_variable(claims.tenant_id, name)
+    if value is None:
+        raise HTTPException(404, f"variable not found: {name}")
+    store.write_event(
+        {
+            "kind": "variable_read",
+            "tenant_id": claims.tenant_id,
+            "user_id": claims.user_id,
+            "run_id": claims.run_id,
+            "component_id": claims.component_id,
+            "name": name,
+        }
+    )
+    return {"name": name, "value": value}
+
+
+@app.get("/runtime/memory")
+def get_memory(session_id: str, claims: auth.RunClaims = Depends(get_claims)) -> list[dict[str, Any]]:
+    require_scope(claims, f"memory:read:session:{session_id}")
+    return store.read_memory(claims.tenant_id, session_id)
+
+
+class MemoryWrite(BaseModel):
+    session_id: str
+    role: str
+    content: str
+
+
+@app.post("/runtime/memory")
+def post_memory(body: MemoryWrite, claims: auth.RunClaims = Depends(get_claims)) -> dict[str, str]:
+    require_scope(claims, f"memory:write:session:{body.session_id}")
+    store.write_memory(
+        claims.tenant_id,
+        body.session_id,
+        {"role": body.role, "content": body.content, "component_id": claims.component_id},
+    )
+    return {"status": "ok"}
+
+
+class ArtifactWrite(BaseModel):
+    kind: str
+    data: dict[str, Any]
+
+
+@app.post("/runtime/artifacts")
+def post_artifact(body: ArtifactWrite, claims: auth.RunClaims = Depends(get_claims)) -> dict[str, str]:
+    require_scope(claims, f"artifacts:write:run:{claims.run_id}")
+    store.write_artifact(
+        claims.tenant_id,
+        claims.run_id,
+        {"kind": body.kind, "data": body.data, "component_id": claims.component_id},
+    )
+    return {"status": "ok"}
+
+
+class EventWrite(BaseModel):
+    kind: str
+    data: dict[str, Any]
+
+
+@app.post("/runtime/events")
+def post_event(body: EventWrite, claims: auth.RunClaims = Depends(get_claims)) -> dict[str, str]:
+    require_scope(claims, f"events:write:run:{claims.run_id}")
+    store.write_event(
+        {
+            "kind": body.kind,
+            "tenant_id": claims.tenant_id,
+            "run_id": claims.run_id,
+            "component_id": claims.component_id,
+            "data": body.data,
+        }
+    )
+    return {"status": "ok"}

prototypes/mt-custom-components/runtime_api/store.py

@@ -0,0 +1,51 @@
+"""In-memory tenant-scoped store. Replaces what Langflow services hit in the DB."""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from threading import Lock
+from typing import Any
+
+
+class Store:
+    def __init__(self) -> None:
+        self._lock = Lock()
+        self._variables: dict[tuple[str, str], str] = {}
+        self._memory: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
+        self._artifacts: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
+        self._events: list[dict[str, Any]] = []
+
+    def seed_variable(self, tenant_id: str, name: str, value: str) -> None:
+        with self._lock:
+            self._variables[(tenant_id, name)] = value
+
+    def get_variable(self, tenant_id: str, name: str) -> str | None:
+        with self._lock:
+            return self._variables.get((tenant_id, name))
+
+    def read_memory(self, tenant_id: str, session_id: str) -> list[dict[str, Any]]:
+        with self._lock:
+            return list(self._memory[(tenant_id, session_id)])
+
+    def write_memory(self, tenant_id: str, session_id: str, entry: dict[str, Any]) -> None:
+        with self._lock:
+            self._memory[(tenant_id, session_id)].append(entry)
+
+    def write_artifact(self, tenant_id: str, run_id: str, artifact: dict[str, Any]) -> None:
+        with self._lock:
+            self._artifacts[(tenant_id, run_id)].append(artifact)
+
+    def read_artifacts(self, tenant_id: str, run_id: str) -> list[dict[str, Any]]:
+        with self._lock:
+            return list(self._artifacts[(tenant_id, run_id)])
+
+    def write_event(self, event: dict[str, Any]) -> None:
+        with self._lock:
+            self._events.append(event)
+
+    def read_events(self) -> list[dict[str, Any]]:
+        with self._lock:
+            return list(self._events)
+
+
+store = Store()

prototypes/mt-custom-components/scripts/orchestrator.py

@@ -0,0 +1,235 @@
+"""Phase B orchestrator.
+
+Control-plane responsibilities:
+- start the Runtime API
+- seed the variables the flow will need
+- mint a per-run token with the right scopes
+- build a scrubbed env for the worker subprocess (no DB creds, no provider keys)
+- spawn the worker subprocess
+- capture its outcome and verify the boundary held
+
+The lfx process runs as the worker subprocess. The orchestrator never imports lfx.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import httpx
+
+PROTOTYPE_ROOT = Path(__file__).resolve().parents[1]
+FLOW_PATH = PROTOTYPE_ROOT / "flows" / "basic_prompting.json"
+WORKER_SCRIPT = PROTOTYPE_ROOT / "scripts" / "worker.py"
+
+API_HOST = "127.0.0.1"
+API_PORT = 8767
+API_URL = f"http://{API_HOST}:{API_PORT}"
+SECRET = "phase-b-prototype-secret-must-be-at-least-32-bytes"
+
+
+def _wait_for_port(host: str, port: int, *, timeout: float = 10.0) -> None:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        try:
+            with socket.create_connection((host, port), timeout=0.3):
+                return
+        except OSError:
+            time.sleep(0.1)
+    msg = f"port {host}:{port} did not come up within {timeout}s"
+    raise RuntimeError(msg)
+
+
+def _start_runtime_api() -> subprocess.Popen:
+    proc = subprocess.Popen(
+        [
+            sys.executable,
+            "-m",
+            "uvicorn",
+            "runtime_api.main:app",
+            "--host",
+            API_HOST,
+            "--port",
+            str(API_PORT),
+            "--log-level",
+            "warning",
+        ],
+        cwd=str(PROTOTYPE_ROOT),
+        env={**os.environ, "RUNTIME_API_SECRET": SECRET},
+    )
+    _wait_for_port(API_HOST, API_PORT)
+    return proc
+
+
+def _seed_and_mint(client: httpx.Client, *, scopes: list[str]) -> tuple[str, str]:
+    """Seed OPENAI_API_KEY and mint a run token with the given scopes.
+
+    The caller controls scopes so we can also exercise the negative path
+    (empty scopes -> worker should be denied at variable lookup time).
+    """
+    openai_key = os.environ.get("OPENAI_API_KEY", "MOCK-not-a-real-key")
+    client.post(
+        "/admin/seed-variable",
+        json={"tenant_id": "t1", "name": "OPENAI_API_KEY", "value": openai_key},
+    ).raise_for_status()
+
+    r = client.post(
+        "/admin/mint-token",
+        json={
+            "tenant_id": "t1",
+            "user_id": "u1",
+            "flow_id": "basic_prompting",
+            "run_id": "run-1",
+            "component_id": "lfx-worker",
+            "scopes": scopes,
+            "ttl_seconds": 300,
+        },
+    )
+    r.raise_for_status()
+    provider = "openai" if not openai_key.startswith("MOCK") else "mock"
+    return r.json()["token"], provider
+
+
+def _build_worker_env(token: str) -> dict[str, str]:
+    """Scrub everything the worker shouldn't see. Keep PATH + Python pieces."""
+    keep_prefixes = ("PATH", "PYTHONPATH", "PYTHON", "HOME", "USER", "LANG", "LC_", "TZ")
+    keep_exact = {"VIRTUAL_ENV"}
+    env: dict[str, str] = {}
+    for k, v in os.environ.items():
+        if k in keep_exact or any(k.startswith(p) for p in keep_prefixes):
+            env[k] = v
+    # Explicitly drop anything DB- or provider-related even if it matched
+    # a prefix by accident (e.g. PG* env vars).
+    for forbidden in (
+        "LANGFLOW_DATABASE_URL",
+        "DATABASE_URL",
+        "POSTGRES_URL",
+        "PGHOST",
+        "PGUSER",
+        "PGPASSWORD",
+        "PGDATABASE",
+        "SQLALCHEMY_DATABASE_URI",
+        "OPENAI_API_KEY",
+        "ANTHROPIC_API_KEY",
+        "GOOGLE_API_KEY",
+    ):
+        env.pop(forbidden, None)
+    env["MT_RUN_TOKEN"] = token
+    env["MT_RUNTIME_API_URL"] = API_URL
+    env["FLOW_PATH"] = str(FLOW_PATH)
+    env["INPUT_TEXT"] = "Hi, who are you?"
+    return env
+
+
+def _run_worker(env: dict[str, str]) -> dict:
+    result = subprocess.run(
+        [sys.executable, str(WORKER_SCRIPT)],
+        check=False,
+        capture_output=True,
+        text=True,
+        env=env,
+        cwd=str(PROTOTYPE_ROOT),
+    )
+    last_line = result.stdout.strip().split("\n")[-1] if result.stdout.strip() else ""
+    try:
+        outcome = json.loads(last_line)
+    except json.JSONDecodeError:
+        return {
+            "status": "unparseable",
+            "stdout": result.stdout[-2000:],
+            "stderr": result.stderr[-2000:],
+        }
+    return outcome
+
+
+def main() -> int:
+    deny_mode = "--deny" in sys.argv
+    scopes: list[str] = [] if deny_mode else ["variables:read:OPENAI_API_KEY"]
+    if deny_mode:
+        print("==> --deny mode: minting token with NO scopes; expecting boundary refusal")
+
+    print(f"==> Starting Runtime API on {API_URL}")
+    api_proc = _start_runtime_api()
+    try:
+        with httpx.Client(base_url=API_URL, timeout=5.0) as admin:
+            token, provider = _seed_and_mint(admin, scopes=scopes)
+            print(f"==> Seeded OPENAI_API_KEY (provider={provider}) and minted run token (scopes={scopes})")
+
+            env = _build_worker_env(token)
+            # Defensive sanity check: confirm the worker env is scrubbed.
+            leaked = [
+                k
+                for k in env
+                if k
+                in (
+                    "LANGFLOW_DATABASE_URL",
+                    "DATABASE_URL",
+                    "OPENAI_API_KEY",
+                    "PGHOST",
+                    "PGUSER",
+                    "PGPASSWORD",
+                )
+            ]
+            if leaked:
+                print(f"FAIL  scrub failed; worker would inherit: {leaked}")
+                return 1
+            print("OK    worker env scrubbed of DB/provider credentials")
+
+            print(f"==> Spawning worker subprocess: {WORKER_SCRIPT.name}")
+            outcome = _run_worker(env)
+            print(f"==> Worker outcome: status={outcome.get('status')}")
+
+            if deny_mode:
+                # The boundary should hold: worker should report denied (or
+                # propagate the PermissionError as a build error). Either way,
+                # not 'ok'.
+                ok_after_denial = outcome.get("status") == "ok"
+                error_blob = json.dumps(outcome)[:500]
+                if ok_after_denial:
+                    print(f"FAIL  scope-less token still produced ok: {error_blob}")
+                    return 1
+                print(f"OK    boundary refused: {error_blob}")
+                return 0
+
+            if outcome.get("status") == "ok":
+                print()
+                print("Final output:")
+                print(f"  {outcome.get('text') or '(no text returned)'}")
+            elif outcome.get("status") == "denied":
+                print(f"  denied: {outcome.get('error')}")
+                return 1
+            elif outcome.get("status") == "boot_failed":
+                print(f"  boot_failed: {outcome}")
+                return 1
+            else:
+                print(json.dumps(outcome, indent=2)[:1500])
+                return 1
+
+            # Verify the variable read went through the Runtime API.
+            events = admin.get("/admin/events").json()
+            reads = [
+                e
+                for e in events
+                if e.get("kind") == "variable_read" and e.get("name") == "OPENAI_API_KEY" and e.get("tenant_id") == "t1"
+            ]
+            print()
+            if reads:
+                print(f"OK    variable lookup went through Runtime API ({len(reads)} read(s) logged)")
+                return 0
+            print("FAIL  no Runtime API event logged; the boundary was bypassed")
+            return 1
+    finally:
+        api_proc.terminate()
+        try:
+            api_proc.wait(timeout=3)
+        except subprocess.TimeoutExpired:
+            api_proc.kill()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

prototypes/mt-custom-components/scripts/run.sh

@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+# Run the multitenant lfx prototype end-to-end.
+#
+# Orchestrator boots the Runtime API, seeds OPENAI_API_KEY, mints a
+# scope-bound run token, then spawns an lfx worker subprocess whose env
+# has been scrubbed of every DB and provider credential. The worker runs
+# the real Basic Prompting flow JSON through real lfx, resolving the
+# api_key through the capability-backed VariableService.
+#
+# Usage:
+#   bash scripts/run.sh           # happy path
+#   bash scripts/run.sh --deny    # mint with no scopes; expect refusal
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+echo "==> Syncing python deps"
+uv sync --quiet
+export PATH="$(pwd)/.venv/bin:${PATH}"
+
+uv run python scripts/orchestrator.py "$@"

prototypes/mt-custom-components/scripts/worker.py

@@ -0,0 +1,143 @@
+"""Phase B worker subprocess.
+
+What this proves:
+- A real Langflow flow JSON runs through real lfx code in an isolated
+  subprocess that holds NO database credentials and NO provider keys in
+  its environment.
+- The subprocess receives only a short-lived run token and the Runtime
+  API URL. Every variable lookup goes through the capability shim.
+- If LANGFLOW_DATABASE_URL or similar leaks into the env, the worker
+  refuses to start, identically to the clean-room worker's boot check.
+
+Inputs (env):
+    RUN_TOKEN, RUNTIME_API_URL, FLOW_PATH, INPUT_TEXT
+    MT_RUN_TOKEN, MT_RUNTIME_API_URL (set by the orchestrator before spawn)
+
+Output (stdout): a single trailing JSON line with the outcome.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+import traceback
+from pathlib import Path
+
+PROTOTYPE_ROOT = Path(__file__).resolve().parents[1]
+if str(PROTOTYPE_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROTOTYPE_ROOT))
+
+# The lfx process IS the worker tier in this prototype. Treat its env as a
+# sandbox surface: refuse to boot if anything that smells like DB or
+# provider credentials leaked in.
+FORBIDDEN_ENV = (
+    "LANGFLOW_DATABASE_URL",
+    "DATABASE_URL",
+    "POSTGRES_URL",
+    "PGHOST",
+    "PGUSER",
+    "PGPASSWORD",
+    "PGDATABASE",
+    "SQLALCHEMY_DATABASE_URI",
+    # Provider keys must not be in the worker env: the design's whole
+    # point is fetching them through capabilities.
+    "OPENAI_API_KEY",
+    "ANTHROPIC_API_KEY",
+    "GOOGLE_API_KEY",
+)
+
+
+def _emit(outcome: dict) -> None:
+    sys.stdout.write(json.dumps(outcome) + "\n")
+    sys.stdout.flush()
+
+
+async def _run() -> dict:
+    # Register the capability-backed variable service. The decorator on
+    # import time replaces lfx's default. Must happen BEFORE the graph
+    # builds any vertex.
+    import capability_variable_service  # noqa: F401
+    from lfx.load import aload_flow_from_json
+    from lfx.run._defaults import apply_run_defaults
+    from lfx.schema.schema import InputValueRequest
+
+    flow_path = os.environ["FLOW_PATH"]
+    input_text = os.environ.get("INPUT_TEXT", "Hi, who are you?")
+
+    graph = await aload_flow_from_json(flow_path, disable_logs=True)
+    apply_run_defaults(graph, session_id=None, user_id=None)
+    graph.prepare()
+
+    results = []
+    async for r in graph.async_start(InputValueRequest(input_value=input_text)):
+        results.append(r)
+
+    # Pull the assistant text out of the last vertex with a results dict.
+    for r in reversed(results):
+        vertex = getattr(r, "vertex", None)
+        if vertex is None:
+            continue
+        vres = getattr(vertex, "results", None)
+        if isinstance(vres, dict):
+            for v in vres.values():
+                text = _text_of(v)
+                if text:
+                    return {"text": text}
+        text = _text_of(vres)
+        if text:
+            return {"text": text}
+    return {"text": None}
+
+
+def _text_of(value) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, str) and value.strip():
+        return value
+    for attr in ("text", "message", "data"):
+        sub = getattr(value, attr, None)
+        if isinstance(sub, str) and sub.strip():
+            return sub
+        if isinstance(sub, dict):
+            for k in ("text", "message"):
+                if isinstance(sub.get(k), str) and sub[k].strip():
+                    return sub[k]
+    if isinstance(value, dict):
+        for k in ("text", "message"):
+            if isinstance(value.get(k), str) and value[k].strip():
+                return value[k]
+    return None
+
+
+def main() -> int:
+    leaked = [name for name in FORBIDDEN_ENV if name in os.environ]
+    if leaked:
+        _emit({"status": "boot_failed", "reason": "forbidden_env", "vars": leaked})
+        return 2
+    required = ("MT_RUN_TOKEN", "MT_RUNTIME_API_URL", "FLOW_PATH")
+    missing = [name for name in required if not os.environ.get(name)]
+    if missing:
+        _emit({"status": "boot_failed", "reason": "missing_env", "missing": missing})
+        return 2
+    try:
+        result = asyncio.run(_run())
+        _emit({"status": "ok", **result})
+        return 0
+    except PermissionError as exc:
+        _emit({"status": "denied", "error": str(exc)})
+        return 0
+    except Exception as exc:  # noqa: BLE001
+        _emit(
+            {
+                "status": "error",
+                "error": f"{type(exc).__name__}: {exc}",
+                "traceback": traceback.format_exc()[-2000:],
+            }
+        )
+        return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pyproject.toml

@@ -222,6 +222,7 @@ exclude = [
     "src/frontend/tests/assets/*",
     "src/lfx/src/lfx/_assets/component_index.json",
     "docs/**/API-Reference/python-examples/**",
+    "prototypes/**",
 ]
 line-length = 120