feat(integrations): add lfx connection resolution contract

BUNDLE_API.md

@@ -43,6 +43,21 @@ that does not list `str(BUNDLE_API_VERSION)` is rejected at install time with
 | `DictInput` / `NestedDictInput` | `lfx.io` |
 | `FileInput` / `LinkInput` | `lfx.io` |
 | `HandleInput` | `lfx.io` |
+| `ConnectionInput` (legacy Composio connection flow) | `lfx.inputs` |
+| `ConnectionRefInput` (portable host-resolved reference) | `lfx.io` |
+
+### Integrations
+
+| Symbol | Source |
+| --- | --- |
+| `ConnectionRef` / `ResolvedCredential` / `CredentialLease` | `lfx.integrations` |
+| `ConnectionResolutionRequest` / `ConnectionStatus` | `lfx.integrations` |
+| `IntegrationError` and typed subclasses / `INTEGRATION_ERROR_CODES` | `lfx.integrations` |
+| `normalize_integration_error()` / `register_error_normalizer()` | `lfx.integrations` |
+| `IntegrationProvider` / `OAuthProfile` / `IntegrationCapability` / `ScopeSet` | `lfx.integrations` |
+| `integration_action()` | `lfx.integrations` |
+| `Component.resolve_connection(field_name)` | `lfx.custom.custom_component.component.Component` |
+| `BaseConnectionResolverService` | `lfx.services.connection` |
 
 ### Outputs
 
@@ -71,6 +86,7 @@ that does not list `str(BUNDLE_API_VERSION)` is rejected at install time with
 | Manifest schema (`extension.json` / `[tool.langflow.extension]`) | `lfx.extension.manifest.ExtensionManifest` |
 | `BundleRef` (one entry in optional `bundles[]`; bundle names must be unique) | `lfx.extension.manifest.BundleRef` |
 | `ProviderManifestEntry` (one entry in the optional `providers[]`) | `lfx.extension.manifest.ProviderManifestEntry` |
+| `IntegrationProvider` entries in optional `integrations[]` | `lfx.integrations.IntegrationProvider` |
 | `LfxCompat` (declared as `manifest.lfx`) | `lfx.extension.manifest.LfxCompat` |
 | `BUNDLE_API_VERSION` (the integer this lfx ships) | `lfx.extension.manifest` |
 | `EXTENSION_SCHEMA_URL` / `SCHEMA_VERSION` | `lfx.extension.manifest` |
@@ -190,6 +206,10 @@ the deserialize half is covered by
 ### v0 (this release)
 
 - Initial surface enumerated above.  Frozen as `BUNDLE_API_VERSION = 1`.
+- Added the provider-neutral connection-reference, resolver, capability,
+  integration-error, and telemetry contracts used by dedicated integration
+  bundles. This is an additive surface change and does not change
+  `BUNDLE_API_VERSION`.
 - `ExtensionManifest.version` now accepts the canonical PEP 440 stable, dev,
   alpha, beta, and release-candidate forms emitted by the repository's bundle
   release pipeline, in addition to the existing SemVer 2.0.0 forms.  Runtime

scripts/migrate/check_bundle_api_changelog.py

@@ -64,6 +64,13 @@
     # The package facade (re-exports define the surface)
     "src/lfx/src/lfx/extension/__init__.py",
     "src/lfx/src/lfx/extension/validate.py",
+    # Dedicated-integration public surface
+    "src/lfx/src/lfx/integrations",
+    "src/lfx/src/lfx/inputs/input_mixin.py",
+    "src/lfx/src/lfx/inputs/inputs.py",
+    "src/lfx/src/lfx/io/__init__.py",
+    "src/lfx/src/lfx/custom/custom_component/component.py",
+    "src/lfx/src/lfx/services/connection",
 )
 
 

src/backend/base/langflow/services/factory.py

@@ -78,7 +78,9 @@ def import_all_services_into_a_dict():
 
             # Shared services live in lfx so both the standalone runtime and
             # the Langflow application use the same contract and instance.
-            if service_name in {"mcp_composer", "model_provider_policy", "policy_bundle"}:
+            if service_name == "connection_resolver":
+                module_name = "lfx.services.connection.env_resolver"
+            elif service_name in {"mcp_composer", "model_provider_policy", "policy_bundle"}:
                 module_name = f"lfx.services.{service_name}.service"
             else:
                 module_name = f"langflow.services.{service_name}.service"

src/backend/base/langflow/services/schema.py

@@ -18,6 +18,7 @@ class ServiceType(str, Enum):
     TASK_SERVICE = "task_service"
     STORE_SERVICE = "store_service"
     VARIABLE_SERVICE = "variable_service"
+    CONNECTION_RESOLVER_SERVICE = "connection_resolver_service"
     STORAGE_SERVICE = "storage_service"
     STATE_SERVICE = "state_service"
     TRACING_SERVICE = "tracing_service"

src/backend/base/langflow/services/telemetry/schema.py

@@ -28,6 +28,18 @@ class DeploymentPayload(BasePayload):
     wxo_tenant_id: str | None = Field(default=None, serialization_alias="wxoTenantId")
 
 
+class IntegrationActionPayload(BasePayload):
+    """Low-cardinality integration action event with no connection identifiers."""
+
+    provider: str
+    capability: str
+    ms: int
+    success: bool
+    error_code: str | None = Field(None, serialization_alias="errorCode")
+    owner_kind: str = Field(serialization_alias="ownerKind")
+    principal_kind: str = Field(serialization_alias="principalKind")
+
+
 class ShutdownPayload(BasePayload):
     time_running: int = Field(serialization_alias="timeRunning")
 

src/lfx/src/lfx/cli/validation/__init__.py

@@ -6,6 +6,7 @@
 
 from lfx.cli.validation._env_validation import (
     is_valid_env_var_name,
+    validate_connection_refs_for_env,
     validate_global_variables_for_env,
 )
 from lfx.cli.validation.core import (
@@ -56,6 +57,7 @@
     "_render_result",
     "is_valid_env_var_name",
     "validate_command",
+    "validate_connection_refs_for_env",
     "validate_flow_file",
     "validate_global_variables_for_env",
 ]

src/lfx/src/lfx/cli/validation/_env_validation.py

@@ -1,10 +1,13 @@
 """Validation utilities for CLI commands."""
 
+from __future__ import annotations
+
 import re
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from lfx.graph.graph.base import Graph
+    from lfx.integrations.errors import ConnectionUnresolvedError
 
 
 def is_valid_env_var_name(name: str) -> bool:
@@ -27,7 +30,7 @@ def is_valid_env_var_name(name: str) -> bool:
     return bool(re.match(pattern, name))
 
 
