hanzoai/flow
2
0
Fork 0
mirror of https://github.com/hanzoai/flow.git synced 2026-10-17 01:29:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Backend imports resolve and the server boots

Browse Source 
The upstream merge left the backend unable to start: lfx.memory imported
has_flow_db_backend, which did not exist, and appended a second
import-time dispatch that named an unimported function. Around it, 17
modules referenced names whose imports were gone, several re-export
packages were empty or pointed at modules that no longer exist, and
settings/base.py still held the monolithic Settings class the groups
replaced (without cache_dir and mcp_base_url).

Each file now matches its upstream counterpart at the merge point, with
this fork's own changes kept: the /v1 and /v2 API paths, the org_id
columns, Hanzo IAM in auth, and product naming. Empty leftover packages
from the rename are removed, and the assistant flow file is
FlowAssistant.json to match FLOW_ASSISTANT_FLOW.

python -m flow run --backend-only starts on SQLite and answers /health,
/health_check and /v1/version.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zooqueen
2026-09-15 11:56:02 -07:00
co-authored by Hanzo Dev
parent 6697718edc
commit 1e1310e05a
83 changed files with 522 additions and 1190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/base/flow/__init__.py
+220
View File
@@ -0,0 +1,220 @@
"""Flow backwards compatibility layer.
This module provides backwards compatibility by forwarding imports from
flow.* to lfx.* to maintain compatibility with existing code that
references the old flow module structure.
"""
from flow.helpers.windows_postgres_helper import configure_windows_postgres_event_loop
configure_windows_postgres_event_loop(source="package_init")
import importlib # noqa: E402
import importlib.util # noqa: E402
import sys # noqa: E402
from types import ModuleType # noqa: E402
from typing import Any # noqa: E402
class FlowCompatibilityModule(ModuleType):
"""A module that forwards attribute access to the corresponding lfx module."""
def __init__(self, name: str, lfx_module_name: str):
super().__init__(name)
self._lfx_module_name = lfx_module_name
self._lfx_module = None
def _get_lfx_module(self):
"""Lazily import and cache the lfx module."""
if self._lfx_module is None:
try:
self._lfx_module = importlib.import_module(self._lfx_module_name)
except ImportError as e:
msg = f"Cannot import {self._lfx_module_name} for backwards compatibility with {self.__name__}"
raise ImportError(msg) from e
return self._lfx_module
def __getattr__(self, name: str) -> Any:
"""Forward attribute access to the lfx module with caching."""
lfx_module = self._get_lfx_module()
try:
attr = getattr(lfx_module, name)
except AttributeError as e:
msg = f"module '{self.__name__}' has no attribute '{name}'"
raise AttributeError(msg) from e
else:
# Cache the attribute in our __dict__ for faster subsequent access
setattr(self, name, attr)
return attr
def __dir__(self):
"""Return directory of the lfx module."""
try:
lfx_module = self._get_lfx_module()
return dir(lfx_module)
except ImportError:
return []
def _setup_compatibility_modules():
"""Set up comprehensive compatibility modules for flow.base imports."""
# First, set up the base attribute on this module (flow)
current_module = sys.modules[__name__]
# Define all the modules we need to support
module_mappings = {
# Core base module
"flow.base": "lfx.base",
# Inputs module - critical for class identity
"flow.inputs": "lfx.inputs",
"flow.inputs.inputs": "lfx.inputs.inputs",
# Schema modules - also critical for class identity
"flow.schema": "lfx.schema",
"flow.schema.data": "lfx.schema.data",
"flow.schema.serialize": "lfx.schema.serialize",
# Template modules
"flow.template": "lfx.template",
"flow.template.field": "lfx.template.field",
"flow.template.field.base": "lfx.template.field.base",
# Components modules
"flow.components": "lfx.components",
"flow.components.helpers": "lfx.components.helpers",
"flow.components.helpers.calculator_core": "lfx.components.helpers.calculator_core",
"flow.components.helpers.create_list": "lfx.components.helpers.create_list",
"flow.components.helpers.current_date": "lfx.components.helpers.current_date",
"flow.components.helpers.id_generator": "lfx.components.helpers.id_generator",
"flow.components.helpers.memory": "lfx.components.helpers.memory",
"flow.components.helpers.output_parser": "lfx.components.helpers.output_parser",
"flow.components.helpers.store_message": "lfx.components.helpers.store_message",
# Individual modules that exist in lfx
"flow.base.agents": "lfx.base.agents",
"flow.base.chains": "lfx.base.chains",
"flow.base.data": "lfx.base.data",
"flow.base.data.utils": "lfx.base.data.utils",
"flow.base.document_transformers": "lfx.base.document_transformers",
"flow.base.embeddings": "lfx.base.embeddings",
"flow.base.flow_processing": "lfx.base.flow_processing",
"flow.base.io": "lfx.base.io",
"flow.base.io.chat": "lfx.base.io.chat",
"flow.base.io.text": "lfx.base.io.text",
"flow.base.langchain_utilities": "lfx.base.langchain_utilities",
"flow.base.memory": "lfx.base.memory",
"flow.base.models": "lfx.base.models",
"flow.base.models.google_generative_ai_constants": "lfx.base.models.google_generative_ai_constants",
"flow.base.models.openai_constants": "lfx.base.models.openai_constants",
"flow.base.models.anthropic_constants": "lfx.base.models.anthropic_constants",
"flow.base.models.aiml_constants": "lfx.base.models.aiml_constants",
"flow.base.models.aws_constants": "lfx.base.models.aws_constants",
"flow.base.models.groq_constants": "lfx.base.models.groq_constants",
"flow.base.models.novita_constants": "lfx.base.models.novita_constants",
"flow.base.models.ollama_constants": "lfx.base.models.ollama_constants",
"flow.base.models.sambanova_constants": "lfx.base.models.sambanova_constants",
"flow.base.models.cometapi_constants": "lfx.base.models.cometapi_constants",
"flow.base.prompts": "lfx.base.prompts",
"flow.base.prompts.api_utils": "lfx.base.prompts.api_utils",
"flow.base.prompts.utils": "lfx.base.prompts.utils",
"flow.base.textsplitters": "lfx.base.textsplitters",
"flow.base.tools": "lfx.base.tools",
"flow.base.vectorstores": "lfx.base.vectorstores",
}
# Create compatibility modules for each mapping
for flow_name, lfx_name in module_mappings.items():
if flow_name not in sys.modules:
# Check if the lfx module exists
try:
spec = importlib.util.find_spec(lfx_name)
if spec is not None:
# Create compatibility module
compat_module = FlowCompatibilityModule(flow_name, lfx_name)
sys.modules[flow_name] = compat_module
# Set up the module hierarchy
parts = flow_name.split(".")
if len(parts) > 1:
parent_name = ".".join(parts[:-1])
parent_module = sys.modules.get(parent_name)
if parent_module is not None:
setattr(parent_module, parts[-1], compat_module)
# Special handling for top-level modules
if flow_name == "flow.base":
current_module.base = compat_module
elif flow_name == "flow.inputs":
current_module.inputs = compat_module
elif flow_name == "flow.schema":
current_module.schema = compat_module
elif flow_name == "flow.template":
current_module.template = compat_module
elif flow_name == "flow.components":
current_module.components = compat_module
except (ImportError, ValueError):
# Skip modules that don't exist in lfx
continue
# Handle modules that exist only in flow (like knowledge_bases)
# These need special handling because they're not in lfx yet
flow_only_modules = {
"flow.base.data.kb_utils": "flow.base.data.kb_utils",
"flow.base.knowledge_bases": "flow.base.knowledge_bases",
"flow.components.knowledge_bases": "flow.components.knowledge_bases",
}
for flow_name in flow_only_modules:
if flow_name not in sys.modules:
try:
# Try to find the actual physical module file
from pathlib import Path
base_dir = Path(__file__).parent
if flow_name == "flow.base.data.kb_utils":
kb_utils_file = base_dir / "base" / "data" / "kb_utils.py"
if kb_utils_file.exists():
spec = importlib.util.spec_from_file_location(flow_name, kb_utils_file)
if spec is not None and spec.loader is not None:
module = importlib.util.module_from_spec(spec)
sys.modules[flow_name] = module
spec.loader.exec_module(module)
# Also add to parent module
parent_module = sys.modules.get("flow.base.data")
if parent_module is not None:
parent_module.kb_utils = module
elif flow_name == "flow.base.knowledge_bases":
kb_dir = base_dir / "base" / "knowledge_bases"
kb_init_file = kb_dir / "__init__.py"
if kb_init_file.exists():
spec = importlib.util.spec_from_file_location(flow_name, kb_init_file)
if spec is not None and spec.loader is not None:
module = importlib.util.module_from_spec(spec)
sys.modules[flow_name] = module
spec.loader.exec_module(module)
# Also add to parent module
parent_module = sys.modules.get("flow.base")
if parent_module is not None:
parent_module.knowledge_bases = module
elif flow_name == "flow.components.knowledge_bases":
components_kb_dir = base_dir / "components" / "knowledge_bases"
components_kb_init_file = components_kb_dir / "__init__.py"
if components_kb_init_file.exists():
spec = importlib.util.spec_from_file_location(flow_name, components_kb_init_file)
if spec is not None and spec.loader is not None:
module = importlib.util.module_from_spec(spec)
sys.modules[flow_name] = module
spec.loader.exec_module(module)
# Also add to parent module
parent_module = sys.modules.get("flow.components")
if parent_module is not None:
parent_module.knowledge_bases = module
except (ImportError, AttributeError):
# If direct file loading fails, skip silently
continue
# Set up all the compatibility modules
_setup_compatibility_modules()
src/backend/base/flow/agentic/api/router.py
-1
View File
@@ -30,7 +30,6 @@ from flow.agentic.services.flow_types import (
MAX_VALIDATION_RETRIES,
)
from flow.agentic.services.provider_service import (
DEFAULT_MODELS,
PREFERRED_PROVIDERS,
get_default_model,
get_enabled_providers_for_user,
src/backend/base/flow/agentic/flows/LangflowAssistant.json → src/backend/base/flow/agentic/flows/FlowAssistant.json
View File
src/backend/base/flow/agentic/flows/__init__.py
-1
View File
@@ -7,7 +7,6 @@ Available flows:
- FlowAssistant.json: Main assistant flow for Q&A and component generation (JSON)
"""
from flow.agentic.flows.flow_assistant import get_graph as get_flow_assistant_graph
from flow.agentic.flows.translation_flow import get_graph as get_translation_flow_graph
__all__ = [
src/backend/base/flow/agentic/services/assistant_service.py
+5 -1
View File
@@ -10,7 +10,9 @@ from fastapi import HTTPException
from lfx.log.logger import logger
from flow.agentic.helpers.code_extraction import extract_component_code
from flow.agentic.helpers.code_security import scan_code_security
from flow.agentic.helpers.error_handling import extract_friendly_error
from flow.agentic.helpers.input_sanitization import REFUSAL_MESSAGE, sanitize_input
from flow.agentic.helpers.sse import (
format_cancelled_event,
format_complete_event,
@@ -18,13 +20,15 @@ from flow.agentic.helpers.sse import (
format_progress_event,
format_token_event,
)
from flow.agentic.helpers.validation import validate_component_code
from flow.agentic.helpers.streaming_retry import emit_execution_retry_events
from flow.agentic.helpers.validation import validate_component_code, validate_component_runtime
from flow.agentic.services.flow_executor import (
execute_flow_file,
execute_flow_file_streaming,
extract_response_text,
)
from flow.agentic.services.flow_types import (
EXECUTION_RETRY_TEMPLATE,
MAX_VALIDATION_RETRIES,
OFF_TOPIC_REFUSAL_MESSAGE,
VALIDATION_RETRY_TEMPLATE,
src/backend/base/flow/agentic/services/flow_types.py
+8 -2
View File
@@ -18,8 +18,14 @@ STREAMING_EVENT_TIMEOUT_SECONDS = 300.0
# Assistant configuration
MAX_VALIDATION_RETRIES = 3
VALIDATION_UI_DELAY_SECONDS = 0.3
FLOW_ASSISTANT_FLOW = "FlowAssistant"
TRANSLATION_FLOW = "TranslationFlow"
FLOW_ASSISTANT_FLOW = "FlowAssistant.json"
TRANSLATION_FLOW = "translation_flow.py"
OFF_TOPIC_REFUSAL_MESSAGE = (
"I appreciate your interest, but I'm the Flow Assistant and can only help with "
"Flow-related topics such as building components, creating flows, configuring "
"deployments, and troubleshooting issues. Could you rephrase your question about Flow?"
)
VALIDATION_RETRY_TEMPLATE = """The previous component code has an error. Please fix it.
src/backend/base/flow/agentic/services/provider_service.py
-1
View File
@@ -12,7 +12,6 @@ from lfx.log.logger import logger
from sqlalchemy.ext.asyncio import AsyncSession
from flow.services.deps import get_variable_service
from flow.services.variable.constants import CREDENTIAL_TYPE
from flow.services.variable.service import DatabaseVariableService, VariableService
# Preferred providers in order of priority
src/backend/base/flow/api/__init__.py
+3 -2
View File
@@ -1,5 +1,6 @@
from flow.api.health_check_router import health_check_router
from flow.api.log_router import log_router
from flow.api.router import router
__all__ = ["health_check_router", "log_router", "router"]
# Note: router is imported directly via flow.api.router to avoid circular imports
# Use: from flow.api.router import router
__all__ = ["health_check_router", "log_router"]
src/backend/base/flow/api/utils/__init__.py
+1 -2
View File
@@ -5,7 +5,7 @@ from the core module. This ensures existing imports continue to work while
allowing for better code organization.
"""
# Re-export everything from core module to maintain backward compatibility
# Re-export everything from core and flow_utils modules to maintain backward compatibility
from flow.api.utils.core import (
API_WORDS,
MAX_PAGE_SIZE,
@@ -18,7 +18,6 @@ from flow.api.utils.core import (
ValidatedFileName,
ValidatedFolderName,
build_input_keys_response,
cascade_delete_flow,
check_flow_version,
custom_params,
extract_global_variables_from_headers,
src/backend/base/flow/api/utils/core.py
-67
View File
@@ -1,7 +1,6 @@
from __future__ import annotations
import json as _json
import uuid
from datetime import timedelta
from enum import Enum
from typing import TYPE_CHECKING, Annotated, Any
@@ -14,18 +13,12 @@ from lfx.utils.validate_cloud import raise_error_if_astra_cloud_disable_componen
from sqlmodel.ext.asyncio.session import AsyncSession
from flow.services.auth.utils import get_current_active_user, get_current_active_user_mcp
from flow.services.deps import session_scope
from flow.services.database.models.flow.model import Flow
from flow.services.database.models.flow_version.model import FlowVersion
from flow.services.database.models.message.model import MessageTable
from flow.services.database.models.transactions.model import TransactionTable
from flow.services.database.models.user.model import User
from flow.services.database.models.vertex_builds.model import VertexBuildTable
from flow.services.store.utils import get_lf_version_from_pypi
from flow.utils.constants import FLOW_GLOBAL_VAR_HEADER_PREFIX
if TYPE_CHECKING:
from flow.services.chat.service import ChatService
from flow.services.store.schema import StoreComponentCreate
@@ -405,66 +398,6 @@ def custom_params(
return Params(page=page or MIN_PAGE_SIZE, size=size or MAX_PAGE_SIZE)
async def verify_public_flow_and_get_user(flow_id: uuid.UUID, client_id: str | None) -> tuple[User, uuid.UUID]:
"""Verify a public flow request and generate a deterministic flow ID.
This utility function:
1. Checks that a client_id cookie is provided
2. Verifies the flow exists and is marked as PUBLIC
3. Creates a deterministic UUID based on client_id and original flow_id
4. Retrieves the flow owner user for permission purposes
This function is used to support public flow endpoints that don't require
authentication but still need to operate within the permission model.
Args:
flow_id: The original flow ID to verify
client_id: The client ID from the request cookie
Returns:
tuple: (flow owner user, deterministic flow ID for tracking)
Raises:
HTTPException:
- 400 if no client_id is provided
- 403 if flow doesn't exist or isn't public
- 403 if unable to retrieve the flow owner user
- 403 if user is not found for public flow
"""
if not client_id:
raise HTTPException(status_code=400, detail="No client_id cookie found")
# Check if the flow is public
async with session_scope() as session:
from sqlmodel import select
from flow.services.database.models.flow.model import AccessTypeEnum, Flow
flow = (await session.exec(select(Flow).where(Flow.id == flow_id))).first()
if not flow or flow.access_type is not AccessTypeEnum.PUBLIC:
raise HTTPException(status_code=403, detail="Flow is not public")
# Create a new flow ID using the client_id and flow_id
new_id = f"{client_id}_{flow_id}"
new_flow_id = uuid.uuid5(uuid.NAMESPACE_DNS, new_id)
# Get the user associated with the flow
try:
from flow.helpers.user import get_user_by_flow_id_or_endpoint_name
user = await get_user_by_flow_id_or_endpoint_name(str(flow_id))
except Exception as exc:
await logger.aexception(f"Error getting user for public flow {flow_id}")
raise HTTPException(status_code=403, detail="Flow is not accessible") from exc
if not user:
msg = f"User not found for public flow {flow_id}"
raise HTTPException(status_code=403, detail=msg)
return user, new_flow_id
def extract_global_variables_from_headers(headers) -> dict[str, str]:
"""Extract global variables from HTTP headers with prefix X-FLOW-GLOBAL-VAR-*.
src/backend/base/flow/api/v1/__init__.py
+2 -2
View File
@@ -1,8 +1,8 @@
from flow.api.v1.api_key import router as api_key_router
from flow.api.v1.chat import router as chat_router
from flow.api.v1.deployments import router as deployment_router
from flow.api.v1.endpoints import router as endpoints_router
from flow.api.v1.files import router as files_router
from flow.api.v1.flow_events import router as flow_events_router
from flow.api.v1.flow_version import router as flow_version_router
from flow.api.v1.flows import router as flows_router
from flow.api.v1.folders import router as folders_router
@@ -26,9 +26,9 @@ from flow.api.v1.voice_mode import router as voice_mode_router
__all__ = [
"api_key_router",
"chat_router",
"deployment_router",
"endpoints_router",
"files_router",
"flow_events_router",
"flow_version_router",
"flows_router",
"folders_router",
src/backend/base/flow/api/v1/chat.py
+3 -2
View File
@@ -43,9 +43,10 @@ from flow.api.v1.schemas import (
VerticesOrderResponse,
)
from flow.exceptions.component import ComponentBuildError
from flow.services.auth.utils import get_current_active_user
from flow.services.auth.utils import get_current_active_user, get_current_user_optional
from flow.services.chat.service import ChatService
from flow.services.database.models.flow.model import Flow
from flow.services.database.models.flow.model import AccessTypeEnum, Flow
from flow.services.database.models.user.model import User
from flow.services.deps import (
get_chat_service,
get_queue_service,
src/backend/base/flow/api/v1/deployments.py
+24
View File
@@ -23,6 +23,30 @@ from lfx.services.adapters.deployment.schema import (
from pydantic import AfterValidator, StringConstraints
from flow.api.utils import CurrentActiveUser, DbSession, DbSessionReadOnly
from flow.api.v1.mappers.deployments import get_deployment_mapper
from flow.api.v1.mappers.deployments.helpers import (
apply_flow_version_patch_attachments,
attach_flow_versions,
deployment_pagination_params,
flow_version_ids_for_flows,
get_deployment_row_or_404,
get_owned_provider_account_or_404,
handle_adapter_errors,
list_deployment_flow_versions_synced,
list_deployments_synced,
page_offset,
raise_http_for_value_error,
resolve_adapter_from_deployment,
resolve_adapter_mapper_from_deployment,
resolve_added_snapshot_bindings_for_update,
resolve_deployment_adapter,
resolve_flow_version_patch_for_update,
resolve_project_id_for_deployment_create,
resolve_snapshot_map_for_create,
rollback_provider_create,
rollback_provider_update,
validate_project_scoped_flow_version_ids,
)
from flow.api.v1.schemas.deployments import (
DeploymentConfigListResponse,
DeploymentCreateRequest,
src/backend/base/flow/api/v1/files.py
+1 -1
View File
@@ -13,7 +13,7 @@ from fastapi.responses import StreamingResponse
from lfx.services.settings.service import SettingsService
from lfx.utils.helpers import build_content_type_from_extension
from flow.api.utils import CurrentActiveUser, DbSession, ValidatedFileName
from flow.api.utils import CurrentActiveUser, DbSession, ValidatedFileName, ValidatedFolderName
from flow.api.v1.schemas import UploadFileResponse
from flow.services.database.models.flow.model import Flow
from flow.services.deps import get_settings_service, get_storage_service
src/backend/base/flow/api/v1/flow_version.py
+3 -1
View File
@@ -10,8 +10,10 @@ from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from flow.api.utils import CurrentActiveUser, DbSession, DbSessionReadOnly
from flow.api.utils import CurrentActiveUser, DbSession
from flow.api.utils.core import remove_api_keys
from flow.api.v1.mappers.deployments.helpers import get_owned_provider_account_or_404
from flow.api.v1.mappers.deployments.sync import sync_flow_version_attachments
from flow.services.database.models.flow.model import Flow, FlowRead
from flow.services.database.models.flow_version.crud import (
create_flow_version_entry,
src/backend/base/flow/api/v1/flows.py
+19 -36
View File
@@ -2,39 +2,47 @@ from __future__ import annotations
import asyncio
import io
import json
import threading
import zipfile
from datetime import datetime, timezone
from typing import Annotated
from uuid import UUID
import orjson
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from fastapi.encoders import jsonable_encoder
from fastapi.responses import StreamingResponse
from fastapi_pagination import Page, Params
from fastapi_pagination.ext.sqlmodel import apaginate
from lfx.services.cache.utils import CACHE_MISS
from sqlmodel import and_, col, select
from flow.api.utils import CurrentActiveUser, DbSession, cascade_delete_flow, remove_api_keys, validate_is_component
from flow.api.utils.core import normalize_code_for_import
from flow.api.utils import (
CurrentActiveUser,
DbSession,
cascade_delete_flow,
normalize_code_for_import,
validate_is_component,
)
from flow.api.utils.zip_utils import extract_flows_from_zip
from flow.api.v1.flows_helpers import (
_build_flows_download_response,
_get_safe_flow_path,
_new_flow,
_patch_flow,
_read_flow,
_save_flow_to_fs,
_update_existing_flow,
_upsert_flow_list,
_verify_fs_path,
)
from flow.api.v1.mappers.deployments.sync import retry_flow_operation_on_deployment_guard
from flow.api.v1.schemas import FlowListCreate
from flow.helpers.user import get_user_by_flow_id_or_endpoint_name
from flow.services.database.models.deployment.exceptions import araise_if_deployment_guard_error_or_skip
from flow.initial_setup.constants import STARTER_FOLDER_NAME
from flow.services.auth.utils import get_current_active_user
from flow.services.cache.service import ThreadingInMemoryCache
from flow.services.database.models.deployment.exceptions import (
araise_if_deployment_guard_error_or_skip,
)
from flow.services.database.models.flow.model import (
AccessTypeEnum,
Flow,
@@ -43,10 +51,12 @@ from flow.services.database.models.flow.model import (
FlowRead,
FlowUpdate,
)
from flow.services.database.models.flow.utils import get_webhook_component_in_flow
# TODO: Full-version import/export is planned as a follow-up feature. When implemented,
# re-add imports for create_flow_version_entry, get_flow_versions_with_provider_status, strip_version_data,
# and FlowVersionError from the flow_version modules.
from flow.services.database.models.folder.constants import DEFAULT_FOLDER_NAME
from flow.services.database.models.folder.model import Folder
from flow.services.database.models.folder.utils import get_default_folder_id
from flow.services.deps import get_settings_service, get_storage_service
from flow.services.storage.service import StorageService
from flow.utils.compression import compress_response
@@ -505,34 +515,7 @@ async def download_multiple_file(
if not flows:
raise HTTPException(status_code=404, detail="No flows found.")
flows_without_api_keys = [remove_api_keys(flow.model_dump()) for flow in flows]
if len(flows_without_api_keys) > 1:
# Create a byte stream to hold the ZIP file
zip_stream = io.BytesIO()
# Create a ZIP file
with zipfile.ZipFile(zip_stream, "w") as zip_file:
for flow in flows_without_api_keys:
# Convert the flow object to JSON
flow_json = json.dumps(jsonable_encoder(flow))
# Write the JSON to the ZIP file
zip_file.writestr(f"{flow['name']}.json", flow_json)
# Seek to the beginning of the byte stream
zip_stream.seek(0)
# Generate the filename with the current datetime
current_time = datetime.now(tz=timezone.utc).astimezone().strftime("%Y%m%d_%H%M%S")
filename = f"{current_time}_flow_flows.zip"
return StreamingResponse(
zip_stream,
media_type="application/x-zip-compressed",
headers={"Content-Disposition": f"attachment; filename={filename}"},
)
return flows_without_api_keys[0]
return _build_flows_download_response(flows)
# 5 minutes
src/backend/base/flow/api/v1/projects.py
+11 -18
View File
@@ -11,40 +11,33 @@ from sqlalchemy import or_, update
from sqlalchemy.orm import selectinload
from sqlmodel import select
from flow.api.utils import CurrentActiveUser, DbSession, cascade_delete_flow, custom_params, remove_api_keys
from flow.api.utils.mcp.config_utils import validate_mcp_server_for_project
from flow.api.v1.auth_helpers import handle_auth_settings_update
from flow.api.v1.flows import create_flows
from flow.api.v1.mcp_projects import (
get_project_sse_url, # noqa: F401
get_project_streamable_http_url,
register_project_with_composer,
from flow.api.utils import (
CurrentActiveUser,
DbSession,
cascade_delete_flow,
custom_params,
)
from flow.api.v1.schemas import FlowListCreate
from flow.api.v2.mcp import update_server
from flow.api.v1.auth_helpers import handle_auth_settings_update
from flow.api.v1.mappers.deployments.sync import (
retry_flow_operation_on_deployment_guard,
retry_project_operation_on_deployment_guard,
)
from flow.api.v1.mcp_projects import register_project_with_composer
from flow.api.v1.projects_files import download_project_flows, upload_project_flows
from flow.api.v1.projects_mcp_helpers import (
cleanup_mcp_on_delete,
handle_mcp_server_rename,
register_mcp_servers_for_project,
)
from flow.helpers.flow import generate_unique_flow_name
from flow.helpers.folders import generate_unique_folder_name
from flow.initial_setup.constants import ASSISTANT_FOLDER_NAME, STARTER_FOLDER_NAME
from flow.services.auth.mcp_encryption import encrypt_auth_settings
from flow.services.database.models.deployment.exceptions import (
araise_if_deployment_guard_error_or_skip,
remap_flow_guard_for_project_delete,
)
from flow.services.database.models.deployment.guards import check_project_has_deployments
from flow.services.database.models.deployment.orm_guards import ensure_flow_moves_allowed
from flow.initial_setup.constants import ASSISTANT_FOLDER_NAME, STARTER_FOLDER_NAME
from flow.services.auth.mcp_encryption import encrypt_auth_settings
from flow.services.database.models.api_key.crud import create_api_key
from flow.services.database.models.api_key.model import ApiKeyCreate
from flow.services.database.models.flow.model import Flow, FlowCreate, FlowRead
from flow.services.database.models.flow.model import Flow, FlowRead
from flow.services.database.models.folder.constants import DEFAULT_FOLDER_NAME
from flow.services.database.models.folder.model import (
Folder,
@@ -54,7 +47,7 @@ from flow.services.database.models.folder.model import (
FolderUpdate,
)
from flow.services.database.models.folder.pagination_model import FolderWithPaginatedFlows
from flow.services.deps import get_service, get_settings_service, get_storage_service
from flow.services.deps import get_service, get_settings_service
from flow.services.schema import ServiceType
router = APIRouter(prefix="/projects", tags=["Projects"])
src/backend/base/flow/api/v1/schemas/deployments.py
+40 -210
View File
@@ -22,30 +22,9 @@ Two identifier domains coexist in these schemas:
are defined by the provider adapter. Flow forwards them without
interpreting their schema.
Service-layer schema reuse (shared-kernel pattern)
---------------------------------------------------
Three service-layer data schemas are imported and subclassed (via
``_Strict*`` wrappers) rather than redefined in this module. They act as
a **shared kernel**: field definitions owned by the service layer that the
API layer extends with stricter validation (``extra = "forbid"``).
* ``BaseDeploymentData`` -- deployment metadata for creation
* ``BaseDeploymentDataUpdate`` -- deployment metadata for partial updates
* ``DeploymentConfig`` -- deployment configuration payload
Additionally, ``DeploymentType`` is imported as a shared vocabulary enum.
This coupling is intentional -- these schemas carry no Flow-managed
identifiers and describe provider-facing data whose shape the API should
track automatically. If the service layer later introduces fields that
must *not* be API-visible, replace the ``_Strict*`` subclass with an
API-owned model and a mapping function.
``BaseDeploymentData`` also carries an optional ``provider_spec`` dict
(inherited from ``ProviderSpecModel``), an opaque provider-owned input
payload similar to ``provider_data``. ``DeploymentConfig`` carries an
analogous ``provider_config`` dict. ``BaseDeploymentDataUpdate`` has no
opaque provider fields.
DeploymentType is imported from the adapter service layer as shared
vocabulary. Request/response models in this module are API-owned to keep
the client-facing schema minimal and avoid exposing service-only fields.
"""
from __future__ import annotations
@@ -157,18 +136,19 @@ def _validate_detect_vars_request_ids(values: list[UUID]) -> list[UUID]:
class DeploymentProviderAccountCreateRequest(BaseModel):
model_config = {"extra": "forbid"}
provider_tenant_id: NonEmptyStr | None = Field(
default=None,
description="Provider-owned tenant/organization id. Flow persists this opaque value.",
name: NonEmptyStr = Field(
description=(
"User-chosen display name for this provider account. Must be unique per user within a provider_key."
),
)
provider_key: NonEmptyStr = Field(description="Deployment provider key.")
provider_url: NonEmptyStr = Field(
description="Provider service URL persisted in Flow DB for provider-account resolution.",
)
api_key: SecretStr = Field(
provider_key: DeploymentProviderKey = Field(description="Deployment provider key.")
provider_data: dict[str, Any] = Field(
min_length=1,
description=(
"Provider credential material. Stored by Flow as secret data and never returned in read responses."
"Provider-specific credential/metadata payload. "
"Contents are opaque to the API schema; the deployment mapper "
"for the target provider_key validates and extracts credentials "
"and provider metadata (for example URL/region and tenant/account identifiers)."
),
)
@@ -203,15 +183,15 @@ class DeploymentProviderAccountUpdateRequest(BaseModel):
class DeploymentProviderAccountGetResponse(BaseModel):
id: UUID = Field(description="Flow DB provider-account UUID (`deployment_provider_account.id`).")
provider_tenant_id: str | None = Field(
name: str = Field(description="User-chosen display name for this provider account.")
provider_key: DeploymentProviderKey = Field(description="Official provider name used by Flow.")
provider_data: dict[str, Any] | None = Field(
default=None,
description=(
"Provider-owned non-sensitive metadata for this provider account "
"(for example URL, tenant/account identifiers). Credentials are excluded."
),
)
provider_key: str = Field(description="Provider adapter key used by Flow.")
provider_url: str = Field(description="Provider service URL persisted in Flow DB.")
created_at: datetime | None = Field(default=None, description="Flow DB row creation timestamp.")
updated_at: datetime | None = Field(default=None, description="Flow DB row update timestamp.")
@@ -240,6 +220,8 @@ class _DeploymentResponseCommon(BaseModel):
"""Shared non-provider-data fields for deployment response schemas."""
id: UUID = Field(description="Flow DB deployment UUID.")
provider_id: UUID = Field(description="Flow DB provider-account UUID (`deployment_provider_account.id`).")
provider_key: str = Field(description="Provider identifier (e.g. 'watsonx-orchestrate').")
name: str
description: str | None = None
type: DeploymentType
@@ -391,169 +373,6 @@ class DeploymentStatusResponse(_DeploymentResponseWithProviderData):
"""API response for deployment status/health."""
class DeploymentRedeployResponse(_DeploymentResponseBase):
"""API response for redeployment."""
class DeploymentDuplicateResponse(_DeploymentResponseBase):
"""API response for deployment duplication."""
# ---------------------------------------------------------------------------
# Flow versions sub-resource schemas
# ---------------------------------------------------------------------------
class FlowVersionsAttach(BaseModel):
"""Flow version ids to attach to a deployment."""
model_config = {"extra": "forbid"}
ids: list[UUID] = Field(
min_length=1,
description="Flow flow version ids to attach to the deployment.",
)
@field_validator("ids")
@classmethod
def validate_ids(cls, values: list[UUID]) -> list[UUID]:
return _validate_uuid_list(values, field_name="ids")
class FlowVersionsPatch(BaseModel):
"""Add or remove flow version bindings on an existing deployment."""
model_config = {"extra": "forbid"}
add: list[UUID] | None = Field(
None,
description="Flow flow version ids to attach to the deployment. Omit to leave unchanged.",
)
remove: list[UUID] | None = Field(
None,
description="Flow flow version ids to detach from the deployment. Omit to leave unchanged.",
)
@field_validator("add", "remove")
@classmethod
def validate_id_lists(cls, values: list[UUID] | None, info: ValidationInfo) -> list[UUID] | None:
if values is None:
return None
return _validate_uuid_list(values, field_name=info.field_name)
@model_validator(mode="after")
def validate_operations(self):
add_values = self.add or []
remove_values = self.remove or []
if not add_values and not remove_values:
msg = "At least one of 'add' or 'remove' must be provided."
raise ValueError(msg)
overlap = set(add_values).intersection(remove_values)
if overlap:
ids = ", ".join(sorted(str(v) for v in overlap))
msg = f"Flow version ids cannot be present in both 'add' and 'remove': {ids}."
raise ValueError(msg)
return self
# ---------------------------------------------------------------------------
# Strict API-layer wrappers (shared-kernel boundary)
# ---------------------------------------------------------------------------
# These thin subclasses inherit field definitions from the service layer and
# add ``extra = "forbid"`` so API callers receive a 422 for unexpected fields
# instead of having data silently dropped. Subclassing (rather than
# redefining fields) keeps the API in lock-step with the service contract.
# If a service-layer field should NOT be API-visible, replace the relevant
# subclass with an API-owned model and a mapping function.
class _StrictBaseDeploymentData(BaseDeploymentData):
model_config = {"extra": "forbid"}
class _StrictBaseDeploymentDataUpdate(BaseDeploymentDataUpdate):
model_config = {"extra": "forbid"}
class _StrictDeploymentConfig(DeploymentConfig):
model_config = {"extra": "forbid"}
# ---------------------------------------------------------------------------
# Deployment config sub-resource schemas (API-owned)
# ---------------------------------------------------------------------------
class DeploymentConfigCreate(BaseModel):
"""Config input for deployment creation.
Exactly one of ``reference_id`` or ``raw_payload`` must be provided.
"""
model_config = {"extra": "forbid"}
reference_id: NonEmptyStr | None = Field(
default=None,
description="Provider-owned config reference id to bind to the deployment.",
)
raw_payload: _StrictDeploymentConfig | None = Field(
default=None,
description="Config payload to create and bind to the deployment.",
)
@model_validator(mode="after")
def validate_exactly_one(self) -> DeploymentConfigCreate:
if (self.reference_id is None) == (self.raw_payload is None):
msg = "Exactly one of 'reference_id' or 'raw_payload' must be provided."
raise ValueError(msg)
return self
class DeploymentConfigBindingUpdate(BaseModel):
"""Config binding patch for an existing deployment.
Exactly one of ``config_id``, ``raw_payload``, or ``unbind`` must be
provided:
* ``config_id`` — bind an existing config by reference.
* ``raw_payload`` — create a new config and bind it.
* ``unbind = true`` — detach the current config.
"""
model_config = {"extra": "forbid"}
config_id: NonEmptyStr | None = Field(
default=None,
description="Provider-owned config id to bind to the deployment.",
)
raw_payload: _StrictDeploymentConfig | None = Field(
default=None,
description="Config payload to create and bind to the deployment.",
)
unbind: bool = Field(
default=False,
description="Set to true to detach the current config from the deployment.",
)
@model_validator(mode="after")
def validate_config_update(self) -> DeploymentConfigBindingUpdate:
provided = sum(
[
self.config_id is not None,
self.raw_payload is not None,
self.unbind,
]
)
if provided != 1:
msg = "Exactly one of 'config_id', 'raw_payload', or 'unbind=true' must be provided."
raise ValueError(msg)
return self
# ---------------------------------------------------------------------------
# Deployment create / update request schemas
# ---------------------------------------------------------------------------
@@ -563,7 +382,13 @@ class DeploymentCreateRequest(BaseModel):
model_config = {"extra": "forbid"}
provider_id: UUID = Field(description="Flow DB provider-account UUID (`deployment_provider_account.id`).")
spec: _StrictBaseDeploymentData = Field(description="Deployment metadata (service-layer schema, no ID fields).")
name: NonEmptyStr = Field(description="Deployment display name.")
description: str = Field(
default="",
max_length=DEPLOYMENT_DESCRIPTION_MAX_LENGTH,
description="Deployment description.",
)
type: DeploymentType = Field(description="Deployment type.")
project_id: UUID | None = Field(
default=None,
description="Flow DB project id to persist the deployment under. Defaults to user's Starter Project.",
@@ -607,12 +432,22 @@ class DeploymentUpdateRequest(BaseModel):
class RunCreateRequest(BaseModel):
model_config = {"extra": "forbid"}
provider_id: UUID = Field(
description=(
"Flow DB provider-account UUID (`deployment_provider_account.id`). "
"Included alongside deployment_id to allow provider routing without an extra DB lookup."
),
provider_data: dict[str, Any] | None = Field(
default=None,
description="Provider-owned opaque run input payload.",
)
class _RunResponseBase(BaseModel):
"""Shared fields for run responses.
Only Flow-owned identifiers live at the top level. All
provider-owned data (including the provider's ``id``)
is returned inside ``provider_data`` so that ownership boundaries
stay clear and a future Flow-managed run id won't
collide with provider terminology.
"""
deployment_id: UUID = Field(description="Flow DB deployment UUID.")
provider_data: dict[str, Any] | None = Field(
default=None,
@@ -622,11 +457,6 @@ class RunCreateRequest(BaseModel):
"when the provider has assigned one."
),
)
deployment_id: UUID = Field(description="Flow DB deployment UUID.")
provider_data: dict[str, Any] | None = Field(
default=None,
description="Provider-owned opaque execution result payload.",
)
class RunCreateResponse(_RunResponseBase):
src/backend/base/flow/api/v1/variable.py
+2
View File
@@ -14,6 +14,8 @@ from flow.api.v1.models import (
get_model_names_for_provider,
get_provider_from_variable_name,
)
from flow.api.v1.schemas.deployments import DetectVarsRequest, DetectVarsResponse
from flow.services.database.models.flow_version.crud import get_flow_version_entries_by_ids
from flow.services.database.models.variable.model import VariableCreate, VariableRead, VariableUpdate
from flow.services.deps import get_variable_service
from flow.services.variable.constants import CREDENTIAL_TYPE, GENERIC_TYPE
src/backend/base/flow/base/__init__.py
+8
View File
@@ -0,0 +1,8 @@
"""Backwards compatibility module for flow.base.
This module imports from lfx.base to maintain compatibility with existing code
that expects to import from flow.base.
"""
# Import all base modules from lfx for backwards compatibility
from lfx.base import * # noqa: F403
src/backend/base/flow/base/agents/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.base.agents."""
from lfx.base.agents import * # noqa: F403
src/backend/base/flow/base/agents/crewai/__init__.py
View File
src/backend/base/flow/base/astra_assistants/__init__.py
View File
src/backend/base/flow/base/chains/__init__.py
View File
src/backend/base/flow/base/compressors/__init__.py
View File
src/backend/base/flow/base/curl/__init__.py
View File
src/backend/base/flow/base/document_transformers/__init__.py
View File
src/backend/base/flow/base/embeddings/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.base.embeddings."""
from lfx.base.embeddings import * # noqa: F403
src/backend/base/flow/base/flow_processing/__init__.py
View File
src/backend/base/flow/base/huggingface/__init__.py
View File
src/backend/base/flow/base/langchain_utilities/__init__.py
View File
src/backend/base/flow/base/mcp/__init__.py
View File
src/backend/base/flow/base/memory/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.base.memory."""
from lfx.base.memory import * # noqa: F403
src/backend/base/flow/base/prompts/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.base.prompts."""
from lfx.base.prompts import * # noqa: F403
src/backend/base/flow/base/textsplitters/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.base.textsplitters."""
from lfx.base.textsplitters import * # noqa: F403
src/backend/base/flow/base/tools/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.base.tools."""
from lfx.base.tools import * # noqa: F403
src/backend/base/flow/base/vectorstores/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.base.vectorstores."""
from lfx.base.vectorstores import * # noqa: F403
src/backend/base/flow/components/__init__.py
+21
View File
@@ -0,0 +1,21 @@
"""LangFlow Components module."""
from __future__ import annotations
from typing import Any
from lfx.components import __all__ as _lfx_all
__all__: list[str] = list(_lfx_all)
def __getattr__(attr_name: str) -> Any:
"""Forward attribute access to lfx.components."""
from lfx import components
return getattr(components, attr_name)
def __dir__() -> list[str]:
"""Forward dir() to lfx.components."""
return list(__all__)
src/backend/base/flow/components/chains/__init__.py
View File
src/backend/base/flow/components/documentloaders/__init__.py
View File
src/backend/base/flow/components/link_extractors/__init__.py
View File
src/backend/base/flow/components/notdiamond/__init__.py
View File
src/backend/base/flow/components/output_parsers/__init__.py
View File
src/backend/base/flow/components/textsplitters/__init__.py
View File
src/backend/base/flow/components/toolkits/__init__.py
View File
src/backend/base/flow/components/vectara/__init__.py
View File
src/backend/base/flow/custom/__init__.py
+26 -3
View File
@@ -1,4 +1,27 @@
from flow.custom.custom_component.component import Component
from flow.custom.custom_component.custom_component import CustomComponent
from lfx import custom as custom
from lfx.custom import custom_component as custom_component
from lfx.custom import utils as utils
from lfx.custom.custom_component.component import Component, get_component_toolkit
from lfx.custom.custom_component.custom_component import CustomComponent
__all__ = ["Component", "CustomComponent"]
# Import commonly used functions
from lfx.custom.utils import build_custom_component_template
from lfx.custom.validate import create_class, create_function, extract_class_name, extract_function_name
# Import the validate module
from . import validate
__all__ = [
"Component",
"CustomComponent",
"build_custom_component_template",
"create_class",
"create_function",
"custom",
"custom_component",
"extract_class_name",
"extract_function_name",
"get_component_toolkit",
"utils",
"validate",
]
src/backend/base/flow/custom/custom_component/__init__.py
+4
View File
@@ -0,0 +1,4 @@
from lfx.custom.custom_component import component, custom_component
from lfx.custom.custom_component.component import Component
__all__ = ["Component", "component", "custom_component"]
src/backend/base/flow/graph/__init__.py
+4 -4
View File
@@ -1,6 +1,6 @@
from flow.graph.edge.base import Edge
from flow.graph.graph.base import Graph
from flow.graph.vertex.base import Vertex
from flow.graph.vertex.vertex_types import CustomComponentVertex, InterfaceVertex, StateVertex
from lfx.graph.edge.base import Edge
from lfx.graph.graph.base import Graph
from lfx.graph.vertex.base import Vertex
from lfx.graph.vertex.vertex_types import CustomComponentVertex, InterfaceVertex, StateVertex
__all__ = ["CustomComponentVertex", "Edge", "Graph", "InterfaceVertex", "StateVertex", "Vertex"]
src/backend/base/flow/graph/edge/__init__.py
View File
src/backend/base/flow/graph/edge/utils.py
View File
src/backend/base/flow/graph/graph/__init__.py
View File
src/backend/base/flow/graph/state/__init__.py
View File
src/backend/base/flow/graph/vertex/__init__.py
View File
src/backend/base/flow/graph/vertex/constants.py
View File
src/backend/base/flow/interface/__init__.py
+4
View File
@@ -0,0 +1,4 @@
"""Interface modules for flow - imports from lfx."""
# Import everything from lfx.interface
from lfx.interface import * # noqa: F403
src/backend/base/flow/legacy_custom/__init__.py
View File
src/backend/base/flow/services/cache/factory.py
Vendored
-1
View File
@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING
from lfx.log.logger import logger
from typing_extensions import override
from flow.services.cache.disk import AsyncDiskCache
from flow.services.cache.service import AsyncInMemoryCache, CacheService, RedisCache, ThreadingInMemoryCache
from flow.services.factory import ServiceFactory
src/backend/base/flow/services/database/models/deployment/crud.py
+5
View File
@@ -9,6 +9,11 @@ from sqlalchemy.exc import IntegrityError
from sqlmodel import col, delete, func, select
from flow.services.database.models.deployment.model import Deployment
from flow.services.database.models.deployment.orm_guards import ensure_deployment_immutable_fields
from flow.services.database.models.flow_version.model import FlowVersion
from flow.services.database.models.flow_version_deployment_attachment.model import (
FlowVersionDeploymentAttachment,
)
from flow.services.database.utils import parse_uuid
if TYPE_CHECKING:
src/backend/base/flow/services/database/models/deployment_provider_account/crud.py
+2
View File
@@ -9,7 +9,9 @@ from sqlalchemy.exc import IntegrityError
from sqlmodel import col, select
from flow.services.auth import utils as auth_utils
from flow.services.database.models.deployment.orm_guards import ensure_provider_account_identity_immutable
from flow.services.database.models.deployment_provider_account.model import DeploymentProviderAccount
from flow.services.database.models.deployment_provider_account.schemas import DeploymentProviderKey
from flow.services.database.utils import normalize_string_or_none, parse_uuid
if TYPE_CHECKING:
src/backend/base/flow/services/database/models/deployment_provider_account/model.py
+5 -2
View File
@@ -10,8 +10,11 @@ from sqlmodel import Column, DateTime, Field, Relationship, SQLModel, func
from typing_extensions import Self
from flow.schema.serialize import UUIDstr
from .schemas import DeploymentProviderKey
from flow.services.database.models.deployment_provider_account.schemas import DeploymentProviderKey
from flow.services.database.models.deployment_provider_account.utils import (
validate_provider_url,
validate_tenant_url_consistency,
)
from flow.services.database.utils import (
normalize_string_or_none,
validate_non_empty_string,
src/backend/base/flow/services/database/models/flow_version/crud.py
+4 -5
View File
@@ -6,11 +6,7 @@ from lfx.log import logger
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlmodel import col, delete, func, select
if TYPE_CHECKING:
from uuid import UUID
from sqlmodel.ext.asyncio.session import AsyncSession
from flow.services.database.models.deployment.model import Deployment
from flow.services.database.models.flow_version.exceptions import (
FlowVersionConflictError,
FlowVersionDeployedError,
@@ -19,6 +15,9 @@ from flow.services.database.models.flow_version.exceptions import (
from flow.services.database.models.flow_version.model import (
FlowVersion,
)
from flow.services.database.models.flow_version_deployment_attachment.model import (
FlowVersionDeploymentAttachment,
)
from flow.services.deps import get_settings_service
if TYPE_CHECKING:
src/backend/base/flow/services/storage/__init__.py
+2 -4
View File
@@ -1,7 +1,5 @@
"""Storage services for flow."""
from .file_storage import FileStorageService
from .local import LocalStorageService
from .s3 import S3StorageService
from .service import StorageService
__all__ = ["FileStorageService", "LocalStorageService", "StorageService"]
__all__ = ["LocalStorageService", "S3StorageService", "StorageService"]
src/backend/base/flow/services/storage/file_storage.py
-146
View File
@@ -1,146 +0,0 @@
"""File-based storage service for flow artifacts."""
from __future__ import annotations
from typing import TYPE_CHECKING
import anyio
from aiofile import async_open
from loguru import logger
from .service import StorageService
if TYPE_CHECKING:
from flow.services.session.service import SessionService
from flow.services.settings.service import SettingsService
class FileStorageService(StorageService):
"""A service class for handling file-based storage operations.
Stores files under data_dir/flow_id/file_name using async I/O.
"""
def __init__(
self, session_service: SessionService, settings_service: SettingsService
) -> None:
"""Initialize the file storage service.
Args:
session_service: Service for managing sessions.
settings_service: Service for accessing configuration settings.
"""
super().__init__(session_service, settings_service)
self.set_ready()
def build_full_path(self, flow_id: str, file_name: str) -> str:
"""Build the full path of a file in the storage.
Args:
flow_id: The identifier for the flow.
file_name: The name of the file.
Returns:
The full path as a string.
"""
return str(self.data_dir / flow_id / file_name)
async def save_file(self, flow_id: str, file_name: str, data: bytes) -> None:
"""Save a file to the storage.
Args:
flow_id: The identifier for the flow.
file_name: The name of the file to be saved.
data: The byte content of the file.
Raises:
FileNotFoundError: If the parent path cannot be created.
IsADirectoryError: If the file name is a directory.
PermissionError: If there is no permission to write the file.
"""
folder_path = self.data_dir / flow_id
await folder_path.mkdir(parents=True, exist_ok=True)
file_path = folder_path / file_name
try:
async with async_open(str(file_path), "wb") as f:
await f.write(data)
logger.info(f"File {file_name} saved successfully in flow {flow_id}.")
except Exception:
logger.exception(f"Error saving file {file_name} in flow {flow_id}")
raise
async def get_file(self, flow_id: str, file_name: str) -> bytes:
"""Retrieve a file from the storage.
Args:
flow_id: The identifier for the flow.
file_name: The name of the file to be retrieved.
Returns:
The byte content of the file.
Raises:
FileNotFoundError: If the file does not exist.
"""
file_path = self.data_dir / flow_id / file_name
if not await file_path.exists():
logger.warning(f"File {file_name} not found in flow {flow_id}.")
msg = f"File {file_name} not found in flow {flow_id}"
raise FileNotFoundError(msg)
async with async_open(str(file_path), "rb") as f:
content = await f.read()
logger.debug(f"File {file_name} retrieved successfully from flow {flow_id}.")
return content
async def list_files(self, flow_id: str) -> list[str]:
"""List all files in a specified flow.
Args:
flow_id: The identifier for the flow.
Returns:
A list of file names.
Raises:
FileNotFoundError: If the flow directory does not exist.
"""
if not isinstance(flow_id, str):
flow_id = str(flow_id)
folder_path = self.data_dir / flow_id
if not await folder_path.exists() or not await folder_path.is_dir():
logger.warning(f"Flow {flow_id} directory does not exist.")
msg = f"Flow {flow_id} directory does not exist."
raise FileNotFoundError(msg)
files = [
file.name
async for file in await anyio.to_thread.run_sync(folder_path.iterdir)
if await anyio.Path(file).is_file()
]
logger.info(f"Listed {len(files)} files in flow {flow_id}.")
return files
async def delete_file(self, flow_id: str, file_name: str) -> None:
"""Delete a file from the storage.
Args:
flow_id: The identifier for the flow.
file_name: The name of the file to be deleted.
"""
file_path = self.data_dir / flow_id / file_name
if await file_path.exists():
await file_path.unlink()
logger.info(f"File {file_name} deleted successfully from flow {flow_id}.")
else:
logger.warning(
f"Attempted to delete non-existent file {file_name} in flow {flow_id}."
)
async def teardown(self) -> None:
"""Perform cleanup operations when the service is torn down."""
# No specific teardown actions required for file storage
pass
src/backend/base/flow/services/storage/local.py
+2
View File
@@ -13,6 +13,8 @@ from flow.services.storage.service import StorageService
if TYPE_CHECKING:
from collections.abc import AsyncIterator
import anyio
from flow.services.session.service import SessionService
from flow.services.settings.service import SettingsService
src/backend/base/flow/services/tracing/native.py
+10 -22
View File
@@ -20,6 +20,11 @@ from typing_extensions import override
from flow.serialization.serialization import serialize
from flow.services.database.models.traces.model import SpanStatus, SpanType
from flow.services.tracing.base import BaseTracer
from flow.services.tracing.span_sorting import (
FLOW_SPAN_NAMESPACE,
resolve_span_uuids,
topological_sort_spans,
)
if TYPE_CHECKING:
from collections.abc import Sequence
@@ -29,8 +34,6 @@ if TYPE_CHECKING:
from flow.services.tracing.schema import Log
FLOW_SPAN_NAMESPACE = UUID("a3e1c2d4-5b6f-7890-abcd-ef1234567890")
TYPE_MAP = {
"chain": SpanType.CHAIN,
"llm": SpanType.LLM,
@@ -318,26 +321,11 @@ class NativeTracer(BaseTracer):
)
await session.merge(trace)
resolved = []
for span_data in self.completed_spans:
try:
span_uuid = UUID(span_data["id"])
except (ValueError, TypeError):
# Span IDs from LangChain callbacks are strings, not UUIDs — derive
# a stable UUID so the same span always maps to the same DB row.
span_uuid = uuid5(FLOW_SPAN_NAMESPACE, f"{self.trace_id}-{span_data['id']}")
parent_uuid = None
if span_data.get("parent_span_id"):
parent_id = span_data["parent_span_id"]
if isinstance(parent_id, UUID):
parent_uuid = parent_id
else:
try:
parent_uuid = UUID(str(parent_id))
except (ValueError, TypeError):
parent_uuid = uuid5(FLOW_SPAN_NAMESPACE, f"{self.trace_id}-{parent_id}")
resolved.append((span_data, span_uuid, parent_uuid))
# Pre-compute UUIDs and topologically sort so parents are inserted
# before children — required by PostgreSQL's immediate FK enforcement
# on span.parent_span_id → span.id.
resolved = resolve_span_uuids(self.completed_spans, self.trace_id)
resolved = topological_sort_spans(resolved)
for span_data, span_uuid, parent_uuid in resolved:
span = SpanTable(
src/backend/base/flow/template/__init__.py
+2 -9
View File
@@ -1,10 +1,3 @@
from flow.template.field.base import Input, Output
from flow.template.frontend_node.base import FrontendNode
from flow.template.template.base import Template
"""Backwards compatibility module for flow.template."""
__all__ = [
"FrontendNode",
"Input",
"Output",
"Template",
]
from lfx.template import * # noqa: F403
src/backend/base/flow/template/field/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Backwards compatibility module for flow.template.field."""
from lfx.template.field import * # noqa: F403
src/backend/base/flow/template/field/prompt.py
-2
View File
@@ -1,2 +0,0 @@
# This file is for backwards compatibility
from flow.inputs.inputs import DEFAULT_PROMPT_INTUT_TYPES, DefaultPromptField # noqa: F401
src/backend/base/flow/template/frontend_node/__init__.py
-6
View File
@@ -1,6 +0,0 @@
from flow.template.frontend_node import base, custom_components
__all__ = [
"base",
"custom_components",
]
src/backend/base/flow/template/template/__init__.py
View File
src/lfx/src/lfx/cli/common.py
+8 -2
View File
@@ -107,8 +107,14 @@ def get_best_access_host(host: str) -> str:
def get_api_key() -> str:
"""Get the API key from environment variable."""
api_key = os.getenv("FLOW_API_KEY")
"""Get the API key from environment variable.
Used by ``lfx serve`` to set the superuser key on the local server.
For *remote* commands (push, pull, login, …), the per-environment key
is resolved via :func:`lfx.config.resolve_environment` and the
``api_key_env`` field in ``.lfx/environments.yaml``.
"""
api_key = os.getenv("FLOW_API_KEY") or os.getenv("LFX_API_KEY")
if not api_key:
msg = "FLOW_API_KEY environment variable is not set"
raise ValueError(msg)
src/lfx/src/lfx/components/models_and_agents/mcp_component.py
+2 -4
View File
@@ -18,7 +18,7 @@ from lfx.base.mcp.util import (
from lfx.custom.custom_component.component_with_cache import ComponentWithCache
from lfx.inputs.inputs import InputTypes # noqa: TC001
from lfx.io import BoolInput, DictInput, DropdownInput, McpInput, MessageTextInput, Output
from lfx.io.schema import flatten_schema, schema_to_flow_inputs
from lfx.io.schema import schema_to_flow_inputs
from lfx.log.logger import logger
from lfx.schema.dataframe import DataFrame
from lfx.schema.message import Message
@@ -633,9 +633,7 @@ class MCPToolsComponent(ComponentWithCache):
if not tool or not hasattr(tool, "name"):
continue
try:
flat_schema = flatten_schema(tool.args_schema.schema())
input_schema = create_input_schema_from_json_schema(flat_schema)
flow_inputs = schema_to_flow_inputs(input_schema)
flow_inputs = schema_to_flow_inputs(tool.args_schema)
inputs[tool.name] = flow_inputs
except (AttributeError, ValueError, TypeError, KeyError) as e:
msg = f"Error getting inputs for tool {getattr(tool, 'name', 'unknown')}: {e!s}"
src/lfx/src/lfx/components/vlmrun/__init__.py
+1 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any
from flow.components._importing import import_mod
from lfx.components._importing import import_mod
if TYPE_CHECKING:
from .vlmrun_transcription import VLMRunTranscription
src/lfx/src/lfx/custom/validate.py
+1 -6
View File
@@ -443,12 +443,7 @@ def prepare_global_scope(module):
exec_globals[variable_name] = sys.modules.get(variable_name, module_obj)
for node in import_froms:
module_names_to_try = [node.module]
# If original module starts with flow, also try lfx equivalent
if node.module and node.module.startswith("flow."):
lfx_module_name = node.module.replace("flow.", "lfx.", 1)
module_names_to_try.append(lfx_module_name)
module_names_to_try = _get_module_fallbacks(node.module)
success = False
last_error = None
src/lfx/src/lfx/interface/utils.py
+1 -1
View File
@@ -95,7 +95,7 @@ def setup_llm_caching() -> None:
def set_langchain_cache(settings) -> None:
from langchain.globals import set_llm_cache
from langchain_core.globals import set_llm_cache
from flow.interface.importing.utils import import_class
if cache_type := os.getenv("FLOW_LANGCHAIN_CACHE"):
src/lfx/src/lfx/load/utils.py
+2
View File
@@ -31,6 +31,8 @@ def upload(file_path: str, host: str, flow_id: str, api_key: str | None = None):
"""
try:
url = f"{host}/v1/upload/{flow_id}"
resolved_api_key = api_key if api_key is not None else os.environ.get("FLOW_API_KEY")
headers = {"x-api-key": resolved_api_key} if resolved_api_key else {}
with Path(file_path).open("rb") as file:
response = httpx.post(url, files={"file": file}, headers=headers)
if response.status_code in {httpx.codes.OK, httpx.codes.CREATED}:
src/lfx/src/lfx/memory/__init__.py
-47
View File
@@ -71,53 +71,6 @@ def get_messages(*args: Any, **kwargs: Any):
def store_message(*args: Any, **kwargs: Any):
return _impl().store_message(*args, **kwargs)
# Import the appropriate implementation
if has_flow_memory():
try:
# Import full flow implementation
from flow.memory import (
aadd_messages,
aadd_messagetables,
add_messages,
adelete_messages,
aget_messages,
astore_message,
aupdate_messages,
delete_message,
delete_messages,
get_messages,
store_message,
)
except ImportError:
# Fallback to lfx implementation if flow import fails
from lfx.memory.stubs import (
aadd_messages,
aadd_messagetables,
add_messages,
adelete_messages,
aget_messages,
astore_message,
aupdate_messages,
delete_message,
delete_messages,
get_messages,
store_message,
)
else:
# Use lfx implementation
from lfx.memory.stubs import (
aadd_messages,
aadd_messagetables,
add_messages,
adelete_messages,
aget_messages,
astore_message,
aupdate_messages,
delete_message,
delete_messages,
get_messages,
store_message,
)
__all__ = [
"aadd_messages",
src/lfx/src/lfx/services/settings/base.py
+20 -566
View File
@@ -20,23 +20,16 @@ from __future__ import annotations
import asyncio
import contextlib
import json
import os
from pathlib import Path
from shutil import copy2
from typing import TYPE_CHECKING, Any, Literal
from typing import TYPE_CHECKING, Any
import aiofiles
import orjson
import yaml
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, EnvSettingsSource, PydanticBaseSettingsSource, SettingsConfigDict
from typing_extensions import override
from lfx.constants import BASE_COMPONENTS_PATH as BASE_COMPONENTS_PATH # noqa: PLC0414 # re-export for back-compat
from lfx.serialization.constants import MAX_ITEMS_LENGTH, MAX_TEXT_LENGTH
from lfx.log.logger import logger
from lfx.services.settings.constants import AGENTIC_VARIABLES, VARIABLES_TO_GET_FROM_ENVIRONMENT
from lfx.utils.util_strings import is_valid_database_url, sanitize_database_url
from lfx.services.settings.groups import (
CacheSettings,
ComponentsSettings,
@@ -93,567 +86,28 @@ class CustomSource(EnvSettingsSource):
return super().prepare_field_value(field_name, field, value, value_is_complex)
class Settings(BaseSettings):
# Define the default FLOW_DIR
config_dir: str | None = None
# Define if flow db should be saved in config dir or
# in the flow directory
save_db_in_config_dir: bool = False
"""Define if flow database should be saved in FLOW_CONFIG_DIR or in the flow directory
(i.e. in the package directory)."""
class Settings(
VariablesSettings,
UiSettings,
ComponentsSettings,
SecuritySettings,
ObservabilitySettings,
TelemetrySettings,
StorageSettings,
CacheSettings,
McpSettings,
DatabaseSettings,
RuntimeSettings,
ServerSettings,
PathSettings,
BaseSettings,
):
"""Top-level Flow settings.
knowledge_bases_dir: str | None = "~/.flow/knowledge_bases"
"""The directory to store knowledge bases."""
dev: bool = False
"""If True, Flow will run in development mode."""
database_url: str | None = None
"""Database URL for Flow. If not provided, Flow will use a SQLite database.
The driver shall be an async one like `sqlite+aiosqlite` (`sqlite` and `postgresql`
will be automatically converted to the async drivers `sqlite+aiosqlite` and
`postgresql+psycopg` respectively)."""
database_connection_retry: bool = False
"""If True, Flow will retry to connect to the database if it fails."""
pool_size: int = 20
"""The number of connections to keep open in the connection pool.
For high load scenarios, this should be increased based on expected concurrent users."""
max_overflow: int = 30
"""The number of connections to allow that can be opened beyond the pool size.
Should be 2x the pool_size for optimal performance under load."""
db_connect_timeout: int = 30
"""The number of seconds to wait before giving up on a lock to released or establishing a connection to the
database."""
migration_lock_namespace: str | None = None
"""Optional namespace identifier for PostgreSQL advisory lock during migrations.
If not provided, a hash of the database URL will be used. Useful when multiple Flow
instances share the same database and need coordinated migration locking."""
mcp_server_timeout: int = 20
"""The number of seconds to wait before giving up on a lock to released or establishing a connection to the
database."""
# ---------------------------------------------------------------------
# MCP Session-manager tuning
# ---------------------------------------------------------------------
mcp_max_sessions_per_server: int = 10
"""Maximum number of MCP sessions to keep per unique server (command/url).
Mirrors the default constant MAX_SESSIONS_PER_SERVER in util.py. Adjust to
control resource usage or concurrency per server."""
mcp_session_idle_timeout: int = 400 # seconds
"""How long (in seconds) an MCP session can stay idle before the background
cleanup task disposes of it. Defaults to 5 minutes."""
mcp_session_cleanup_interval: int = 120 # seconds
"""Frequency (in seconds) at which the background cleanup task wakes up to
reap idle sessions."""
# sqlite configuration
sqlite_pragmas: dict | None = {"synchronous": "NORMAL", "journal_mode": "WAL", "busy_timeout": 30000}
"""SQLite pragmas to use when connecting to the database."""
db_driver_connection_settings: dict | None = None
"""Database driver connection settings."""
db_connection_settings: dict | None = {
"pool_size": 20, # Match the pool_size above
"max_overflow": 30, # Match the max_overflow above
"pool_timeout": 30, # Seconds to wait for a connection from pool
"pool_pre_ping": True, # Check connection validity before using
"pool_recycle": 1800, # Recycle connections after 30 minutes
"echo": False, # Set to True for debugging only
}
"""Database connection settings optimized for high load scenarios.
Note: These settings are most effective with PostgreSQL. For SQLite:
- Reduce pool_size and max_overflow if experiencing lock contention
- SQLite has limited concurrent write capability even with WAL mode
- Best for read-heavy or moderate write workloads
Settings:
- pool_size: Number of connections to maintain (increase for higher concurrency)
- max_overflow: Additional connections allowed beyond pool_size
- pool_timeout: Seconds to wait for an available connection
- pool_pre_ping: Validates connections before use to prevent stale connections
- pool_recycle: Seconds before connections are recycled (prevents timeouts)
- echo: Enable SQL query logging (development only)
Composed from per-group mixins. See module docstring for the inheritance
order rationale.
"""
use_noop_database: bool = False
"""If True, disables all database operations and uses a no-op session.
Controlled by FLOW_USE_NOOP_DATABASE env variable."""
# cache configuration
cache_type: Literal["async", "redis", "memory", "disk"] = "async"
"""The cache type can be 'async' or 'redis'."""
cache_expire: int = 3600
"""The cache expire in seconds."""
variable_store: str = "db"
"""The store can be 'db' or 'kubernetes'."""
prometheus_enabled: bool = False
"""If set to True, Flow will expose Prometheus metrics."""
prometheus_port: int = 9090
"""The port on which Flow will expose Prometheus metrics. 9090 is the default port."""
disable_track_apikey_usage: bool = False
remove_api_keys: bool = False
components_path: list[str] = []
components_index_path: str | None = None
"""Path or URL to a prebuilt component index JSON file.
If None, uses the built-in index at lfx/_assets/component_index.json.
Set to a file path (e.g., '/path/to/index.json') or URL (e.g., 'https://example.com/index.json')
to use a custom index.
"""
langchain_cache: str = "InMemoryCache"
load_flows_path: str | None = None
bundle_urls: list[str] = []
# Redis
redis_host: str = "localhost"
redis_port: int = 6379
redis_db: int = 0
redis_url: str | None = None
redis_cache_expire: int = 3600
# Sentry
sentry_dsn: str | None = None
sentry_traces_sample_rate: float | None = 1.0
sentry_profiles_sample_rate: float | None = 1.0
store: bool | None = True
store_url: str | None = "https://api.flow.store"
download_webhook_url: str | None = "https://api.flow.store/flows/trigger/ec611a61-8460-4438-b187-a4f65e5559d4"
like_webhook_url: str | None = "https://api.flow.store/flows/trigger/64275852-ec00-45c1-984e-3bff814732da"
storage_type: str = "local"
"""Storage type for file storage. Defaults to 'local'. Supports 'local' and 's3'."""
object_storage_bucket_name: str | None = "flow-bucket"
"""Object storage bucket name for file storage. Defaults to 'flow-bucket'."""
object_storage_prefix: str | None = "files"
"""Object storage prefix for file storage. Defaults to 'files'."""
object_storage_tags: dict[str, str] | None = None
"""Object storage tags for file storage."""
celery_enabled: bool = False
fallback_to_env_var: bool = True
"""If set to True, Global Variables set in the UI will fallback to a environment variable
with the same name in case Flow fails to retrieve the variable value."""
store_environment_variables: bool = True
"""Whether to store environment variables as Global Variables in the database."""
variables_to_get_from_environment: list[str] = VARIABLES_TO_GET_FROM_ENVIRONMENT
"""List of environment variables to get from the environment and store in the database."""
worker_timeout: int = 300
"""Timeout for the API calls in seconds."""
frontend_timeout: int = 0
"""Timeout for the frontend API calls in seconds."""
user_agent: str = "flow"
"""User agent for the API calls."""
backend_only: bool = False
"""If set to True, Flow will not serve the frontend."""
# CORS Settings
cors_origins: list[str] | str = "*"
"""Allowed origins for CORS. Can be a list of origins or '*' for all origins.
Default is '*' for backward compatibility. In production, specify exact origins."""
cors_allow_credentials: bool = True
"""Whether to allow credentials in CORS requests.
Default is True for backward compatibility. In v2.0, this will be changed to False when using wildcard origins."""
cors_allow_methods: list[str] | str = "*"
"""Allowed HTTP methods for CORS requests."""
cors_allow_headers: list[str] | str = "*"
"""Allowed headers for CORS requests."""
# Telemetry
do_not_track: bool = False
"""If set to True, Flow will not track telemetry."""
telemetry_base_url: str = "https://flow.gateway.scarf.sh"
transactions_storage_enabled: bool = True
"""If set to True, Flow will track transactions between flows."""
vertex_builds_storage_enabled: bool = True
"""If set to True, Flow will keep track of each vertex builds (outputs) in the UI for any flow."""
# Config
allow_custom_components: bool = True
"""If False, blocks execution of components whose code does not match a known server template
(see SecuritySettings.allow_custom_components — this is the flat field the runtime reads)."""
host: str = "localhost"
"""The host on which Flow will run."""
port: int = 7860
"""The port on which Flow will run."""
root_path: str = ""
"""ASGI root_path when Flow is served behind a prefix-stripping reverse proxy (empty = served at /)."""
runtime_port: int | None = Field(default=None, exclude=True)
"""TEMPORARY: The port detected at runtime after checking for conflicts.
This field is system-managed only and will be removed in future versions
when strict port enforcement is implemented (errors will be raised if port unavailable)."""
workers: int = 1
"""The number of workers to run."""
log_level: str = "critical"
"""The log level for Flow."""
log_file: str | None = "logs/flow.log"
"""The path to log file for Flow."""
alembic_log_file: str = "alembic/alembic.log"
"""The path to log file for Alembic for SQLAlchemy."""
alembic_log_to_stdout: bool = False
"""If set to True, the log file will be ignored and Alembic will log to stdout."""
frontend_path: str | None = None
"""The path to the frontend directory containing build files. This is for development purposes only.."""
open_browser: bool = False
"""If set to True, Flow will open the browser on startup."""
auto_saving: bool = True
"""If set to True, Flow will auto save flows."""
auto_saving_interval: int = 1000
"""The interval in ms at which Flow will auto save flows."""
health_check_max_retries: int = 5
"""The maximum number of retries for the health check."""
max_file_size_upload: int = 1024
"""The maximum file size for the upload in MB."""
deactivate_tracing: bool = False
"""If set to True, tracing will be deactivated."""
max_transactions_to_keep: int = 3000
"""The maximum number of transactions to keep in the database."""
max_vertex_builds_to_keep: int = 3000
"""The maximum number of vertex builds to keep in the database."""
max_vertex_builds_per_vertex: int = 50
"""The maximum number of builds to keep per vertex. Older builds will be deleted."""
max_flow_version_entries_per_flow: int = 50
"""Max version history entries per flow. Oldest entries pruned on next snapshot.
If retroactively lowered below the current count for a flow,
the oldest entries are deleted only when the next entry is created.
"""
webhook_polling_interval: int = 0
"""The polling interval for the webhook in ms. Set to 0 to disable (SSE provides real-time updates)."""
fs_flows_polling_interval: int = 10000
"""The polling interval in milliseconds for synchronizing flows from the file system."""
ssl_cert_file: str | None = None
"""Path to the SSL certificate file on the local system."""
ssl_key_file: str | None = None
"""Path to the SSL key file on the local system."""
max_text_length: int = MAX_TEXT_LENGTH
"""Maximum number of characters to store and display in the UI. Responses longer than this
will be truncated when displayed in the UI. Does not truncate responses between components nor outputs."""
max_items_length: int = MAX_ITEMS_LENGTH
"""Maximum number of items to store and display in the UI. Lists longer than this
will be truncated when displayed in the UI. Does not affect data passed between components nor outputs."""
# MCP Server
mcp_server_enabled: bool = True
"""If set to False, Flow will not enable the MCP server."""
mcp_server_enable_progress_notifications: bool = False
"""If set to False, Flow will not send progress notifications in the MCP server."""
# Add projects to MCP servers automatically on creation
add_projects_to_mcp_servers: bool = True
"""If set to True, newly created projects will be added to the user's MCP servers config automatically."""
# MCP Composer
mcp_composer_enabled: bool = True
"""If set to False, Flow will not start the MCP Composer service."""
mcp_composer_version: str = "==0.1.0.8.10"
"""Version constraint for mcp-composer when using uvx. Uses PEP 440 syntax."""
# Agentic Experience
agentic_experience: bool = False
"""If set to True, Flow will start the agentic MCP server that provides tools for
flow/component operations, template search, and graph visualization."""
# Developer API
developer_api_enabled: bool = False
"""If set to True, Flow will enable developer API endpoints for advanced debugging and introspection."""
# Public Flow Settings
public_flow_cleanup_interval: int = Field(default=3600, gt=600)
"""The interval in seconds at which public temporary flows will be cleaned up.
Default is 1 hour (3600 seconds). Minimum is 600 seconds (10 minutes)."""
public_flow_expiration: int = Field(default=86400, gt=600)
"""The time in seconds after which a public temporary flow will be considered expired and eligible for cleanup.
Default is 24 hours (86400 seconds). Minimum is 600 seconds (10 minutes)."""
event_delivery: Literal["polling", "streaming", "direct"] = "streaming"
"""How to deliver build events to the frontend. Can be 'polling', 'streaming' or 'direct'."""
lazy_load_components: bool = False
"""If set to True, Flow will only partially load components at startup and fully load them on demand.
This significantly reduces startup time but may cause a slight delay when a component is first used."""
# Starter Projects
create_starter_projects: bool = True
"""If set to True, Flow will create starter projects. If False, skips all starter project setup.
Note that this doesn't check if the starter projects are already loaded in the db;
this is intended to be used to skip all startup project logic."""
update_starter_projects: bool = True
"""If set to True, Flow will update starter projects."""
# SSRF Protection
ssrf_protection_enabled: bool = False
"""If set to True, Flow will enable SSRF (Server-Side Request Forgery) protection.
When enabled, blocks requests to private IP ranges, localhost, and cloud metadata endpoints.
When False (default), no URL validation is performed, allowing requests to any destination
including internal services, private networks, and cloud metadata endpoints.
Default is False for backward compatibility. In v2.0, this will be changed to True.
Note: When ssrf_protection_enabled is disabled, the ssrf_allowed_hosts setting is ignored and has no effect."""
ssrf_allowed_hosts: list[str] = []
"""Comma-separated list of hosts/IPs/CIDR ranges to allow despite SSRF protection.
Examples: 'internal-api.company.local,192.168.1.0/24,10.0.0.5,*.dev.internal'
Supports exact hostnames, wildcard domains (*.example.com), exact IPs, and CIDR ranges.
Note: This setting only takes effect when ssrf_protection_enabled is True.
When protection is disabled, all hosts are allowed regardless of this setting."""
@field_validator("runtime_port", mode="before")
@classmethod
def validate_runtime_port(cls, value):
"""Parse port from Kubernetes service discovery env vars.
Kubernetes auto-creates env vars like FLOW_RUNTIME_PORT=tcp://<ip>:<port>
for services, which collides with the FLOW_ env prefix. Extract the port
number from URL-like values instead of failing.
"""
if value is None:
return None
if isinstance(value, int):
return value
if isinstance(value, str):
if value.isdigit():
return int(value)
if "://" in value:
from urllib.parse import urlparse
try:
parsed_port = urlparse(value).port
except ValueError:
return None
if parsed_port is not None:
return parsed_port
return None
@field_validator("cors_origins", mode="before")
@classmethod
def validate_cors_origins(cls, value):
"""Convert comma-separated string to list if needed."""
if isinstance(value, str) and value != "*":
if "," in value:
# Convert comma-separated string to list
return [origin.strip() for origin in value.split(",")]
# Convert single origin to list for consistency
return [value]
return value
@field_validator("use_noop_database", mode="before")
@classmethod
def set_use_noop_database(cls, value):
if value:
logger.info("Running with NOOP database session. All DB operations are disabled.")
return value
@field_validator("event_delivery", mode="before")
@classmethod
def set_event_delivery(cls, value, info):
# If workers > 1, we need to use direct delivery
# because polling and streaming are not supported
# in multi-worker environments
if info.data.get("workers", 1) > 1:
logger.warning("Multi-worker environment detected, using direct event delivery")
return "direct"
return value
@field_validator("user_agent", mode="after")
@classmethod
def set_user_agent(cls, value):
if not value:
value = "Flow"
import os
os.environ["USER_AGENT"] = value
logger.debug(f"Setting user agent to {value}")
return value
@field_validator("mcp_composer_version", mode="before")
@classmethod
def validate_mcp_composer_version(cls, value):
"""Ensure the version string has a version specifier prefix.
If a bare version like '0.1.0.7' is provided, prepend '~=' to allow patch updates.
Supports PEP 440 specifiers: ==, !=, <=, >=, <, >, ~=, ===
"""
if not value:
return "==0.1.0.8.10" # Default
# Check if it already has a version specifier
# Order matters: check longer specifiers first to avoid false matches
specifiers = ["===", "==", "!=", "<=", ">=", "~=", "<", ">"]
if any(value.startswith(spec) for spec in specifiers):
return value
# If it's a bare version number, add ~= prefix
# This regex matches version numbers like 0.1.0.7, 1.2.3, etc.
import re
if re.match(r"^\d+(\.\d+)*", value):
logger.debug(f"Adding ~= prefix to bare version '{value}' -> '~={value}'")
return f"~={value}"
# If we can't determine, return as-is and let uvx handle it
return value
@field_validator("variables_to_get_from_environment", mode="before")
@classmethod
def set_variables_to_get_from_environment(cls, value):
import os
if isinstance(value, str):
value = value.split(",")
result = list(set(VARIABLES_TO_GET_FROM_ENVIRONMENT + value))
# Add agentic variables if agentic_experience is enabled
# Check env var directly since we can't access instance attributes in validator
if os.getenv("FLOW_AGENTIC_EXPERIENCE", "true").lower() == "true":
result.extend(AGENTIC_VARIABLES)
return list(set(result))
@field_validator("log_file", mode="before")
@classmethod
def set_log_file(cls, value):
if isinstance(value, Path):
value = str(value)
return value
@field_validator("config_dir", mode="before")
@classmethod
def set_flow_dir(cls, value):
if not value:
from platformdirs import user_cache_dir
# Define the app name and author
app_name = "flow"
app_author = "flow"
# Get the cache directory for the application
cache_dir = user_cache_dir(app_name, app_author)
# Create a .flow directory inside the cache directory
value = Path(cache_dir)
value.mkdir(parents=True, exist_ok=True)
if isinstance(value, str):
value = Path(value)
# Resolve to absolute path to handle relative paths correctly
value = value.resolve()
if not value.exists():
value.mkdir(parents=True, exist_ok=True)
return str(value)
@field_validator("database_url", mode="before")
@classmethod
def set_database_url(cls, value, info):
if value and not is_valid_database_url(value):
sanitized = sanitize_database_url(value)
msg = f"Invalid database_url provided: '{sanitized}'"
raise ValueError(msg)
if flow_database_url := os.getenv("FLOW_DATABASE_URL"):
value = flow_database_url
logger.debug("Using FLOW_DATABASE_URL env variable")
else:
# Originally, we used sqlite:///./flow.db
# so we need to migrate to the new format
# if there is a database in that location
if not info.data["config_dir"]:
msg = "config_dir not set, please set it or provide a database_url"
raise ValueError(msg)
from lfx.utils.version import get_version_info
from lfx.utils.version import is_pre_release as flow_is_pre_release
version = get_version_info()["version"]
is_pre_release = flow_is_pre_release(version)
if info.data["save_db_in_config_dir"]:
database_dir = info.data["config_dir"]
else:
# Use flow package path, not lfx, for backwards compatibility
try:
import flow
database_dir = Path(flow.__file__).parent.resolve()
except ImportError:
database_dir = Path(__file__).parent.parent.parent.resolve()
pre_db_file_name = "flow-pre.db"
db_file_name = "flow.db"
new_pre_path = f"{database_dir}/{pre_db_file_name}"
new_path = f"{database_dir}/{db_file_name}"
final_path = None
if is_pre_release:
if Path(new_pre_path).exists():
final_path = new_pre_path
elif Path(new_path).exists() and info.data["save_db_in_config_dir"]:
# We need to copy the current db to the new location
logger.debug("Copying existing database to new location")
copy2(new_path, new_pre_path)
logger.debug(f"Copied existing database to {new_pre_path}")
elif Path(f"./{db_file_name}").exists() and info.data["save_db_in_config_dir"]:
logger.debug("Copying existing database to new location")
copy2(f"./{db_file_name}", new_pre_path)
logger.debug(f"Copied existing database to {new_pre_path}")
else:
logger.debug(f"Creating new database at {new_pre_path}")
final_path = new_pre_path
elif Path(new_path).exists():
final_path = new_path
elif Path(f"./{db_file_name}").exists():
try:
logger.debug("Copying existing database to new location")
copy2(f"./{db_file_name}", new_path)
logger.debug(f"Copied existing database to {new_path}")
except OSError:
logger.exception("Failed to copy database, using default path")
new_path = f"./{db_file_name}"
else:
final_path = new_path
if final_path is None:
final_path = new_pre_path if is_pre_release else new_path
value = f"sqlite:///{final_path}"
return value
@field_validator("components_path", mode="before")
@classmethod
def set_components_path(cls, value):
"""Processes and updates the components path list, incorporating environment variable overrides.
If the `FLOW_COMPONENTS_PATH` environment variable is set and points to an existing path, it is
appended to the provided list if not already present. If the input list is empty or missing, it is
set to an empty list.
"""
if os.getenv("FLOW_COMPONENTS_PATH"):
logger.debug("Adding FLOW_COMPONENTS_PATH to components_path")
flow_component_path = os.getenv("FLOW_COMPONENTS_PATH")
if Path(flow_component_path).exists() and flow_component_path not in value:
if isinstance(flow_component_path, list):
for path in flow_component_path:
if path not in value:
value.append(path)
logger.debug(f"Extending {flow_component_path} to components_path")
elif flow_component_path not in value:
value.append(flow_component_path)
logger.debug(f"Appending {flow_component_path} to components_path")
if not value:
value = [BASE_COMPONENTS_PATH]
elif isinstance(value, Path):
value = [str(value)]
elif isinstance(value, list):
value = [str(p) if isinstance(p, Path) else p for p in value]
return value
model_config = SettingsConfigDict(validate_assignment=True, extra="ignore", env_prefix="FLOW_")
async def update_from_yaml(self, file_path: str, *, dev: bool = False) -> None:
src/lfx/src/lfx/services/storage/local.py
+1
View File
@@ -11,6 +11,7 @@ from lfx.services.base import Service
from lfx.services.storage.service import StorageService
if TYPE_CHECKING:
import anyio
from flow.services.session.service import SessionService
from lfx.services.settings.service import SettingsService
src/lfx/src/lfx/utils/concurrency.py
+1 -1
View File
@@ -34,7 +34,7 @@ class KeyedWorkerLockManager:
"""A manager for acquiring locks between workers based on a key."""
def __init__(self) -> None:
self.locks_dir = Path(user_cache_dir("flow"), ensure_exists=True) / "worker_locks"
self.locks_dir = Path(user_cache_dir("flow", ensure_exists=True)) / "worker_locks"
@staticmethod
def _validate_key(key: str) -> bool:
src/lfx/src/lfx/utils/flow_utils.py
+19
View File
@@ -46,3 +46,22 @@ def has_flow_memory():
_FlowModule.set_available(is_flow_available)
return is_flow_available
def has_flow_db_backend() -> bool:
"""Return True iff flow-backed memory calls have a real DB to hit.
Requires both flow to be importable AND the registered database service to be
a non-noop implementation. Evaluated on every call because the database
service is typically registered *after* this module is first imported (e.g.,
from Component class definitions loaded before graph setup).
"""
if not has_flow_memory():
return False
from lfx.services.database.service import NoopDatabaseService
from lfx.services.deps import get_db_service
try:
return not isinstance(get_db_service(), NoopDatabaseService)
except Exception: # noqa: BLE001
return False
src/lfx/src/lfx/utils/ssrf_protection.py
+1 -11
View File
@@ -13,15 +13,9 @@ IMPORTANT: HTTP Redirects
See: https://cheatsheetseries.owasp.org/cheatsheets/Server_Side_Request_Forgery_Prevention_Cheat_Sheet.html
Configuration:
FLOW_SSRF_PROTECTION_ENABLED: Enable/disable SSRF protection (default: false)
TODO: Change default to true in next major version (2.0)
FLOW_SSRF_PROTECTION_ENABLED: Enable/disable SSRF protection (default: true)
FLOW_SSRF_ALLOWED_HOSTS: Comma-separated list of allowed hosts/CIDR ranges
Examples: "192.168.1.0/24,internal-api.company.local,10.0.0.5"
TODO: In next major version (2.0):
- Change FLOW_SSRF_PROTECTION_ENABLED default to "true"
- Remove warning-only mode and enforce blocking
- Update documentation to reflect breaking change
"""
import functools
@@ -293,7 +287,6 @@ def _validate_direct_ip_address(hostname: str) -> bool:
if is_ip_blocked(ip_obj):
msg = (
f"Access to IP address {hostname} is blocked by SSRF protection. "
"Requests to private/internal IP ranges are not allowed for security reasons. "
"To allow this IP, add it to FLOW_SSRF_ALLOWED_HOSTS environment variable."
)
raise SSRFProtectionError(msg)
@@ -335,9 +328,6 @@ def _validate_hostname_resolution(hostname: str) -> None:
if blocked_ips:
msg = (
f"Hostname {hostname} resolves to blocked IP address(es): {', '.join(blocked_ips)}. "
"Requests to private/internal IP ranges are not allowed for security reasons. "
"This protection prevents access to internal services, cloud metadata endpoints "
"(e.g., AWS 169.254.169.254), and other sensitive internal resources. "
"To allow this hostname, add it to FLOW_SSRF_ALLOWED_HOSTS environment variable."
)
raise SSRFProtectionError(msg)