feat(integrations): add persistent connection API

design/dedicated-integrations/connection-contract.md

@@ -5,7 +5,7 @@ Decision ID: connection-contract
 Applies to: INT-2 (lfx), with the langflow-base obligations INT-4 and INT-5 must meet and the Enterprise seams
 Owners (sign-off roles): lfx owner, langflow-base owner, Enterprise owner, frontend owner
 Last verified: 2026-09-01
-Last amended: 2026-09-03 (INT-2 implementation review)
+Last amended: 2026-09-03 (connection persistence implementation review)
 
 This document is the INT-2 design that the discovery gate asks the lfx, langflow-base, and Enterprise owners to sign
 off before INT-2 is built. Each section states the recommended decision, why, and what was rejected. Section 12
@@ -78,8 +78,8 @@ authorization.
   account identity is ever stored in flow JSON.
 - `required_connections`: the deployment artifact manifest adds `required_connections: [{provider, name, scopes}]`
   per flow and aggregated, beside `required_variables`
-  (`src/backend/base/langflow/services/deployment_artifacts/builder.py:347`); scopes come from the input's declared
-  `required_scopes`. The manifest `schema_version` bump is an INT-4 decision.
+  (`src/backend/base/langflow/services/deployment_artifacts/builder.py`); scopes come from the input's declared
+  `required_scopes`. Artifacts with connection requirements use manifest `schema_version: 4`.
 
 Rejected: an opaque connection UUID (does not survive export or `lfx run`, and the UI needs a lookup to display it);
 a `SecretStrInput` subclass with a synthetic variable name (password rendering, `load_from_db=True` from
@@ -385,9 +385,9 @@ identifiers.**
 3. Per-connection `allow_non_interactive` semantics, including `mcp_projects` with auth `none`.
 4. Cross-worker single-flight refresh: a DB lease column versus a Redis lock; the `background_execution`
    lease-claim code is the precedent.
-5. Encryption envelope: extend the `sso_secret.py` HKDF scheme with a new info label, or the Fernet
-   `encrypt_api_key` path used by MCP and variables.
-6. Artifact manifest schema version for `required_connections`.
+5. Encryption envelope decision: use the existing Fernet `encrypt_api_key` path used by MCP and variables, and
+   isolate the encrypted envelope in `connection_secret` so metadata queries never load credential material.
+6. Artifact manifest schema decision: use version 4 when `required_connections` is non-empty.
 7. Desktop: the same `GET /api/v1/connections/{provider}/callback` on `localhost:7860` with a PKCE public client
    that is Langflow-owned by default and customer-owned as the override (`decisions/desktop-oauth-ownership.md`),
    and the redirect allowlist (`127.0.0.1` loopback; Microsoft ignores the port when matching localhost redirects).

docs/docs/API-Reference/api-reference-api-examples.mdx

@@ -377,6 +377,16 @@ The following endpoints are most often used when contributing to the Langflow co
   * PATCH `/v1/variables/{variable_id}`: Update a variable.
   * DELETE `/v1/variables/{variable_id}`: Delete a variable.
 
+* Connections:
+  * GET `/v1/connections`: List owned, instance-owned, and explicitly shared connection metadata. Use the optional `provider` query parameter to filter the list for a connection picker.
+  * POST `/v1/connections`: Create connection metadata and optionally provide credentials for encrypted storage.
+  * POST `/v1/connections/{connection_id}/test`: Test the stored credential and required scope coverage.
+  * POST `/v1/connections/{connection_id}/health`: Refresh the stored connection health.
+  * POST `/v1/connections/{connection_id}/revoke`: Remove stored credentials and mark the connection revoked.
+  * DELETE `/v1/connections/{connection_id}`: Delete a connection.
+
+  Connection responses contain metadata, status, health, granted scopes, and a `has_credentials` flag. They never contain access tokens, refresh tokens, or encrypted credential payloads. User-owned connections can be used non-interactively only when `allow_non_interactive` is enabled; connections cannot be shared publicly.
+
 * [Use voice mode](/concepts-voice-mode):
   * WS `/v1/voice/ws/flow_tts/{flow_id}`: Speech-to-text session that runs a flow and returns TTS.
   * WS `/v1/voice/ws/flow_tts/{flow_id}/{session_id}`: Same as above with explicit session ID.

docs/docs/Develop/memory.mdx

@@ -49,6 +49,8 @@ On OSS, the default is the SQLite file `langflow.db`. Desktop uses `database.db`
 
 • **DeploymentProviderAccount**: Stores configured connections to external deployment providers, including provider URL, tenant settings, and encrypted credentials.
 
+• **Connection** and **ConnectionSecret**: Store integration connection metadata separately from Fernet-encrypted credential payloads. Connection records carry ownership, provider, granted scopes, executing identity, health, and non-interactive-use policy; API reads never join or return the secret payload.
+
 • **File**: Stores metadata for files uploaded to Langflow's file management system, including file names, paths, sizes, and storage providers. For more information, see [Manage files](/concepts-file-management).
 
 • **Flow**: Contains flow definitions, including nodes, edges, and components, stored as JSON or database records. For more information, see [Build flows](/concepts-flows).
@@ -220,4 +222,4 @@ For more information and examples, see [**Message History** component](/message-
 * [Configure an external PostgreSQL database](/configuration-custom-database)
 * [Langflow file management](/concepts-file-management)
 * [Langflow logs](/logging)
-* [Langflow environment variables](/environment-variables)
+* [Langflow environment variables](/environment-variables)

scripts/ci/authz_endpoint_matrix.json

@@ -437,6 +437,28 @@
         "GET|/{job_id}/events|reattach_workflow_events|read|authenticated"
       ]
     },