-def validate_global_variables_for_env(graph: "Graph") -> list[str]:
+def validate_global_variables_for_env(graph: Graph) -> list[str]:
     """Validate that all global variables with load_from_db=True can be used as environment variables.
 
     When the database is not available (noop mode), global variables with load_from_db=True
@@ -70,3 +73,37 @@ def validate_global_variables_for_env(graph: "Graph") -> list[str]:
                 )
 
     return errors
+
+
+def validate_connection_refs_for_env(graph: Graph) -> list[ConnectionUnresolvedError]:
+    """Return typed failures for connection refs absent from headless injection channels."""
+    from lfx.integrations.errors import ConnectionUnresolvedError
+    from lfx.integrations.models import ConnectionRef
+    from lfx.services.variable.request_scope import normalize_parsed_variables
+    from lfx.utils.env_var_security import safe_getenv
+
+    errors: list[ConnectionUnresolvedError] = []
+    request_variables = normalize_parsed_variables(graph.context.get("request_variables") or {})
+    no_env_fallback = bool(graph.context.get("no_env_fallback"))
+
+    for vertex in graph.vertices:
+        template = vertex.data.get("node", {}).get("template", {})
+        for field_name, field in template.items():
+            if not isinstance(field, dict) or field.get("type") != "connection_ref":
+                continue
+            value = vertex.params.get(field_name, field.get("value"))
+            if not value:
+                continue
+            try:
+                ref = ConnectionRef.parse(value)
+            except ValueError:
+                errors.append(ConnectionUnresolvedError(str(value)))
+                continue
+            env_key = ref.env_key()
+            alias = f"x-langflow-global-var-{env_key.lower().replace('_', '-')}"
+            injected = request_variables.get(env_key) or request_variables.get(alias)
+            if not injected and not no_env_fallback:
+                injected = safe_getenv(env_key) or safe_getenv(alias)
+            if not injected:
+                errors.append(ConnectionUnresolvedError(ref.to_handle(), env_key=env_key, provider=ref.provider))
+    return errors

src/lfx/src/lfx/custom/custom_component/component.py

@@ -57,6 +57,7 @@
     from lfx.graph.edge.schema import EdgeData
     from lfx.graph.vertex.base import Vertex
     from lfx.inputs.inputs import InputTypes
+    from lfx.integrations.models import CredentialLease
     from lfx.schema.dataframe import DataFrame
     from lfx.schema.log import LoggableType
     from lfx.services.model_provider_policy import ModelProviderPolicyPurpose
@@ -604,6 +605,37 @@ def get_input(self, name: str) -> Any:
         msg = f"Input {name} not found in {self.__class__.__name__}"
         raise ValueError(msg)
 
+    def resolve_connection(self, field_name: str) -> CredentialLease:
+        """Create a lazy credential lease for a declared connection-reference input."""
+        from lfx.inputs.inputs import ConnectionRefInput
+        from lfx.integrations.models import ConnectionRef, ConnectionResolutionRequest, CredentialLease
+        from lfx.services.authorization.base import ExecutionPrincipal
+        from lfx.services.deps import get_connection_resolver
+
+        input_model = self._inputs.get(field_name)
+        if not isinstance(input_model, ConnectionRefInput):
+            msg = f"Input {field_name!r} is not a ConnectionRefInput"
+            raise TypeError(msg)
+        value = getattr(self, field_name, input_model.value)
+        ref = ConnectionRef.parse(value)
+        if ref.provider != input_model.provider:
+            msg = (
+                f"Connection reference provider {ref.provider!r} does not match "
+                f"declared provider {input_model.provider!r}"
+            )
+            raise ValueError(msg)
+        graph = getattr(self, "graph", None)
+        principal = getattr(graph, "execution_principal", ExecutionPrincipal.unknown())
+        request = ConnectionResolutionRequest(
+            ref=ref,
+            principal=principal,
+            required_scopes=frozenset(input_model.required_scopes),
+            component_id=self.get_id(),
+            flow_id=str(graph.flow_id) if graph is not None and graph.flow_id is not None else None,
+            run_id=str(graph.run_id) if graph is not None and graph.run_id else None,
+        )
+        return CredentialLease(get_connection_resolver(), request)
+
     def get_output(self, name: str) -> Any:
         """Retrieves the output with the specified name.
 

src/lfx/src/lfx/extension/manifest.py

@@ -49,6 +49,9 @@
     model_validator,
 )
 
+from lfx.integrations.capabilities import IntegrationProvider
+from lfx.integrations.models import provider_env_segment
+
 # ---------------------------------------------------------------------------
 # Constants
 # ---------------------------------------------------------------------------
@@ -480,7 +483,7 @@ class ExtensionManifest(BaseModel):
     Required fields:
         - id, version, name, bundles, lfx
     Optional:
-        - description, capabilities, schema (``$schema``)
+        - description, capabilities, integrations, schema (``$schema``)
     Deferred (rejected with ``field-deferred-in-this-milestone`` when set):
         - services, routes, hooks, starter_projects, userConfig
     """
@@ -553,6 +556,11 @@ class ExtensionManifest(BaseModel):
         description="Optional declared capabilities (v0: requiresCredentials only).",
     )
 
+    integrations: tuple[IntegrationProvider, ...] = Field(
+        default=(),
+        description="Provider authentication profiles and executable integration capabilities.",
+    )
+
     # ------------------------------------------------------------------
     # Deferred fields.  We model them as ``None``-only so that downstream
     # tooling can distinguish "absent" from "explicitly set to a value the
@@ -614,6 +622,18 @@ def _validate_provider_name_uniqueness(self) -> ExtensionManifest:
             raise ValueError(msg)
         return self
 
+    @model_validator(mode="after")
+    def _validate_integration_provider_uniqueness(self) -> ExtensionManifest:
+        provider_ids = [provider.provider_id for provider in self.integrations]
+        if len(set(provider_ids)) != len(provider_ids):
+            msg = "Integration provider ids must be unique within an extension"
+            raise ValueError(msg)
+        env_segments = [provider_env_segment(provider_id) for provider_id in provider_ids]
+        if len(set(env_segments)) != len(env_segments):
+            msg = "Integration provider ids must map to unique environment-key segments"
+            raise ValueError(msg)
+        return self
+
     @model_validator(mode="after")
     def _validate_bundle_uniqueness(self) -> ExtensionManifest:
         # Bundle names are public registry and saved-flow namespaces, so they

src/lfx/src/lfx/graph/graph/base.py

@@ -48,6 +48,7 @@
 )
 from lfx.schema.dotdict import dotdict
 from lfx.schema.schema import INPUT_FIELD_NAME, InputType, OutputValue
+from lfx.services.authorization.base import ExecutionPrincipal
 from lfx.services.cache.utils import CacheMiss
 from lfx.services.deps import get_chat_service, get_tracing_service
 from lfx.utils.async_helpers import run_until_complete
@@ -162,6 +163,7 @@ def __init__(
         self.flow_name = flow_name
         self.description = description
         self.user_id = user_id
+        self.execution_principal = ExecutionPrincipal.unknown()
         # Warm-registry templates need the parsed graph structure without
         # executing component constructors at preload/reconcile time. Normal
         # graphs keep the historical eager-instantiation behavior.
@@ -1680,6 +1682,7 @@ def _copy_graph(
                 before_initialize(new_graph)
 
         new_graph.requires_extension_event_replay = self.requires_extension_event_replay
+        new_graph.execution_principal = self.execution_principal
 
         # Store the newly created object in memo
         memo[id(self)] = new_graph
@@ -1724,6 +1727,7 @@ def __setstate__(self, state):
         # Graphs cached before source-flow provenance was introduced remain
         # loadable and simply have no additional trusted storage namespace.
         state.setdefault("source_flow_id", None)
+        state.setdefault("execution_principal", ExecutionPrincipal.unknown())
         run_manager = state["run_manager"]
         if isinstance(run_manager, RunnableVerticesManager):
             state["run_manager"] = run_manager

src/lfx/src/lfx/inputs/__init__.py

@@ -3,6 +3,7 @@
     BoolInput,
     CodeInput,
     ConnectionInput,
+    ConnectionRefInput,
     DataDisplayInput,
     DataFrameInput,
     DataInput,
@@ -41,6 +42,7 @@
     "BoolInput",
     "CodeInput",
     "ConnectionInput",
+    "ConnectionRefInput",
     "DBProviderInput",
     "DataDisplayInput",
     "DataFrameInput",

src/lfx/src/lfx/inputs/input_mixin.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Annotated, Any
+from typing import Annotated, Any, Literal
 
 from pydantic import (
     BaseModel,
@@ -12,6 +12,8 @@
 
 from lfx.field_typing.range_spec import RangeSpec
 from lfx.inputs.validators import CoalesceBool
+from lfx.integrations.capabilities import ConditionalScopeRequirement
+from lfx.integrations.models import PROVIDER_ID_PATTERN
 from lfx.schema.cross_module import CrossModuleModel
 
 
@@ -27,6 +29,7 @@ class FieldTypes(str, Enum):
     ACTION_PICKER = "actionPicker"
     DURATION = "duration"
     CONNECTION = "connect"
+    CONNECTION_REF = "connection_ref"
     AUTH = "auth"
     FILE = "file"
     PROMPT = "prompt"
@@ -55,6 +58,7 @@ class FieldTypes(str, Enum):
     FieldTypes.AUTH,
     FieldTypes.FILE,
     FieldTypes.CONNECTION,
+    FieldTypes.CONNECTION_REF,
     FieldTypes.MCP,
 }
 
@@ -356,6 +360,17 @@ class ConnectionMixin(BaseModel):
     """List of dictionaries with metadata for each option."""
 
 
+class ConnectionRefMixin(BaseModel):
+    """Provider and capability metadata for a portable connection reference."""
+
+    provider: str = Field(pattern=PROVIDER_ID_PATTERN, max_length=120)
+    auth_profile_id: str = ""
+    required_scopes: list[str] = Field(default_factory=list)
+    conditional_scopes: list[ConditionalScopeRequirement] = Field(default_factory=list)
+    identity_kind: Literal["user", "instance", "any"] = "any"
+    capabilities: list[str] = Field(default_factory=list)
+
+
 class TabMixin(BaseModel):
     """Mixin for tab input fields that allows a maximum of 3 values, each with a maximum of 20 characters."""