+    {
+      "family": "connections",
+      "source": "api/v1/connections.py",
+      "resource": "connection",
+      "domain": "caller-global owner or explicit share scope",
+      "privacy": "connection identifier denials are returned as 404; responses contain metadata and health only",
+      "side_effects": "authorize before credential decryption, health checks, revocation, or deletion",
+      "frontend": "connection pickers list only owned, instance, or explicitly shared metadata and never receive token material",
+      "personas": "canonical_v1",
+      "test_references": [
+        "src/backend/tests/unit/api/v1/test_connections.py::test_non_owner_cannot_test_or_delete_connection",
+        "src/backend/tests/unit/api/v1/test_connections.py::test_connection_responses_never_include_tokens"
+      ],
+      "routes": [
+        "GET||list_connections|read|authenticated",
+        "POST||create_connection|create|authenticated",
+        "POST|/{connection_id}/test|test_connection|execute|authenticated",
+        "POST|/{connection_id}/health|refresh_connection_health|execute|authenticated",
+        "POST|/{connection_id}/revoke|revoke_connection|write|authenticated",
+        "DELETE|/{connection_id}|delete_connection|delete|authenticated"
+      ]
+    },
     {
       "family": "workflow_public_v2",
       "source": "api/v2/workflow_public.py",

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

@@ -0,0 +1,103 @@
+"""Add connection metadata and encrypted credential tables.
+
+Revision ID: f3b6a9d2e4c1
+Revises: c9f2e5a7b1d4
+Create Date: 2026-09-03
+
+Phase: EXPAND
+"""
+
+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 = "f3b6a9d2e4c1"  # pragma: allowlist secret
+down_revision: str | None = "c9f2e5a7b1d4"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+CONNECTION_TABLE = "connection"
+SECRET_TABLE = "connection_secret"  # noqa: S105  # pragma: allowlist secret - table name
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if not migration.table_exists(CONNECTION_TABLE, conn):
+        op.create_table(
+            CONNECTION_TABLE,
+            sa.Column("id", sa.Uuid(), nullable=False),
+            sa.Column("owner_id", sa.Uuid(), nullable=True),
+            sa.Column("provider_key", sa.String(length=120), nullable=False),
+            sa.Column("name", sa.String(length=64), nullable=False),
+            sa.Column("display_name", sa.String(length=255), nullable=False),
+            sa.Column("ownership_mode", sa.String(length=16), nullable=False),
+            sa.Column("status", sa.String(length=16), nullable=False),
+            sa.Column("health", sa.String(length=16), nullable=False),
+            sa.Column("granted_scopes", sa.JSON(), nullable=False),
+            sa.Column("executing_identity", sa.JSON(), nullable=False),
+            sa.Column("allow_non_interactive", sa.Boolean(), nullable=False, server_default=sa.false()),
+            sa.Column("health_checked_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.CheckConstraint(
+                "(ownership_mode = 'user' AND owner_id IS NOT NULL) OR "
+                "(ownership_mode = 'instance' AND owner_id IS NULL)",
+                name="ck_connection_owner_mode",
+            ),
+            sa.CheckConstraint(
+                "status IN ('pending', 'ready', 'expired', 'revoked', 'error')",
+                name="ck_connection_status",
+            ),
+            sa.CheckConstraint(
+                "health IN ('unknown', 'healthy', 'unhealthy')",
+                name="ck_connection_health",
+            ),
+            sa.ForeignKeyConstraint(["owner_id"], ["user.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+        op.create_index("ix_connection_owner_id", CONNECTION_TABLE, ["owner_id"], unique=False)
+        op.create_index(
+            "uq_connection_user_provider_name",
+            CONNECTION_TABLE,
+            ["owner_id", "provider_key", "name"],
+            unique=True,
+            sqlite_where=sa.text("ownership_mode = 'user'"),
+            postgresql_where=sa.text("ownership_mode = 'user'"),
+        )
+        op.create_index(
+            "uq_connection_instance_provider_name",
+            CONNECTION_TABLE,
+            ["provider_key", "name"],
+            unique=True,
+            sqlite_where=sa.text("ownership_mode = 'instance'"),
+            postgresql_where=sa.text("ownership_mode = 'instance'"),
+        )
+
+    if not migration.table_exists(SECRET_TABLE, conn):
+        op.create_table(
+            SECRET_TABLE,
+            sa.Column("connection_id", sa.Uuid(), nullable=False),
+            sa.Column("encrypted_payload", sa.Text(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.ForeignKeyConstraint(["connection_id"], ["connection.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("connection_id"),
+        )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if migration.table_exists(SECRET_TABLE, conn):
+        op.drop_table(SECRET_TABLE)
+    if migration.table_exists(CONNECTION_TABLE, conn):
+        op.drop_index("uq_connection_instance_provider_name", table_name=CONNECTION_TABLE)
+        op.drop_index("uq_connection_user_provider_name", table_name=CONNECTION_TABLE)
+        op.drop_index("ix_connection_owner_id", table_name=CONNECTION_TABLE)
+        op.drop_table(CONNECTION_TABLE)

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

@@ -17,6 +17,7 @@
     authz_teams_router,
     catalog_policy_router,
     chat_router,
+    connections_router,
     endpoints_router,
     extensions_router,
     files_router,
@@ -69,6 +70,7 @@ def include_deployment_router(target_router: APIRouter) -> None:
 
 
 router_v1.include_router(chat_router)
+router_v1.include_router(connections_router)
 router_v1.include_router(endpoints_router)
 router_v1.include_router(validate_router)
 router_v1.include_router(store_router)

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

@@ -9,6 +9,7 @@
 from langflow.api.v1.authz_teams import router as authz_teams_router
 from langflow.api.v1.catalog_policy import router as catalog_policy_router
 from langflow.api.v1.chat import router as chat_router
+from langflow.api.v1.connections import router as connections_router
 from langflow.api.v1.endpoints import router as endpoints_router
 from langflow.api.v1.extensions import router as extensions_router
 from langflow.api.v1.files import router as files_router
@@ -48,6 +49,7 @@
     "authz_teams_router",
     "catalog_policy_router",
     "chat_router",
+    "connections_router",
     "endpoints_router",
     "extensions_router",
     "files_router",

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

@@ -26,6 +26,7 @@
     SharePermissionLevel,
     ShareScope,
 )
+from langflow.services.database.models.connection import Connection
 from langflow.services.database.models.deployment.model import Deployment
 from langflow.services.database.models.file.model import File as UserFile
 from langflow.services.database.models.flow.model import Flow
@@ -49,6 +50,7 @@
     "knowledge_base": (KnowledgeBaseRecord, "user_id"),
     "variable": (Variable, "user_id"),
     "file": (UserFile, "user_id"),
+    "connection": (Connection, "owner_id"),
 }
 
 
@@ -291,6 +293,11 @@ async def _ensure_can_administer_share(
 
 def _ensure_supported_share_permission(*, resource_type: str, scope: str, permission_level: str) -> None:
     """Reject share levels that have no matching public flow product behavior."""
+    if resource_type == "connection" and scope == ShareScope.PUBLIC.value:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="Connections cannot be shared publicly.",
+        )
     if resource_type != "flow" or scope != ShareScope.PUBLIC.value:
         return
     if permission_level == SharePermissionLevel.EXECUTE.value: