CVE-2026-6598

@@ -29,6 +29,7 @@ RUN apt-get update \
 # Copy only backend source (excludes frontend)
 COPY ./src/backend ./src/backend
 COPY ./src/lfx ./src/lfx
+COPY ./src/sdk ./src/sdk
 
 # Create venv and install langflow-base with dependencies
 # Using uv pip instead of uv sync to avoid workspace complexities
@@ -37,7 +38,7 @@ ENV PATH="/app/.venv/bin:$PATH"
 ENV VIRTUAL_ENV="/app/.venv"
 
 RUN --mount=type=cache,target=/root/.cache/uv \
-    uv pip install ./src/lfx "./src/backend/base[complete,postgresql]"
+    uv pip install ./src/sdk ./src/lfx "./src/backend/base[complete,postgresql]"
 
 ################################
 # RUNTIME

@@ -48,6 +48,8 @@ COPY ./src/backend/base/uv.lock /app/src/backend/base/uv.lock
 COPY ./src/backend/base/pyproject.toml /app/src/backend/base/pyproject.toml
 COPY ./src/lfx/README.md /app/src/lfx/README.md
 COPY ./src/lfx/pyproject.toml /app/src/lfx/pyproject.toml
+COPY ./src/sdk/README.md /app/src/sdk/README.md
+COPY ./src/sdk/pyproject.toml /app/src/sdk/pyproject.toml
 
 RUN --mount=type=cache,target=/root/.cache/uv \
     RUSTFLAGS='--cfg reqwest_unstable' \

@@ -69,6 +69,7 @@ explicit = true
 langflow-base = { workspace = true }
 langflow = { workspace = true }
 lfx = { workspace = true }
+langflow-sdk = { workspace = true }
 torch = { index = "pytorch-cpu" }
 torchvision = { index = "pytorch-cpu" }
 
@@ -77,6 +78,7 @@ members = [
     "src/backend/base",
     ".",
     "src/lfx",
+    "src/sdk",
 ]
 
 [tool.hatch.build.targets.wheel]
@@ -313,6 +315,46 @@ external = ["RUF027"]
 "src/lfx/src/lfx/inputs/input_mixin.py" = [
     "S105",  # False positive: PASSWORD is a type constant
 ]
+"src/lfx/src/lfx/__main__.py" = [
+    "B008",   # Typer CLI requires function calls in argument defaults
+    "FBT001", # Bool flags are the standard typer pattern
+    "FBT003", # Boolean positional values in typer.Option() calls
+]
+"src/lfx/src/lfx/cli/_setup_commands.py" = [
+    "B008",   # Typer CLI requires function calls in argument defaults
+    "FBT001", # Bool flags are the standard typer pattern
+    "FBT003", # Boolean positional values in typer.Option() calls
+]
+"src/lfx/src/lfx/cli/_authoring_commands.py" = [
+    "B008",   # Typer CLI requires function calls in argument defaults
+    "FBT001", # Bool flags are the standard typer pattern
+    "FBT003", # Boolean positional values in typer.Option() calls
+]
+"src/lfx/src/lfx/cli/_running_commands.py" = [
+    "B008",   # Typer CLI requires function calls in argument defaults
+    "FBT001", # Bool flags are the standard typer pattern
+    "FBT003", # Boolean positional values in typer.Option() calls
+]
+"src/lfx/src/lfx/cli/_remote_commands.py" = [
+    "B008",   # Typer CLI requires function calls in argument defaults
+    "FBT001", # Bool flags are the standard typer pattern
+    "FBT003", # Boolean positional values in typer.Option() calls
+]
+"src/backend/base/langflow/api/utils/*" = [
+    "TCH",  # Imports are used at runtime (FastAPI deps, SQLAlchemy queries, model constructors)
+]
+"src/sdk/src/langflow_sdk/_http.py" = [
+    "TCH",  # httpx is used at runtime (response object methods)
+]
+"src/sdk/src/langflow_sdk/environments.py" = [
+    "TRY003", # Contextual error messages are necessary here
+    "EM102",  # f-string messages assigned before raise where possible
+]
+"src/sdk/tests/*" = [
+    "S101",   # assert is the standard pytest assertion style
+    "INP001", # Not a package namespace issue in tests
+    "TC003",  # pytest fixture type hints are evaluated at runtime
+]
 "src/lfx/src/lfx/schema/table.py" = [
     "S105",  # False positive: PASSWORD is a formatter type
 ]

@@ -1643,7 +1643,7 @@
         "filename": "src/backend/base/langflow/api/utils/core.py",
         "hashed_secret": "e5e9fa1ba31ecd1ae84f75caaa474f3a663f05f4",
         "is_verified": false,
-        "line_number": 440,
+        "line_number": 413,
         "is_secret": false
       }
     ],
@@ -7147,16 +7147,6 @@
         "is_secret": false
       }
     ],
-    "src/lfx/tests/unit/cli/test_serve_simple.py": [
-      {
-        "type": "Secret Keyword",
-        "filename": "src/lfx/tests/unit/cli/test_serve_simple.py",
-        "hashed_secret": "3acfb2c2b433c0ea7ff107e33df91b18e52f960f",
-        "is_verified": false,
-        "line_number": 67,
-        "is_secret": false
-      }
-    ],
     "src/lfx/tests/unit/components/langchain_utilities/test_csv_agent.py": [
       {
         "type": "Secret Keyword",

@@ -1,32 +1,24 @@
 from __future__ import annotations
 
-import uuid
-from ast import literal_eval
+import json as _json
 from datetime import timedelta
 from enum import Enum
 from typing import TYPE_CHECKING, Annotated, Any
 
 from fastapi import Depends, HTTPException, Path, Query
 from fastapi_pagination import Params
-from lfx.graph.graph.base import Graph
 from lfx.log.logger import logger
-from lfx.services.deps import injectable_session_scope, injectable_session_scope_readonly, session_scope
+from lfx.services.deps import injectable_session_scope, injectable_session_scope_readonly
 from lfx.utils.validate_cloud import raise_error_if_astra_cloud_disable_component
-from sqlalchemy import delete
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from langflow.services.auth.utils import get_current_active_user, get_current_active_user_mcp
 from langflow.services.database.models.flow.model import Flow
-from langflow.services.database.models.flow_version.model import FlowVersion
-from langflow.services.database.models.message.model import MessageTable
-from langflow.services.database.models.transactions.model import TransactionTable
 from langflow.services.database.models.user.model import User
-from langflow.services.database.models.vertex_builds.model import VertexBuildTable
 from langflow.services.store.utils import get_lf_version_from_pypi
 from langflow.utils.constants import LANGFLOW_GLOBAL_VAR_HEADER_PREFIX
 
 if TYPE_CHECKING:
-    from langflow.services.chat.service import ChatService
     from langflow.services.store.schema import StoreComponentCreate
 
 
@@ -43,29 +35,25 @@
 DbSessionReadOnly = Annotated[AsyncSession, Depends(injectable_session_scope_readonly)]
 
 
-def _get_validated_file_name(file_name: str = Path()) -> str:
-    """Validate file_name path parameter to prevent path traversal attacks."""
-    if ".." in file_name or "/" in file_name or "\\" in file_name:
+def _get_validated_path_segment(value: str, *, label: str = "name") -> str:
+    """Validate a path segment to prevent path traversal attacks."""
+    if ".." in value or "/" in value or "\\" in value:
         raise HTTPException(
             status_code=400,
-            detail="Invalid file name. Use a simple file name without directory paths or '..'.",
+            detail=f"Invalid {label}. Use a simple {label} without directory paths or '..'.",
         )
-    return file_name
+    return value
 
 
-ValidatedFileName = Annotated[str, Depends(_get_validated_file_name)]
+def _get_validated_file_name(file_name: str = Path()) -> str:
+    return _get_validated_path_segment(file_name, label="file name")
 
 
 def _get_validated_folder_name(folder_name: str = Path()) -> str:
-    """Validate folder_name path parameter to prevent path traversal attacks."""
-    if ".." in folder_name or "/" in folder_name or "\\" in folder_name:
-        raise HTTPException(
-            status_code=400,
-            detail="Invalid folder name. Use a simple folder name without directory paths or '..'.",
-        )
-    return folder_name
+    return _get_validated_path_segment(folder_name, label="folder name")
 
 
+ValidatedFileName = Annotated[str, Depends(_get_validated_file_name)]
 ValidatedFolderName = Annotated[str, Depends(_get_validated_folder_name)]
 
 # Message to raise if we're in an Astra cloud environment and a component or endpoint is not supported
@@ -112,6 +100,112 @@ def remove_api_keys(flow: dict):
     return flow
 
 
+# ---------------------------------------------------------------------------
+# Export normalisation
+# ---------------------------------------------------------------------------
+
+# Top-level fields that vary between instances / users without changing logic.
+_VOLATILE_TOP_LEVEL: frozenset[str] = frozenset(
+    {"updated_at", "created_at", "user_id", "folder_id", "access_type", "gradient"}
+)
+
+# Node-level fields that track UI interaction state (position, drag, selection).
+_VOLATILE_NODE_FIELDS: frozenset[str] = frozenset({"positionAbsolute", "dragging", "selected"})
+
+
+def _split_code_to_lines(flow: dict) -> None:
+    """In-place: split code template field values from strings to line arrays.
+
+    Converts ``template.<field>.value`` from a single string to a
+    ``list[str]`` (one element per line) when the field type is ``"code"``.
+    This gives git line-level diffs instead of a single opaque blob.
+    """
+    for node in flow.get("data", {}).get("nodes", []):
+        template = node.get("data", {}).get("node", {}).get("template", {})
+        if not isinstance(template, dict):
+            continue
+        for field_data in template.values():
+            if not isinstance(field_data, dict):
+                continue
+            if field_data.get("type") == "code":
+                value = field_data.get("value")
+                if isinstance(value, str):
+                    # split("\n") — not splitlines() — so that the trailing newline
+                    # is preserved as a final empty string, keeping the round-trip
+                    # lossless: "\n".join(s.split("\n")) == s for any string s.
+                    field_data["value"] = value.split("\n")
+
+
+def _join_code_from_lines(flow: dict) -> None:
+    """In-place: rejoin code template line arrays back to strings.
+
+    Inverse of :func:`_split_code_to_lines`.  Safe to call on flows that
+    already use the string format — ``isinstance`` guard means it's a no-op.
+    """
+    for node in flow.get("data", {}).get("nodes", []):
+        template = node.get("data", {}).get("node", {}).get("template", {})
+        if not isinstance(template, dict):
+            continue
+        for field_data in template.values():
+            if not isinstance(field_data, dict):
+                continue
+            if field_data.get("type") == "code":
+                value = field_data.get("value")
+                if isinstance(value, list):
+                    field_data["value"] = "\n".join(value)
+
+
+def normalize_flow_for_export(flow: dict) -> dict:
+    """Return a git-friendly, deterministic copy of a flow dict.
+
+    Applied to every flow before it is written into a download ZIP.
+
+    Transformations
+    ---------------
+    * Strips volatile top-level fields (``updated_at``, ``created_at``,
+      ``user_id``, ``folder_id``, ``access_type``, ``gradient``) — these
+      change between instances / users without affecting flow logic.
+    * Strips node UI-state fields (``positionAbsolute``, ``dragging``,
+      ``selected``) — these change on every canvas interaction.
+    * Converts ``template.<field>.value`` strings to ``list[str]`` for
+      ``type == "code"`` fields, enabling line-level git diffs.
+
+    Key sorting is handled at serialisation time via
+    ``orjson_dumps(sort_keys=True)``.
+    """
+    import copy
+
+    flow = copy.deepcopy(flow)
+
+    # Strip volatile top-level metadata
+    for key in _VOLATILE_TOP_LEVEL:
+        flow.pop(key, None)
+
+    # Strip node UI state
+    for node in flow.get("data", {}).get("nodes", []):
+        for key in _VOLATILE_NODE_FIELDS:
+            node.pop(key, None)
+
+    # Code → line arrays
+    _split_code_to_lines(flow)
+
+    return flow
+
+
+def normalize_code_for_import(flow: dict) -> dict:
+    """Rejoin code-as-lines back to strings for backward-compatible import.
+
+    Accepts both the list format produced by :func:`normalize_flow_for_export`
+    and the legacy single-string format, so this function is safe to call
+    unconditionally on every uploaded flow.
+    """
+    import copy
+
+    flow = copy.deepcopy(flow)
+    _join_code_from_lines(flow)
+    return flow
+
+
 def build_input_keys_response(langchain_object, artifacts):
     """Build the input keys response."""
     input_keys_response = {
@@ -142,7 +236,13 @@ def build_input_keys_response(langchain_object, artifacts):
     return input_keys_response
 
 
-def validate_is_component(flows: list[Flow]):
+def validate_is_component(flows: list[Flow]) -> list[Flow]:
+    """Return flows with ``is_component`` inferred from flow data when unset.
+
+    Note: mutates the ORM instances in-place because SQLAlchemy requires
+    mutation for dirty-tracking.  This is an intentional exception to the
+    immutability guideline — creating copies would detach them from the session.
+    """
     for flow in flows:
         if not flow.data or flow.is_component is not None:
             continue
@@ -202,68 +302,6 @@ def format_elapsed_time(elapsed_time: float) -> str:
     return f"{minutes} {minutes_unit}, {seconds} {seconds_unit}"
 
 
-async def _get_flow_name(flow_id: uuid.UUID) -> str:
-    async with session_scope() as session:
-        flow = await session.get(Flow, flow_id)
-        if flow is None:
-            msg = f"Flow {flow_id} not found"
-            raise ValueError(msg)
-    return flow.name
-
-
-async def build_graph_from_data(flow_id: uuid.UUID | str, payload: dict, **kwargs):
-    """Build and cache the graph."""
-    # Get flow name
-    if "flow_name" not in kwargs:
-        flow_name = await _get_flow_name(flow_id if isinstance(flow_id, uuid.UUID) else uuid.UUID(flow_id))
-    else:
-        flow_name = kwargs["flow_name"]
-    str_flow_id = str(flow_id)
-    session_id = kwargs.get("session_id") or str_flow_id
-
-    graph = Graph.from_payload(payload, str_flow_id, flow_name, kwargs.get("user_id"))
-    for vertex_id in graph.has_session_id_vertices:
-        vertex = graph.get_vertex(vertex_id)
-        if vertex is None:
-            msg = f"Vertex {vertex_id} not found"
-            raise ValueError(msg)
-        if not vertex.raw_params.get("session_id"):
-            vertex.update_raw_params({"session_id": session_id}, overwrite=True)
-
-    graph.session_id = session_id
-    await graph.initialize_run()
-    return graph
-
-
-async def build_graph_from_db_no_cache(flow_id: uuid.UUID, session: AsyncSession, **kwargs):
-    """Build and cache the graph."""
-    flow: Flow | None = await session.get(Flow, flow_id)
-    if not flow or not flow.data:
-        msg = "Invalid flow ID"
-        raise ValueError(msg)
-    kwargs["user_id"] = kwargs.get("user_id") or str(flow.user_id)
-    return await build_graph_from_data(flow_id, flow.data, flow_name=flow.name, **kwargs)
-
-
-async def build_graph_from_db(flow_id: uuid.UUID, session: AsyncSession, chat_service: ChatService, **kwargs):
-    graph = await build_graph_from_db_no_cache(flow_id=flow_id, session=session, **kwargs)
-    await chat_service.set_cache(str(flow_id), graph)
-    return graph
-
-
-async def build_and_cache_graph_from_data(
-    flow_id: uuid.UUID | str,
-    chat_service: ChatService,
-    graph_data: dict,
-):  # -> Graph | Any:
-    """Build and cache the graph."""
-    # Convert flow_id to str if it's UUID
-    str_flow_id = str(flow_id) if isinstance(flow_id, uuid.UUID) else flow_id
-    graph = Graph.from_payload(graph_data, str_flow_id)
-    await chat_service.set_cache(str_flow_id, graph)
-    return graph
-
-
 def format_syntax_error_message(exc: SyntaxError) -> str:
     """Format a SyntaxError message for returning to the frontend."""
     if exc.text is None:
@@ -344,31 +382,13 @@ def parse_value(value: Any, input_type: str) -> Any:
         if isinstance(value, dict):
             return value
         try:
-            return literal_eval(value) if value is not None else {}
-        except (ValueError, SyntaxError):
+            parsed = _json.loads(value) if value is not None else {}
+            return parsed if isinstance(parsed, dict) else {}
+        except (ValueError, TypeError):
             return {}
     return value
 
 
-async def cascade_delete_flow(session: AsyncSession, flow_id: uuid.UUID) -> None:
-    try:
-        # TODO: Verify if deleting messages is safe in terms of session id relevance
-        # If we delete messages directly, rather than setting flow_id to null,
-        # it might cause unexpected behaviors because the session id could still be
-        # used elsewhere to search for these messages.
-        await session.exec(delete(MessageTable).where(MessageTable.flow_id == flow_id))
-        await session.exec(delete(TransactionTable).where(TransactionTable.flow_id == flow_id))
-        await session.exec(delete(VertexBuildTable).where(VertexBuildTable.flow_id == flow_id))
-        # Explicit delete despite FK CASCADE — SQLite doesn't enforce FK cascades
-        # by default (requires PRAGMA foreign_keys = ON), and this function follows
-        # the existing pattern of explicitly deleting all child records.
-        await session.exec(delete(FlowVersion).where(FlowVersion.flow_id == flow_id))
-        await session.exec(delete(Flow).where(Flow.id == flow_id))
-    except Exception as e:
-        msg = f"Unable to cascade delete flow: {flow_id}"
-        raise RuntimeError(msg, e) from e
-
-
 def custom_params(
     page: int | None = Query(None),
     size: int | None = Query(None),
@@ -378,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 langflow.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 langflow.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-LANGFLOW-GLOBAL-VAR-*.
 

@@ -0,0 +1,163 @@
+"""Flow graph building, cascade deletion, and public flow verification utilities."""
+
+from __future__ import annotations
+
+import uuid
+from typing import TYPE_CHECKING
+
+from fastapi import HTTPException
+from lfx.graph.graph.base import Graph
+from lfx.log.logger import logger
+from lfx.services.deps import session_scope
+from sqlalchemy import delete
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from langflow.services.database.models.flow.model import Flow
+from langflow.services.database.models.flow_version.model import FlowVersion
+from langflow.services.database.models.message.model import MessageTable
+from langflow.services.database.models.transactions.model import TransactionTable
+from langflow.services.database.models.user.model import User
+from langflow.services.database.models.vertex_builds.model import VertexBuildTable
+
+if TYPE_CHECKING:
+    from langflow.services.chat.service import ChatService
+
+
+async def _get_flow_name(flow_id: uuid.UUID) -> str:
+    async with session_scope() as session:
+        flow = await session.get(Flow, flow_id)
+        if flow is None:
+            msg = f"Flow {flow_id} not found"
+            raise ValueError(msg)
+    return flow.name
+
+
+async def build_graph_from_data(flow_id: uuid.UUID | str, payload: dict, **kwargs):
+    """Build and cache the graph."""
+    # Get flow name
+    if "flow_name" not in kwargs:
+        flow_name = await _get_flow_name(flow_id if isinstance(flow_id, uuid.UUID) else uuid.UUID(flow_id))
+    else:
+        flow_name = kwargs["flow_name"]
+    str_flow_id = str(flow_id)
+    session_id = kwargs.get("session_id") or str_flow_id
+
+    graph = Graph.from_payload(payload, str_flow_id, flow_name, kwargs.get("user_id"))
+    for vertex_id in graph.has_session_id_vertices:
+        vertex = graph.get_vertex(vertex_id)
+        if vertex is None:
+            msg = f"Vertex {vertex_id} not found"
+            raise ValueError(msg)
+        if not vertex.raw_params.get("session_id"):
+            vertex.update_raw_params({"session_id": session_id}, overwrite=True)
+
+    graph.session_id = session_id
+    await graph.initialize_run()
+    return graph
+
+
+async def build_graph_from_db_no_cache(flow_id: uuid.UUID, session: AsyncSession, **kwargs):
+    """Build and cache the graph."""
+    flow: Flow | None = await session.get(Flow, flow_id)
+    if not flow or not flow.data:
+        msg = "Invalid flow ID"
+        raise ValueError(msg)
+    kwargs["user_id"] = kwargs.get("user_id") or str(flow.user_id)
+    return await build_graph_from_data(flow_id, flow.data, flow_name=flow.name, **kwargs)
+
+
+async def build_graph_from_db(flow_id: uuid.UUID, session: AsyncSession, chat_service: ChatService, **kwargs):
+    graph = await build_graph_from_db_no_cache(flow_id=flow_id, session=session, **kwargs)
+    await chat_service.set_cache(str(flow_id), graph)
+    return graph
+
+
+async def build_and_cache_graph_from_data(
+    flow_id: uuid.UUID | str,
+    chat_service: ChatService,
+    graph_data: dict,
+):  # -> Graph | Any:
+    """Build and cache the graph."""
+    # Convert flow_id to str if it's UUID
+    str_flow_id = str(flow_id) if isinstance(flow_id, uuid.UUID) else flow_id
+    graph = Graph.from_payload(graph_data, str_flow_id)
+    await chat_service.set_cache(str_flow_id, graph)
+    return graph
+
+
+async def cascade_delete_flow(session: AsyncSession, flow_id: uuid.UUID) -> None:
+    try:
+        # TODO: Verify if deleting messages is safe in terms of session id relevance
+        # If we delete messages directly, rather than setting flow_id to null,
+        # it might cause unexpected behaviors because the session id could still be
+        # used elsewhere to search for these messages.
+        await session.exec(delete(MessageTable).where(MessageTable.flow_id == flow_id))
+        await session.exec(delete(TransactionTable).where(TransactionTable.flow_id == flow_id))
+        await session.exec(delete(VertexBuildTable).where(VertexBuildTable.flow_id == flow_id))
+        # Explicit delete despite FK CASCADE -- SQLite doesn't enforce FK cascades
+        # by default (requires PRAGMA foreign_keys = ON), and this function follows
+        # the existing pattern of explicitly deleting all child records.
+        await session.exec(delete(FlowVersion).where(FlowVersion.flow_id == flow_id))
+        await session.exec(delete(Flow).where(Flow.id == flow_id))
+    except Exception as e:
+        msg = f"Unable to cascade delete flow: {flow_id}"
+        raise RuntimeError(msg, e) from e
+
+
+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 langflow.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 langflow.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("Error getting user for public flow %s", flow_id)
+        raise HTTPException(status_code=403, detail="Flow is not accessible") from exc
+
+    if not user:
+        raise HTTPException(status_code=403, detail="Flow is not accessible")
+
+    return user, new_flow_id

@@ -5,7 +5,7 @@
 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 langflow.api.utils.core import (
     API_WORDS,
     MAX_PAGE_SIZE,
@@ -17,12 +17,7 @@
     EventDeliveryType,
     ValidatedFileName,
     ValidatedFolderName,
-    build_and_cache_graph_from_data,
-    build_graph_from_data,
-    build_graph_from_db,
-    build_graph_from_db_no_cache,
     build_input_keys_response,
-    cascade_delete_flow,
     check_langflow_version,
     custom_params,
     extract_global_variables_from_headers,
@@ -34,11 +29,20 @@
     get_suggestion_message,
     get_top_level_vertices,
     has_api_terms,
+    normalize_code_for_import,
+    normalize_flow_for_export,
     parse_exception,
     parse_value,
     raise_error_if_astra_cloud_env,
     remove_api_keys,
     validate_is_component,
+)
+from langflow.api.utils.flow_utils import (
+    build_and_cache_graph_from_data,
+    build_graph_from_data,
+    build_graph_from_db,
+    build_graph_from_db_no_cache,
+    cascade_delete_flow,
     verify_public_flow_and_get_user,
 )
 
@@ -75,6 +79,8 @@
     "get_top_level_vertices",
     # Functions
     "has_api_terms",
+    "normalize_code_for_import",
+    "normalize_flow_for_export",
     "parse_exception",
     "parse_value",
     "raise_error_if_astra_cloud_env",

@@ -0,0 +1,572 @@
+"""Helper functions for flow CRUD and filesystem operations.
+
+Extracted from flows.py to keep the route-handler module concise.
+"""
+
+from __future__ import annotations
+
+import io
+import re
+import zipfile
+from datetime import datetime, timezone
+from pathlib import Path as StdlibPath
+from typing import TYPE_CHECKING, Any
+from uuid import UUID
+
+from aiofile import async_open
+from anyio import Path
+from fastapi import HTTPException
+from fastapi.responses import StreamingResponse
+from lfx.log import logger
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from langflow.api.utils import normalize_flow_for_export, remove_api_keys
+from langflow.services.database.models.base import orjson_dumps
+from langflow.services.database.models.flow.model import (
+    Flow,
+    FlowCreate,
+    FlowRead,
+    FlowUpdate,
+)
+from langflow.services.database.models.flow.utils import get_webhook_component_in_flow
+from langflow.services.database.models.folder.model import Folder
+from langflow.services.database.models.folder.utils import get_default_folder_id
+from langflow.services.deps import get_settings_service
+from langflow.services.storage.service import StorageService
+
+if TYPE_CHECKING:
+    from langflow.services.database.models.user.model import User
+
+
+def _get_safe_flow_path(fs_path: str, user_id: UUID, storage_service: StorageService) -> Path:
+    """Get a safe filesystem path for flow storage, restricted to user's flows directory.
+
+    Allows both absolute and relative paths, but ensures they're within the user's flows directory.
+    """
+    if not fs_path:
+        raise HTTPException(status_code=400, detail="fs_path cannot be empty")
+
+    # Normalize path separators first (before security checks to prevent backslash bypass)
+    normalized_path = fs_path.replace("\\", "/")
+
+    # Reject directory traversal and null bytes (check normalized path)
+    if ".." in normalized_path:
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid fs_path: directory traversal (..) is not allowed",
+        )
+    if "\x00" in normalized_path:
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid fs_path: null bytes are not allowed",
+        )
+
+    # Build the safe base directory path
+    base_dir = storage_service.data_dir / "flows" / str(user_id)
+    base_dir_str = str(base_dir)
+
+    # Normalize base directory path (resolve to absolute, handle symlinks)
+    # resolve() doesn't require the path to exist, it just resolves symlinks
+    try:
+        base_dir_stdlib = StdlibPath(base_dir_str).resolve()
+        base_dir_resolved = str(base_dir_stdlib)
+    except (OSError, ValueError) as e:
+        raise HTTPException(status_code=400, detail=f"Invalid base directory: {e}") from e
+
+    # Determine if path is absolute (Unix or Windows style)
+    is_absolute = normalized_path.startswith("/") or (len(normalized_path) > 1 and normalized_path[1] == ":")
+
+    if is_absolute:
+        # Absolute path - resolve and validate it's within base directory
+        try:
+            requested_path = StdlibPath(normalized_path).resolve()
+            requested_resolved = str(requested_path)
+            try:
+                # Ensure it's a subpath of the base directory
+                requested_path.relative_to(base_dir_stdlib)
+            except ValueError:
+                raise HTTPException(
+                    status_code=400,
+                    detail=(f"Absolute path must be within your flows directory: {base_dir_resolved}"),
+                ) from None
+            return Path(requested_resolved)
+        except (OSError, ValueError) as e:
+            raise HTTPException(
+                status_code=400,
+                detail=(
+                    f"Invalid file save path: {e}. "
+                    f"Verify that the path is within your flows directory: {base_dir_resolved}"
+                ),
+            ) from e
+    else:
+        # Relative path - validate that it's within the base directory
+        relative_part = normalized_path.lstrip("/")
+        safe_path = base_dir / relative_part if relative_part else base_dir
+        safe_path_stdlib = base_dir_stdlib / relative_part if relative_part else base_dir_stdlib
+        try:
+            final_resolved_str = str(safe_path_stdlib.resolve())
+
+            # Ensure resolved path stays within base (prevent symlink attacks)
+            if not final_resolved_str.startswith(base_dir_resolved):
+                raise HTTPException(
+                    status_code=400,
+                    detail="Invalid path: resolves outside allowed directory",
+                )
+        except (OSError, ValueError) as e:
+            raise HTTPException(status_code=400, detail=f"Invalid path: {e}") from e
+
+        return safe_path
+
+
+# Fields that may be updated via setattr on a Flow ORM instance.
+# Any key not in this set is silently dropped to prevent callers from
+# overwriting internal fields (e.g. ``id``, ``user_id``).
+_UPDATABLE_FLOW_FIELDS: frozenset[str] = frozenset(
+    {
+        "name",
+        "description",
+        "data",
+        "is_component",
+        "endpoint_name",
+        "tags",
+        "folder_id",
+        "icon",
+        "icon_bg_color",
+        "gradient",
+        "locked",
+        "mcp_enabled",
+        "action_name",
+        "action_description",
+        "access_type",
+        "fs_path",
+    }
+)
+
+
+def _apply_update_data(target: Flow, update_data: dict[str, Any]) -> None:
+    """Apply *update_data* to the ORM *target*, restricted to the allowlist."""
+    for key, value in update_data.items():
+        if key in _UPDATABLE_FLOW_FIELDS:
+            setattr(target, key, value)
+
+
+def _endpoint_name_was_explicitly_cleared(flow: FlowCreate | FlowUpdate) -> bool:
+    """Return whether the request explicitly asked to clear the endpoint name."""
+    return "endpoint_name" in flow.model_fields_set and flow.endpoint_name in (None, "")
+
+
+async def _verify_fs_path(path: str | None, user_id: UUID, storage_service: StorageService) -> None:
+    """Verify and prepare the filesystem path for flow storage."""
+    if path is not None:
+        # Empty strings should be rejected (None is allowed, empty string is not)
+        if path == "":
+            raise HTTPException(status_code=400, detail="fs_path cannot be empty")
+        safe_path = _get_safe_flow_path(path, user_id, storage_service)
+        await safe_path.parent.mkdir(parents=True, exist_ok=True)
+        if not await safe_path.exists():
+            await safe_path.touch()
+
+
+async def _save_flow_to_fs(flow: Flow, user_id: UUID, storage_service: StorageService) -> None:
+    """Save flow data to the filesystem at the validated path."""
+    if not flow.fs_path:
+        return
+
+    try:
+        safe_path = _get_safe_flow_path(flow.fs_path, user_id, storage_service)
+        await safe_path.parent.mkdir(parents=True, exist_ok=True)
+        # async_open expects a string path, not a Path object
+        async with async_open(str(safe_path), "w") as f:
+            await f.write(flow.model_dump_json())
+    except HTTPException:
+        raise
+    except OSError as e:
+        await logger.aexception("Failed to write flow %s to path %s", flow.name, flow.fs_path)
+        raise HTTPException(status_code=500, detail=f"Failed to write flow to filesystem: {e}") from e
+
+
+async def _deduplicate_flow_name(session: AsyncSession, name: str, user_id: UUID) -> str:
+    """Return a unique flow name for *user_id*, appending ``(N)`` if needed."""
+    if not (await session.exec(select(Flow).where(Flow.name == name).where(Flow.user_id == user_id))).first():
+        return name
+
+    flows = (
+        await session.exec(
+            select(Flow).where(Flow.name.like(f"{name} (%")).where(Flow.user_id == user_id)  # type: ignore[attr-defined]
+        )
+    ).all()
+
+    # Extract copy-number suffixes: "MyFlow (2)" → 2
+    extract_number = re.compile(rf"^{re.escape(name)} \((\d+)\)$")
+    numbers = [int(m.group(1)) for f in flows if (m := extract_number.search(f.name))]
+
+    return f"{name} ({max(numbers) + 1})" if numbers else f"{name} (1)"
+
+
+async def _deduplicate_endpoint_name(
+    session: AsyncSession,
+    endpoint_name: str,
+    user_id: UUID,
+    *,
+    fail_on_conflict: bool = False,
+) -> str:
+    """Return a unique endpoint name for *user_id*, appending ``-N`` if needed.
+
+    Raises :class:`HTTPException` 409 when *fail_on_conflict* is ``True`` and
+    the name already exists.
+    """
+    existing = (
+        await session.exec(select(Flow).where(Flow.endpoint_name == endpoint_name).where(Flow.user_id == user_id))
+    ).first()
+    if not existing:
+        return endpoint_name
+
+    if fail_on_conflict:
+        raise HTTPException(status_code=409, detail="Endpoint name must be unique")
+
+    flows = (
+        await session.exec(
+            select(Flow)
+            .where(Flow.endpoint_name.like(f"{endpoint_name}-%"))  # type: ignore[union-attr]
+            .where(Flow.user_id == user_id)
+        )
+    ).all()
+
+    numbers: list[int] = []
+    for f in flows:
+        try:
+            numbers.append(int(f.endpoint_name.split("-")[-1]))
+        except ValueError:
+            continue
+
+    next_num = (max(numbers) + 1) if numbers else 1
+    return f"{endpoint_name}-{next_num}"
+
+
+async def _validate_and_assign_folder(
+    session: AsyncSession,
+    db_flow: Flow,
+    user_id: UUID,
+) -> None:
+    """Ensure *db_flow* has a valid ``folder_id`` belonging to *user_id*.
+
+    Falls back to the default folder when the current ``folder_id`` is
+    ``None`` or references a non-existent / other-user's folder.
+    """
+    if db_flow.folder_id is not None:
+        folder_exists = (
+            await session.exec(select(Folder).where(Folder.id == db_flow.folder_id, Folder.user_id == user_id))
+        ).first()
+        if not folder_exists:
+            db_flow.folder_id = None
+
+    if db_flow.folder_id is None:
+        db_flow.folder_id = await get_default_folder_id(session, user_id)
+
+
+async def _new_flow(
+    *,
+    session: AsyncSession,
+    flow: FlowCreate,
+    user_id: UUID,
+    storage_service: StorageService,
+    flow_id: UUID | None = None,
+    fail_on_endpoint_conflict: bool = False,
+    validate_folder: bool = False,
+):
+    """Create or upsert a flow.
+
+    Args:
+        session: Database session.
+        flow: Flow creation data.
+        user_id: Owner of the new flow.
+        storage_service: Service for filesystem operations.
+        flow_id: Allows PUT upsert to create flows with a specific ID for syncing between instances.
+        fail_on_endpoint_conflict: PUT should fail predictably on conflicts rather than silently renaming.
+        validate_folder: Validates folder_id exists and belongs to user when upserting from external sources.
+    """
+    try:
+        await _verify_fs_path(flow.fs_path, user_id, storage_service)
+
+        if validate_folder and flow.folder_id is not None:
+            folder = (
+                await session.exec(select(Folder).where(Folder.id == flow.folder_id, Folder.user_id == user_id))
+            ).first()
+            if not folder:
+                raise HTTPException(status_code=400, detail="Folder not found")
+
+        # Set user_id (ignore any user_id from body for security)
+        flow.user_id = user_id
+        flow.name = await _deduplicate_flow_name(session, flow.name, user_id)
+
+        if flow.endpoint_name:
+            flow.endpoint_name = await _deduplicate_endpoint_name(
+                session, flow.endpoint_name, user_id, fail_on_conflict=fail_on_endpoint_conflict
+            )
+
+        # Exclude the id field from FlowCreate so that Flow.id (UUID, non-optional)
+        # always gets its default_factory uuid4 unless we explicitly override it below.
+        db_flow = Flow.model_validate(flow.model_dump(exclude={"id"}))
+
+        # Apply the stable ID: explicit flow_id param (PUT upsert) takes precedence,
+        # then flow.id (stable import from FlowCreate), then the uuid4 default.
+        effective_id = flow_id if flow_id is not None else flow.id
+        if effective_id is not None:
+            db_flow.id = effective_id
+
+        db_flow.updated_at = datetime.now(timezone.utc)
+        await _validate_and_assign_folder(session, db_flow, user_id)
+
+        session.add(db_flow)
+        await session.flush()
+        await session.refresh(db_flow)
+        await _save_flow_to_fs(db_flow, user_id, storage_service)
+
+        return FlowRead.model_validate(db_flow, from_attributes=True)
+    except Exception as e:
+        if hasattr(e, "errors"):
+            raise HTTPException(status_code=400, detail=str(e)) from e
+        if isinstance(e, HTTPException):
+            raise
+        logger.exception("Error creating flow")
+        raise HTTPException(status_code=500, detail="An internal error occurred while creating the flow.") from e
+
+
+async def _read_flow(
+    session: AsyncSession,
+    flow_id: UUID,
+    user_id: UUID,
+):
+    """Read a flow."""
+    stmt = select(Flow).where(Flow.id == flow_id).where(Flow.user_id == user_id)
+
+    return (await session.exec(stmt)).first()
+
+
+async def _update_existing_flow(
+    *,
+    session: AsyncSession,
+    existing_flow: Flow,
+    flow: FlowCreate,
+    current_user: User,
+    storage_service: StorageService,
+) -> FlowRead:
+    """Update an existing flow (PUT update path).
+
+    Similar to update_flow but:
+    - Fails on name/endpoint_name conflict with OTHER flows (409)
+    - Keeps existing folder_id if not provided in request
+    """
+    settings_service = get_settings_service()
+    user_id = current_user.id
+
+    # Validate fs_path if provided (use `is not None` to catch empty strings)
+    if flow.fs_path is not None:
+        await _verify_fs_path(flow.fs_path, user_id, storage_service)
+
+    # Validate folder_id if provided
+    if flow.folder_id is not None:
+        folder = (
+            await session.exec(select(Folder).where(Folder.id == flow.folder_id, Folder.user_id == user_id))
+        ).first()
+        if not folder:
+            raise HTTPException(status_code=400, detail="Folder not found")
+
+    # Check name uniqueness (excluding current flow)
+    if flow.name and flow.name != existing_flow.name:
+        name_conflict = (
+            await session.exec(
+                select(Flow).where(
+                    Flow.name == flow.name,
+                    Flow.user_id == user_id,
+                    Flow.id != existing_flow.id,
+                )
+            )
+        ).first()
+        if name_conflict:
+            raise HTTPException(status_code=409, detail="Name must be unique")
+
+    # Check endpoint_name uniqueness (excluding current flow)
+    if flow.endpoint_name and flow.endpoint_name != existing_flow.endpoint_name:
+        endpoint_conflict = (
+            await session.exec(
+                select(Flow).where(
+                    Flow.endpoint_name == flow.endpoint_name,
+                    Flow.user_id == user_id,
+                    Flow.id != existing_flow.id,
+                )
+            )
+        ).first()
+        if endpoint_conflict:
+            raise HTTPException(status_code=409, detail="Endpoint name must be unique")
+
+    # Build update data
+    update_data = flow.model_dump(exclude_unset=True, exclude_none=True)
+
+    # Preserve the existing endpoint unless the request explicitly clears it.
+    if _endpoint_name_was_explicitly_cleared(flow):
+        update_data["endpoint_name"] = None
+
+    # Remove id and user_id from update data (security)
+    update_data.pop("id", None)
+    update_data.pop("user_id", None)
+
+    # If folder_id not provided, keep existing
+    if "folder_id" not in update_data or update_data.get("folder_id") is None:
+        update_data.pop("folder_id", None)
+
+    if settings_service.settings.remove_api_keys:
+        update_data = remove_api_keys(update_data)
+
+    _apply_update_data(existing_flow, update_data)
+
+    webhook_component = get_webhook_component_in_flow(existing_flow.data or {})
+    existing_flow.webhook = webhook_component is not None
+    existing_flow.updated_at = datetime.now(timezone.utc)
+
+    session.add(existing_flow)
+    await session.flush()
+    await session.refresh(existing_flow)
+    await _save_flow_to_fs(existing_flow, user_id, storage_service)
+
+    return FlowRead.model_validate(existing_flow, from_attributes=True)
+
+
+async def _patch_flow(
+    *,
+    session: AsyncSession,
+    db_flow: Flow,
+    flow: FlowUpdate,
+    user_id: UUID,
+    storage_service: StorageService,
+) -> FlowRead:
+    """Apply a partial update (PATCH) to an existing flow and return a FlowRead."""
+    settings_service = get_settings_service()
+
+    update_data = flow.model_dump(exclude_unset=True, exclude_none=True)
+
+    # Preserve the existing endpoint unless the request explicitly clears it.
+    if _endpoint_name_was_explicitly_cleared(flow):
+        update_data["endpoint_name"] = None
+
+    if settings_service.settings.remove_api_keys:
+        update_data = remove_api_keys(update_data)
+
+    _apply_update_data(db_flow, update_data)
+
+    # Validate fs_path if it was changed (will raise HTTPException if invalid)
+    if "fs_path" in update_data:
+        await _verify_fs_path(db_flow.fs_path, user_id, storage_service)
+
+    webhook_component = get_webhook_component_in_flow(db_flow.data) if db_flow.data else None
+    db_flow.webhook = webhook_component is not None
+    db_flow.updated_at = datetime.now(timezone.utc)
+
+    await _validate_and_assign_folder(session, db_flow, user_id)
+
+    session.add(db_flow)
+    await session.flush()
+    await session.refresh(db_flow)
+    await _save_flow_to_fs(db_flow, user_id, storage_service)
+
+    return FlowRead.model_validate(db_flow, from_attributes=True)
+
+
+async def _upsert_flow_list(
+    *,
+    session: AsyncSession,
+    flows: list[FlowCreate],
+    current_user: User,
+    storage_service: StorageService,
+    folder_id: UUID | None = None,
+) -> list[FlowRead]:
+    """Import a list of flows with upsert semantics (used by the upload endpoint).
+
+    For each flow:
+    - If it has an ID matching an existing flow owned by the user, update in place.
+    - If it has an ID claimed by another user, mint a fresh UUID.
+    - Otherwise create with the provided or generated ID.
+    """
+    flow_reads: list[FlowRead] = []
+    for flow in flows:
+        flow.user_id = current_user.id
+        if folder_id:
+            flow.folder_id = folder_id
+
+        if flow.id is not None:
+            existing = (await session.exec(select(Flow).where(Flow.id == flow.id))).first()
+
+            if existing is not None and existing.user_id == current_user.id:
+                flow_read = await _update_existing_flow(
+                    session=session,
+                    existing_flow=existing,
+                    flow=flow,
+                    current_user=current_user,
+                    storage_service=storage_service,
+                )
+            elif existing is not None:
+                flow.id = None
+                flow_read = await _new_flow(
+                    session=session, flow=flow, user_id=current_user.id, storage_service=storage_service
+                )
+            else:
+                flow_read = await _new_flow(
+                    session=session,
+                    flow=flow,
+                    user_id=current_user.id,
+                    storage_service=storage_service,
+                    flow_id=flow.id,
+                )
+        else:
+            flow_read = await _new_flow(
+                session=session, flow=flow, user_id=current_user.id, storage_service=storage_service
+            )
+
+        flow_reads.append(flow_read)
+    return flow_reads
+
+
+def _sanitize_flow_filename(raw_name: str, fallback_id: str = "flow") -> str:
+    """Return a filesystem-safe filename from a flow name.
+
+    Strips directory separators, null bytes, and Windows reserved device names.
+    """
+    name = str(raw_name).replace("/", "_").replace("\\", "_")
+    name = name.replace("\x00", "").replace("..", "_").strip()
+    # Reject Windows reserved device names (CON, PRN, AUX, NUL, COM1-9, LPT1-9)
+    import re as _re
+
+    if _re.match(r"^(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(\..+)?$", name, _re.IGNORECASE):
+        name = f"_{name}"
+    return name or fallback_id
+
+
+def _build_flows_download_response(
+    flows: list[Flow],
+) -> StreamingResponse | dict:
+    """Build a download response (ZIP or single JSON) for the given flows.
+
+    Strips API keys and normalises for git-friendly export before packaging.
+    """
+    normalised_flows = [normalize_flow_for_export(remove_api_keys(flow.model_dump())) for flow in flows]
+
+    if len(normalised_flows) > 1:
+        zip_stream = io.BytesIO()
+        with zipfile.ZipFile(zip_stream, "w") as zip_file:
+            for flow_dict in normalised_flows:
+                flow_json = orjson_dumps(flow_dict, sort_keys=True)
+                raw_name = str(flow_dict.get("name", "flow"))
+                safe_name = _sanitize_flow_filename(raw_name, str(flow_dict.get("id", "flow")))
+                zip_file.writestr(f"{safe_name}.json", flow_json)
+
+        zip_stream.seek(0)
+        current_time = datetime.now(tz=timezone.utc).astimezone().strftime("%Y%m%d_%H%M%S")
+        filename = f"{current_time}_langflow_flows.zip"
+
+        return StreamingResponse(
+            zip_stream,
+            media_type="application/x-zip-compressed",
+            headers={"Content-Disposition": f"attachment; filename={filename}"},
+        )
+    return normalised_flows[0]

@@ -1,32 +1,44 @@
 from __future__ import annotations
 
+import asyncio
 import io
-import json
-import re
+import threading
 import zipfile
-from datetime import datetime, timezone
-from pathlib import Path as StdlibPath
 from typing import Annotated
 from uuid import UUID
 
 import orjson
-from aiofile import async_open
-from anyio import Path
-from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile
+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.log import logger
+from lfx.services.cache.utils import CACHE_MISS
 from sqlmodel import and_, col, select
-from sqlmodel.ext.asyncio.session import AsyncSession
 
-from langflow.api.utils import CurrentActiveUser, DbSession, cascade_delete_flow, remove_api_keys, validate_is_component
+from langflow.api.utils import (
+    CurrentActiveUser,
+    DbSession,
+    cascade_delete_flow,
+    normalize_code_for_import,
+    validate_is_component,
+)
 from langflow.api.utils.zip_utils import extract_flows_from_zip
+from langflow.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 langflow.api.v1.schemas import FlowListCreate
 from langflow.helpers.user import get_user_by_flow_id_or_endpoint_name
 from langflow.initial_setup.constants import STARTER_FOLDER_NAME
 from langflow.services.auth.utils import get_current_active_user
+from langflow.services.cache.service import ThreadingInMemoryCache
 from langflow.services.database.models.flow.model import (
     AccessTypeEnum,
     Flow,
@@ -35,267 +47,39 @@
     FlowRead,
     FlowUpdate,
 )
-from langflow.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_version_list, strip_version_data,
 # and FlowVersionError from the flow_version modules.
 from langflow.services.database.models.folder.constants import DEFAULT_FOLDER_NAME
 from langflow.services.database.models.folder.model import Folder
-from langflow.services.database.models.folder.utils import get_default_folder_id
 from langflow.services.deps import get_settings_service, get_storage_service
 from langflow.services.storage.service import StorageService
 from langflow.utils.compression import compress_response
 
-# build router
-router = APIRouter(prefix="/flows", tags=["Flows"])
-
+# Re-export helpers so existing ``from langflow.api.v1.flows import ...`` still works.
+__all__ = [
+    "_get_safe_flow_path",
+    "_new_flow",
+    "_read_flow",
+    "_save_flow_to_fs",
+    "_update_existing_flow",
+    "_verify_fs_path",
+]
 
-def _get_safe_flow_path(fs_path: str, user_id: UUID, storage_service: StorageService) -> Path:
-    """Get a safe filesystem path for flow storage, restricted to user's flows directory.
 
-    Allows both absolute and relative paths, but ensures they're within the user's flows directory.
-    """
-    if not fs_path:
-        raise HTTPException(status_code=400, detail="fs_path cannot be empty")
-
-    # Normalize path separators first (before security checks to prevent backslash bypass)
-    normalized_path = fs_path.replace("\\", "/")
-
-    # Reject directory traversal and null bytes (check normalized path)
-    if ".." in normalized_path:
-        raise HTTPException(
-            status_code=400,
-            detail="Invalid fs_path: directory traversal (..) is not allowed",
-        )
-    if "\x00" in normalized_path:
-        raise HTTPException(
-            status_code=400,
-            detail="Invalid fs_path: null bytes are not allowed",
-        )
-
-    # Build the safe base directory path
-    base_dir = storage_service.data_dir / "flows" / str(user_id)
-    base_dir_str = str(base_dir)
-
-    # Normalize base directory path (resolve to absolute, handle symlinks)
-    # resolve() doesn't require the path to exist, it just resolves symlinks
-    try:
-        base_dir_stdlib = StdlibPath(base_dir_str).resolve()
-        base_dir_resolved = str(base_dir_stdlib)
-    except (OSError, ValueError) as e:
-        raise HTTPException(status_code=400, detail=f"Invalid base directory: {e}") from e
-
-    # Determine if path is absolute (Unix or Windows style)
-    is_absolute = normalized_path.startswith("/") or (len(normalized_path) > 1 and normalized_path[1] == ":")
-
-    if is_absolute:
-        # Absolute path - resolve and validate it's within base directory
-        try:
-            requested_path = StdlibPath(normalized_path).resolve()
-            requested_resolved = str(requested_path)
-            try:
-                # Ensure it's a subpath of the base directory
-                requested_path.relative_to(base_dir_stdlib)
-            except ValueError:
-                raise HTTPException(
-                    status_code=400,
-                    detail=(f"Absolute path must be within your flows directory: {base_dir_resolved}"),
-                ) from None
-            return Path(requested_resolved)
-        except (OSError, ValueError) as e:
-            raise HTTPException(
-                status_code=400,
-                detail=(
-                    f"Invalid file save path: {e}. "
-                    f"Verify that the path is within your flows directory: {base_dir_resolved}"
-                ),
-            ) from e
-    else:
-        # Relative path - validate that it's within the base directory
-        relative_part = normalized_path.lstrip("/")
-        safe_path = base_dir / relative_part if relative_part else base_dir
-        safe_path_stdlib = base_dir_stdlib / relative_part if relative_part else base_dir_stdlib
-        try:
-            final_resolved_str = str(safe_path_stdlib.resolve())
+def _handle_unique_constraint_error(exc: Exception, *, status_code: int = 400) -> HTTPException:
+    """Parse a UNIQUE constraint error and return an appropriate HTTPException."""
+    msg = str(exc)
+    if "UNIQUE constraint failed" not in msg:
+        return HTTPException(status_code=500, detail=msg)
+    columns = msg.split("UNIQUE constraint failed: ")[1].split(".")[1].split("\n")[0]
+    column = columns.split(",")[1] if "id" in columns.split(",")[0] else columns.split(",")[0]
+    return HTTPException(status_code=status_code, detail=f"{column.capitalize().replace('_', ' ')} must be unique")
 
-            # Ensure resolved path stays within base (prevent symlink attacks)
-            if not final_resolved_str.startswith(base_dir_resolved):
-                raise HTTPException(
-                    status_code=400,
-                    detail="Invalid path: resolves outside allowed directory",
-                )
-        except (OSError, ValueError) as e:
-            raise HTTPException(status_code=400, detail=f"Invalid path: {e}") from e
-
-        return safe_path
-
-
-async def _verify_fs_path(path: str | None, user_id: UUID, storage_service: StorageService) -> None:
-    """Verify and prepare the filesystem path for flow storage."""
-    if path is not None:
-        # Empty strings should be rejected (None is allowed, empty string is not)
-        if path == "":
-            raise HTTPException(status_code=400, detail="fs_path cannot be empty")
-        safe_path = _get_safe_flow_path(path, user_id, storage_service)
-        await safe_path.parent.mkdir(parents=True, exist_ok=True)
-        if not await safe_path.exists():
-            await safe_path.touch()
-
-
-async def _save_flow_to_fs(flow: Flow, user_id: UUID, storage_service: StorageService) -> None:
-    """Save flow data to the filesystem at the validated path."""
-    if not flow.fs_path:
-        return
-
-    try:
-        safe_path = _get_safe_flow_path(flow.fs_path, user_id, storage_service)
-        await safe_path.parent.mkdir(parents=True, exist_ok=True)
-        # async_open expects a string path, not a Path object
-        async with async_open(str(safe_path), "w") as f:
-            await f.write(flow.model_dump_json())
-    except HTTPException:
-        raise
-    except OSError as e:
-        await logger.aexception("Failed to write flow %s to path %s", flow.name, flow.fs_path)
-        raise HTTPException(status_code=500, detail=f"Failed to write flow to filesystem: {e}") from e
-
-
-async def _new_flow(
-    *,
-    session: AsyncSession,
-    flow: FlowCreate,
-    user_id: UUID,
-    storage_service: StorageService,
-    flow_id: UUID | None = None,
-    fail_on_endpoint_conflict: bool = False,
-    validate_folder: bool = False,
-):
-    """Create or upsert a flow.
-
-    Args:
-        session: Database session.
-        flow: Flow creation data.
-        user_id: Owner of the new flow.
-        storage_service: Service for filesystem operations.
-        flow_id: Allows PUT upsert to create flows with a specific ID for syncing between instances.
-        fail_on_endpoint_conflict: PUT should fail predictably on conflicts rather than silently renaming.
-        validate_folder: Validates folder_id exists and belongs to user when upserting from external sources.
-    """
-    try:
-        # Validate fs_path if provided (will raise HTTPException if invalid)
-        await _verify_fs_path(flow.fs_path, user_id, storage_service)
-
-        # Validate folder_id if requested
-        if validate_folder and flow.folder_id is not None:
-            folder = (
-                await session.exec(select(Folder).where(Folder.id == flow.folder_id, Folder.user_id == user_id))
-            ).first()
-            if not folder:
-                raise HTTPException(status_code=400, detail="Folder not found")
-
-        # Set user_id (ignore any user_id from body for security)
-        flow.user_id = user_id
-
-        # Check if the flow.name is unique
-        # there might be flows with name like: "MyFlow", "MyFlow (1)", "MyFlow (2)"
-        # so we need to check if the name is unique with `like` operator
-        # if we find a flow with the same name, we add a number to the end of the name
-        # based on the highest number found
-        if (await session.exec(select(Flow).where(Flow.name == flow.name).where(Flow.user_id == user_id))).first():
-            flows = (
-                await session.exec(
-                    select(Flow).where(Flow.name.like(f"{flow.name} (%")).where(Flow.user_id == user_id)  # type: ignore[attr-defined]
-                )
-            ).all()
-            if flows:
-                # Use regex to extract numbers only from flows that follow the copy naming pattern:
-                # "{original_name} ({number})"
-                # This avoids extracting numbers from the original flow name if it naturally contains parentheses
-                #
-                # Examples:
-                # - For flow "My Flow": matches "My Flow (1)", "My Flow (2)" → extracts 1, 2
-                # - For flow "Analytics (Q1)": matches "Analytics (Q1) (1)" → extracts 1
-                #   but does NOT match "Analytics (Q1)" → avoids extracting the original "1"
-                extract_number = re.compile(rf"^{re.escape(flow.name)} \((\d+)\)$")
-                numbers = []
-                for _flow in flows:
-                    result = extract_number.search(_flow.name)
-                    if result:
-                        numbers.append(int(result.groups(1)[0]))
-                if numbers:
-                    flow.name = f"{flow.name} ({max(numbers) + 1})"
-                else:
-                    flow.name = f"{flow.name} (1)"
-            else:
-                flow.name = f"{flow.name} (1)"
-
-        # Check if the endpoint is unique
-        if (
-            flow.endpoint_name
-            and (
-                await session.exec(
-                    select(Flow).where(Flow.endpoint_name == flow.endpoint_name).where(Flow.user_id == user_id)
-                )
-            ).first()
-        ):
-            if fail_on_endpoint_conflict:
-                raise HTTPException(status_code=409, detail="Endpoint name must be unique")
-
-            # Auto-rename endpoint
-            flows = (
-                await session.exec(
-                    select(Flow)
-                    .where(Flow.endpoint_name.like(f"{flow.endpoint_name}-%"))  # type: ignore[union-attr]
-                    .where(Flow.user_id == user_id)
-                )
-            ).all()
-            if flows:
-                # The endpoint name is like "my-endpoint","my-endpoint-1", "my-endpoint-2"
-                # so we need to get the highest number and add 1
-                # we need to get the last part of the endpoint name
-                numbers = [int(flow.endpoint_name.split("-")[-1]) for flow in flows]
-                flow.endpoint_name = f"{flow.endpoint_name}-{max(numbers) + 1}"
-            else:
-                flow.endpoint_name = f"{flow.endpoint_name}-1"
-
-        db_flow = Flow.model_validate(flow, from_attributes=True)
-
-        # Use specified ID if provided (for PUT upsert)
-        if flow_id is not None:
-            db_flow.id = flow_id
-
-        db_flow.updated_at = datetime.now(timezone.utc)
-
-        # Validate folder_id exists, or fall back to default folder
-        if db_flow.folder_id is not None:
-            folder_exists = (
-                await session.exec(select(Folder).where(Folder.id == db_flow.folder_id, Folder.user_id == user_id))
-            ).first()
-            if not folder_exists:
-                # Folder doesn't exist or doesn't belong to user, use default
-                db_flow.folder_id = None
-
-        if db_flow.folder_id is None:
-            # Make sure flows always have a folder (auto-create default folder if needed)
-            db_flow.folder_id = await get_default_folder_id(session, user_id)
 
-        session.add(db_flow)
-
-        # Persist and refresh
-        await session.flush()
-        await session.refresh(db_flow)
-        await _save_flow_to_fs(db_flow, user_id, storage_service)
-
-        # Convert to FlowRead while session is still active
-        return FlowRead.model_validate(db_flow, from_attributes=True)
-    except Exception as e:
-        # If it is a validation error, return the error message
-        if hasattr(e, "errors"):
-            raise HTTPException(status_code=400, detail=str(e)) from e
-        if isinstance(e, HTTPException):
-            raise
-        raise HTTPException(status_code=500, detail=str(e)) from e
+# build router
+router = APIRouter(prefix="/flows", tags=["Flows"])
 
 
 @router.post("/", response_model=FlowRead, status_code=201)
@@ -308,21 +92,10 @@ async def create_flow(
 ):
     try:
         return await _new_flow(session=session, flow=flow, user_id=current_user.id, storage_service=storage_service)
+    except HTTPException:
+        raise
     except Exception as e:
-        if "UNIQUE constraint failed" in str(e):
-            # Get the name of the column that failed
-            columns = str(e).split("UNIQUE constraint failed: ")[1].split(".")[1].split("\n")[0]
-            # UNIQUE constraint failed: flow.user_id, flow.name
-            # or UNIQUE constraint failed: flow.name
-            # if the column has id in it, we want the other column
-            column = columns.split(",")[1] if "id" in columns.split(",")[0] else columns.split(",")[0]
-
-            raise HTTPException(
-                status_code=400, detail=f"{column.capitalize().replace('_', ' ')} must be unique"
-            ) from e
-        if isinstance(e, HTTPException):
-            raise
-        raise HTTPException(status_code=500, detail=str(e)) from e
+        raise _handle_unique_constraint_error(e) from e
 
 
 @router.get("/", response_model=list[FlowRead] | Page[FlowRead] | list[FlowHeader], status_code=200)
@@ -337,26 +110,7 @@ async def read_flows(
     params: Annotated[Params, Depends()],
     header_flows: bool = False,
 ):
-    """Retrieve a list of flows with pagination support.
-
-    Args:
-        current_user (User): The current authenticated user.
-        session (Session): The database session.
-        settings_service (SettingsService): The settings service.
-        components_only (bool, optional): Whether to return only components. Defaults to False.
-
-        get_all (bool, optional): Whether to return all flows without pagination. Defaults to True.
-        **This field must be True because of backward compatibility with the frontend - Release: 1.0.20**
-
-        folder_id (UUID, optional): The project ID. Defaults to None.
-        params (Params): Pagination parameters.
-        remove_example_flows (bool, optional): Whether to remove example flows. Defaults to False.
-        header_flows (bool, optional): Whether to return only specific headers of the flows. Defaults to False.
-
-    Returns:
-        list[FlowRead] | Page[FlowRead] | list[FlowHeader]
-        A list of flows or a paginated response containing the list of flows or a list of flow headers.
-    """
+    """Retrieve a list of flows with optional pagination, filtering, and header-only mode."""
     try:
         auth_settings = get_settings_service().auth_settings
 
@@ -415,18 +169,10 @@ async def read_flows(
             return await apaginate(session, stmt, params=params)
 
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e)) from e
+        import logging as _logging
 
-
-async def _read_flow(
-    session: AsyncSession,
-    flow_id: UUID,
-    user_id: UUID,
-):
-    """Read a flow."""
-    stmt = select(Flow).where(Flow.id == flow_id).where(Flow.user_id == user_id)
-
-    return (await session.exec(stmt)).first()
+        _logging.getLogger(__name__).exception("Error listing flows")
+        raise HTTPException(status_code=500, detail="An internal error occurred while listing flows.") from e
 
 
 @router.get("/{flow_id}", response_model=FlowRead, status_code=200)
@@ -468,80 +214,25 @@ async def update_flow(
     storage_service: Annotated[StorageService, Depends(get_storage_service)],
 ):
     """Update a flow."""
-    settings_service = get_settings_service()
     try:
-        db_flow = await _read_flow(
-            session=session,
-            flow_id=flow_id,
-            user_id=current_user.id,
-        )
-
+        db_flow = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.id)
         if not db_flow:
             raise HTTPException(status_code=404, detail="Flow not found")
 
-        update_data = flow.model_dump(exclude_unset=True, exclude_none=True)
-
-        # Specifically handle endpoint_name when it's explicitly set to null or empty string
-        if flow.endpoint_name is None or flow.endpoint_name == "":
-            update_data["endpoint_name"] = None
-
-        if settings_service.settings.remove_api_keys:
-            update_data = remove_api_keys(update_data)
-
-        for key, value in update_data.items():
-            setattr(db_flow, key, value)
-
-        # Validate fs_path if it was changed (will raise HTTPException if invalid)
-        if "fs_path" in update_data:
-            await _verify_fs_path(db_flow.fs_path, current_user.id, storage_service)
-
-        webhook_component = get_webhook_component_in_flow(db_flow.data)
-        db_flow.webhook = webhook_component is not None
-        db_flow.updated_at = datetime.now(timezone.utc)
-
-        # Validate folder_id exists, or fall back to default folder
-        if db_flow.folder_id is not None:
-            folder_exists = (
-                await session.exec(
-                    select(Folder).where(Folder.id == db_flow.folder_id, Folder.user_id == current_user.id)
-                )
-            ).first()
-            if not folder_exists:
-                # Folder doesn't exist or doesn't belong to user, use default
-                db_flow.folder_id = None
-
-        if db_flow.folder_id is None:
-            # Make sure flows always have a folder (auto-create default folder if needed)
-            db_flow.folder_id = await get_default_folder_id(session, current_user.id)
-
-        session.add(db_flow)
-        await session.flush()
-        await session.refresh(db_flow)
-        await _save_flow_to_fs(db_flow, current_user.id, storage_service)
-
-        # Convert to FlowRead while session is still active to avoid detached instance errors
-        flow_read = FlowRead.model_validate(db_flow, from_attributes=True)
-
+        return await _patch_flow(
+            session=session,
+            db_flow=db_flow,
+            flow=flow,
+            user_id=current_user.id,
+            storage_service=storage_service,
+        )
+    except HTTPException:
+        raise
     except Exception as e:
-        if "UNIQUE constraint failed" in str(e):
-            # Get the name of the column that failed
-            columns = str(e).split("UNIQUE constraint failed: ")[1].split(".")[1].split("\n")[0]
-            # UNIQUE constraint failed: flow.user_id, flow.name
-            # or UNIQUE constraint failed: flow.name
-            # if the column has id in it, we want the other column
-            column = columns.split(",")[1] if "id" in columns.split(",")[0] else columns.split(",")[0]
-            raise HTTPException(
-                status_code=400, detail=f"{column.capitalize().replace('_', ' ')} must be unique"
-            ) from e
-
-        if hasattr(e, "status_code"):
-            raise HTTPException(status_code=e.status_code, detail=str(e)) from e
-        raise HTTPException(status_code=500, detail=str(e)) from e
-
-    return flow_read
+        raise _handle_unique_constraint_error(e) from e
 
 
-@router.put("/{flow_id}", response_model=FlowRead, include_in_schema=False)
+@router.put("/{flow_id}", response_model=FlowRead)
 async def upsert_flow(
     *,
     session: DbSession,
@@ -552,11 +243,7 @@ async def upsert_flow(
 ):
     """Create or update a flow with a specific ID (upsert).
 
-    - If the flow doesn't exist: creates it with the specified ID
-    - If the flow exists and belongs to the current user: updates it
-    - If the flow exists but belongs to another user: returns 404
-
-    Returns 201 for creation, 200 for update.
+    Returns 201 for creation, 200 for update.  Returns 404 if owned by another user.
     """
     from fastapi.responses import JSONResponse
 
@@ -596,103 +283,7 @@ async def upsert_flow(
     except HTTPException:
         raise
     except Exception as e:
-        if "UNIQUE constraint failed" in str(e):
-            columns = str(e).split("UNIQUE constraint failed: ")[1].split(".")[1].split("\n")[0]
-            column = columns.split(",")[1] if "id" in columns.split(",")[0] else columns.split(",")[0]
-            raise HTTPException(
-                status_code=409, detail=f"{column.capitalize().replace('_', ' ')} must be unique"
-            ) from e
-        raise HTTPException(status_code=500, detail=str(e)) from e
-
-
-async def _update_existing_flow(
-    *,
-    session: AsyncSession,
-    existing_flow: Flow,
-    flow: FlowCreate,
-    current_user,
-    storage_service: StorageService,
-) -> FlowRead:
-    """Update an existing flow (PUT update path).
-
-    Similar to update_flow but:
-    - Fails on name/endpoint_name conflict with OTHER flows (409)
-    - Keeps existing folder_id if not provided in request
-    """
-    settings_service = get_settings_service()
-    user_id = current_user.id
-
-    # Validate fs_path if provided (use `is not None` to catch empty strings)
-    if flow.fs_path is not None:
-        await _verify_fs_path(flow.fs_path, user_id, storage_service)
-
-    # Validate folder_id if provided
-    if flow.folder_id is not None:
-        folder = (
-            await session.exec(select(Folder).where(Folder.id == flow.folder_id, Folder.user_id == user_id))
-        ).first()
-        if not folder:
-            raise HTTPException(status_code=400, detail="Folder not found")
-
-    # Check name uniqueness (excluding current flow)
-    if flow.name and flow.name != existing_flow.name:
-        name_conflict = (
-            await session.exec(
-                select(Flow).where(
-                    Flow.name == flow.name,
-                    Flow.user_id == user_id,
-                    Flow.id != existing_flow.id,
-                )
-            )
-        ).first()
-        if name_conflict:
-            raise HTTPException(status_code=409, detail="Name must be unique")
-
-    # Check endpoint_name uniqueness (excluding current flow)
-    if flow.endpoint_name and flow.endpoint_name != existing_flow.endpoint_name:
-        endpoint_conflict = (
-            await session.exec(
-                select(Flow).where(
-                    Flow.endpoint_name == flow.endpoint_name,
-                    Flow.user_id == user_id,
-                    Flow.id != existing_flow.id,
-                )
-            )
-        ).first()
-        if endpoint_conflict:
-            raise HTTPException(status_code=409, detail="Endpoint name must be unique")
-
-    # Build update data
-    update_data = flow.model_dump(exclude_unset=True, exclude_none=True)
-
-    # Handle endpoint_name explicitly set to null or empty string (allow clearing)
-    if flow.endpoint_name is None or flow.endpoint_name == "":
-        update_data["endpoint_name"] = None
-
-    # Remove id and user_id from update data (security)
-    update_data.pop("id", None)
-    update_data.pop("user_id", None)
-
-    # If folder_id not provided, keep existing
-    if "folder_id" not in update_data or update_data.get("folder_id") is None:
-        update_data.pop("folder_id", None)
-
-    if settings_service.settings.remove_api_keys:
-        update_data = remove_api_keys(update_data)
-
-    for key, value in update_data.items():
-        setattr(existing_flow, key, value)
-
-    webhook_component = get_webhook_component_in_flow(existing_flow.data or {})
-    existing_flow.webhook = webhook_component is not None
-    existing_flow.updated_at = datetime.now(timezone.utc)
-
-    session.add(existing_flow)
-    await session.flush()
-    await session.refresh(existing_flow)
-    await _save_flow_to_fs(existing_flow, user_id, storage_service)
-
-    return FlowRead.model_validate(existing_flow, from_attributes=True)
+        raise _handle_unique_constraint_error(e, status_code=409) from e
 
 
 @router.delete("/{flow_id}", status_code=200)
@@ -722,10 +313,26 @@ async def create_flows(
     current_user: CurrentActiveUser,
 ):
     """Create multiple new flows."""
+    # Guard against duplicate IDs up-front so callers get a clean 422 instead
+    # of an unhandled DB IntegrityError.  Use upload_file() for upsert semantics.
+    requested_ids = [f.id for f in flow_list.flows if f.id is not None]
+    if requested_ids:
+        existing_ids = (await session.exec(select(Flow.id).where(col(Flow.id).in_(requested_ids)))).all()
+        if existing_ids:
+            conflict = ", ".join(str(i) for i in existing_ids)
+            msg = (
+                f"Flow(s) with the following IDs already exist: {conflict}. "
+                "Use the update endpoint or upload_file() for upsert semantics."
+            )
+            raise HTTPException(status_code=422, detail=msg)
+
     db_flows = []
     for flow in flow_list.flows:
         flow.user_id = current_user.id
-        db_flow = Flow.model_validate(flow, from_attributes=True)
+        # Exclude id from model_validate (same reasoning as _new_flow) and apply separately.
+        db_flow = Flow.model_validate(flow.model_dump(exclude={"id"}))
+        if flow.id is not None:
+            db_flow.id = flow.id
         session.add(db_flow)
         db_flows.append(db_flow)
 
@@ -745,11 +352,7 @@ async def upload_file(
     folder_id: UUID | None = None,
     storage_service: Annotated[StorageService, Depends(get_storage_service)],
 ):
-    """Upload flows from a file.
-
-    Accepts either a JSON file (single flow or multi-flow format) or a ZIP file
-    containing individual flow JSON files (as produced by the download endpoint).
-    """
+    """Upload flows from a JSON or ZIP file (upsert semantics for flows with stable IDs)."""
     if file is None:
         raise HTTPException(status_code=400, detail="No file provided")
 
@@ -772,39 +375,30 @@ async def upload_file(
         except orjson.JSONDecodeError as e:
             raise HTTPException(status_code=400, detail=f"Invalid JSON file: {e}") from e
 
-    flow_list = FlowListCreate(**data) if "flows" in data else FlowListCreate(flows=[FlowCreate(**data)])
+    # Normalise code fields: if exported with code-as-lines format, rejoin to
+    # strings before creating the Pydantic models so the DB always stores strings.
+    if "flows" in data:
+        data = {**data, "flows": [normalize_code_for_import(f) for f in data["flows"]]}
+        flow_list = FlowListCreate(**data)
+    else:
+        flow_list = FlowListCreate(flows=[FlowCreate(**normalize_code_for_import(data))])
 
     # TODO: Full-version import is planned as a follow-up feature.
     # When implemented, extract raw flow dicts here to read embedded "version"
     # arrays and create FlowVersion entries for each imported flow.
 
     try:
-        flow_reads = []
-        for flow in flow_list.flows:
-            flow.user_id = current_user.id
-            if folder_id:
-                flow.folder_id = folder_id
-            flow_read = await _new_flow(
-                session=session, flow=flow, user_id=current_user.id, storage_service=storage_service
-            )
-            flow_reads.append(flow_read)
+        return await _upsert_flow_list(
+            session=session,
+            flows=flow_list.flows,
+            current_user=current_user,
+            storage_service=storage_service,
+            folder_id=folder_id,
+        )
+    except HTTPException:
+        raise
     except Exception as e:
-        if "UNIQUE constraint failed" in str(e):
-            # Get the name of the column that failed
-            columns = str(e).split("UNIQUE constraint failed: ")[1].split(".")[1].split("\n")[0]
-            # UNIQUE constraint failed: flow.user_id, flow.name
-            # or UNIQUE constraint failed: flow.name
-            # if the column has id in it, we want the other column
-            column = columns.split(",")[1] if "id" in columns.split(",")[0] else columns.split(",")[0]
-
-            raise HTTPException(
-                status_code=400, detail=f"{column.capitalize().replace('_', ' ')} must be unique"
-            ) from e
-        if isinstance(e, HTTPException):
-            raise
-        raise HTTPException(status_code=500, detail=str(e)) from e
-
-    return flow_reads
+        raise _handle_unique_constraint_error(e) from e
 
 
 @router.delete("/")
@@ -813,17 +407,7 @@ async def delete_multiple_flows(
     user: CurrentActiveUser,
     db: DbSession,
 ):
-    """Delete multiple flows by their IDs.
-
-    Args:
-        flow_ids (List[str]): The list of flow IDs to delete.
-        user (User, optional): The user making the request. Defaults to the current active user.
-        db (Session, optional): The database session.
-
-    Returns:
-        dict: A dictionary containing the number of flows deleted.
-
-    """
+    """Delete multiple flows by their IDs."""
     try:
         flows_to_delete = (
             await db.exec(select(Flow).where(col(Flow.id).in_(flow_ids)).where(Flow.user_id == user.id))
@@ -834,7 +418,10 @@ async def delete_multiple_flows(
         await db.flush()
         return {"deleted": len(flows_to_delete)}
     except Exception as exc:
-        raise HTTPException(status_code=500, detail=str(exc)) from exc
+        import logging as _logging
+
+        _logging.getLogger(__name__).exception("Error deleting multiple flows")
+        raise HTTPException(status_code=500, detail="An internal error occurred while deleting flows.") from exc
 
 
 @router.post("/download/", status_code=200)
@@ -852,104 +439,61 @@ 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]
+    return _build_flows_download_response(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}_langflow_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]
-
-
-all_starter_folder_flows_response: Response | None = None
+# 5 minutes
+_STARTER_FLOWS_TTL_SECONDS: float = 300.0
+_starter_flows_cache: ThreadingInMemoryCache[threading.RLock] = ThreadingInMemoryCache(
+    max_size=1,
+    expiration_time=int(_STARTER_FLOWS_TTL_SECONDS),
+)
+_starter_flows_lock = asyncio.Lock()
 
 
 @router.get("/basic_examples/", response_model=list[FlowRead], status_code=200)
 async def read_basic_examples(
     *,
     session: DbSession,
 ):
-    """Retrieve a list of basic example flows.
-
-    Args:
-        session (Session): The database session.
+    """Retrieve a list of basic example flows."""
+    cached_response = _starter_flows_cache.get("starter_flows")
+    if cached_response is not CACHE_MISS:
+        return cached_response
 
-    Returns:
-        list[FlowRead]: A list of basic example flows.
-    """
-    try:
-        global all_starter_folder_flows_response  # noqa: PLW0603
+    async with _starter_flows_lock:
+        cached_response = _starter_flows_cache.get("starter_flows")
+        if cached_response is not CACHE_MISS:
+            return cached_response
 
-        if all_starter_folder_flows_response:
-            return all_starter_folder_flows_response
-        # Get the starter folder
-        starter_folder = (await session.exec(select(Folder).where(Folder.name == STARTER_FOLDER_NAME))).first()
+        try:
+            starter_folder = (await session.exec(select(Folder).where(Folder.name == STARTER_FOLDER_NAME))).first()
 
-        if not starter_folder:
-            return []
+            if not starter_folder:
+                return []
 
-        # Get all flows in the starter folder
-        all_starter_folder_flows = (await session.exec(select(Flow).where(Flow.folder_id == starter_folder.id))).all()
+            all_starter_folder_flows = (
+                await session.exec(select(Flow).where(Flow.folder_id == starter_folder.id))
+            ).all()
 
-        flow_reads = [FlowRead.model_validate(flow, from_attributes=True) for flow in all_starter_folder_flows]
-        all_starter_folder_flows_response = compress_response(flow_reads)
+            flow_reads = [FlowRead.model_validate(flow, from_attributes=True) for flow in all_starter_folder_flows]
+            response = compress_response(flow_reads)
+            _starter_flows_cache.set("starter_flows", response)
 
-        # Return compressed response using our utility function
-        return all_starter_folder_flows_response  # noqa: TRY300
+        except Exception as e:
+            import logging as _logging
 
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e)) from e
+            _logging.getLogger(__name__).exception("Error loading basic examples")
+            raise HTTPException(status_code=500, detail="An internal error occurred while loading examples.") from e
+        else:
+            return response
 
 
 @router.post("/expand/", status_code=200, dependencies=[Depends(get_current_active_user)], include_in_schema=False)
 async def expand_compact_flow_endpoint(
     compact_data: dict,
 ):
-    """Expand a compact flow format to full flow format.
-
-    This endpoint takes a minimal flow representation (as generated by AI agents)
-    and expands it to the full format expected by the Langflow UI.
-
-    The compact format only requires:
-    - nodes: list of {id, type, values?}
-    - edges: list of {source, source_output, target, target_input}
-
-    The endpoint returns the full flow data with complete component templates.
-
-    Example input:
-    ```json
-    {
-        "nodes": [
-            {"id": "1", "type": "ChatInput"},
-            {"id": "2", "type": "OpenAIModel", "values": {"model_name": "gpt-4"}}
-        ],
-        "edges": [
-            {"source": "1", "source_output": "message", "target": "2", "target_input": "input_value"}
-        ]
-    }
-    ```
-    """
+    """Expand a compact flow format (minimal nodes/edges) to the full flow format."""
     from lfx.interface.components import component_cache, get_and_cache_all_types_dict
 
     from langflow.processing.expand_flow import expand_compact_flow

@@ -0,0 +1,211 @@
+"""File upload and download handlers for projects.
+
+Extracted from projects.py to reduce file size and separate file I/O concerns.
+"""
+
+import io
+import zipfile
+from datetime import datetime, timezone
+from typing import Annotated
+from urllib.parse import quote
+from uuid import UUID
+
+import orjson
+from fastapi import File, HTTPException, UploadFile
+from fastapi.responses import StreamingResponse
+from lfx.log.logger import logger
+from sqlmodel import select
+
+from langflow.api.utils import (
+    CurrentActiveUser,
+    DbSession,
+    normalize_code_for_import,
+    normalize_flow_for_export,
+    remove_api_keys,
+)
+from langflow.api.utils.zip_utils import extract_flows_from_zip
+from langflow.api.v1.flows import create_flows
+from langflow.api.v1.flows_helpers import _sanitize_flow_filename
+from langflow.api.v1.schemas import FlowListCreate
+from langflow.helpers.flow import generate_unique_flow_name
+from langflow.helpers.folders import generate_unique_folder_name
+from langflow.services.auth.mcp_encryption import encrypt_auth_settings
+from langflow.services.database.models.base import orjson_dumps
+from langflow.services.database.models.flow.model import Flow, FlowCreate, FlowRead
+from langflow.services.database.models.folder.model import (
+    Folder,
+    FolderCreate,
+)
+from langflow.services.deps import get_settings_service
+
+
+async def download_project_flows(
+    *,
+    session: DbSession,
+    project_id: UUID,
+    current_user: CurrentActiveUser,
+) -> StreamingResponse:
+    """Download all flows from project as a zip file."""
+    try:
+        query = select(Folder).where(Folder.id == project_id, Folder.user_id == current_user.id)
+        result = await session.exec(query)
+        project = result.first()
+
+        if not project:
+            raise HTTPException(status_code=404, detail="Project not found")
+
+        flows_query = select(Flow).where(Flow.folder_id == project_id)
+        flows_result = await session.exec(flows_query)
+        flows = [FlowRead.model_validate(flow, from_attributes=True) for flow in flows_result.all()]
+
+        if not flows:
+            raise HTTPException(status_code=404, detail="No flows found in project")
+
+        # Strip API keys then normalise for git-friendly export (sorted keys,
+        # volatile fields removed, code fields as line arrays).
+        normalised_flows = [normalize_flow_for_export(remove_api_keys(flow.model_dump())) for flow in flows]
+        zip_stream = io.BytesIO()
+
+        with zipfile.ZipFile(zip_stream, "w") as zip_file:
+            for flow in normalised_flows:
+                safe_name = _sanitize_flow_filename(str(flow["name"]), str(flow.get("id", "flow")))
+                # Serialise with sorted keys and 2-space indent for stable diffs.
+                flow_json = orjson_dumps(flow, sort_keys=True)
+                zip_file.writestr(f"{safe_name}.json", flow_json.encode("utf-8"))
+
+        zip_stream.seek(0)
+
+        current_time = datetime.now(tz=timezone.utc).astimezone().strftime("%Y%m%d_%H%M%S")
+        filename = f"{current_time}_{project.name}_flows.zip"
+
+        # URL encode filename handle non-ASCII (ex. Cyrillic)
+        encoded_filename = quote(filename)
+
+        return StreamingResponse(
+            zip_stream,
+            media_type="application/x-zip-compressed",
+            headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}"},
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        if "No result found" in str(e):
+            raise HTTPException(status_code=404, detail="Project not found") from e
+        logger.exception("Error downloading project flows for project_id=%s", project_id)
+        raise HTTPException(
+            status_code=500, detail="An internal error occurred while downloading project flows."
+        ) from e
+
+
+async def upload_project_flows(
+    *,
+    session: DbSession,
+    file: Annotated[UploadFile | None, File()] = None,
+    current_user: CurrentActiveUser,
+) -> list[FlowRead]:
+    """Upload flows from a file.
+
+    Accepts either a JSON file with project metadata (folder_name, folder_description, flows)
+    or a ZIP file containing individual flow JSON files (as produced by the download endpoint).
+    """
+    if file is None:
+        raise HTTPException(status_code=400, detail="No file provided")
+
+    contents = await file.read()
+
+    if not contents:
+        raise HTTPException(status_code=400, detail="The uploaded file is empty")
+
+    # Detect ZIP files and extract flow data
+    if zipfile.is_zipfile(io.BytesIO(contents)):
+        try:
+            flows_data = await extract_flows_from_zip(contents)
+        except ValueError as e:
+            raise HTTPException(status_code=400, detail=str(e)) from e
+        if not flows_data:
+            raise HTTPException(status_code=400, detail="No valid flow JSON files found in the ZIP")
+
+        # Use the uploaded filename (without extension) as the project name
+        project_name_base = file.filename.rsplit(".", 1)[0] if file.filename else "Imported Project"
+        project_name_base = project_name_base or "Imported Project"
+        data: dict = {
+            "folder_name": project_name_base,
+            "folder_description": "",
+            "flows": flows_data,
+        }
+    else:
+        try:
+            data = orjson.loads(contents)
+        except orjson.JSONDecodeError as e:
+            raise HTTPException(status_code=400, detail=f"Invalid JSON file: {e}") from e
+
+    if not data:
+        raise HTTPException(status_code=400, detail="No flows found in the file")
+
+    # Validate that the uploaded JSON has the required structure before accessing keys
+    if not isinstance(data, dict):
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid project data: expected a JSON object with 'folder_name' and 'flows' fields",
+        )
+
+    missing_keys = [key for key in ("folder_name", "flows") if key not in data]
+    if missing_keys:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Missing required field(s): {', '.join(missing_keys)}",
+        )
+    project_name = await generate_unique_folder_name(data["folder_name"], current_user.id, session)
+
+    data["folder_name"] = project_name
+
+    project = FolderCreate(name=data["folder_name"], description=data.get("folder_description", ""))
+
+    new_project = Folder.model_validate(project, from_attributes=True)
+    new_project.id = None
+    new_project.user_id = current_user.id
+
+    settings_service = get_settings_service()
+
+    # If AUTO_LOGIN is false, automatically enable API key authentication
+    if not settings_service.auth_settings.AUTO_LOGIN and not new_project.auth_settings:
+        default_auth = {"auth_type": "apikey"}
+        new_project.auth_settings = encrypt_auth_settings(default_auth)
+        await logger.adebug(
+            "Auto-enabled API key authentication for uploaded project %s (%s) due to AUTO_LOGIN=false",
+            new_project.name,
+            new_project.id,
+        )
+
+    session.add(new_project)
+    await session.flush()
+    await session.refresh(new_project)
+    del data["folder_name"]
+    data.pop("folder_description", None)
+
+    if "flows" in data:
+        # Normalise code fields: if exported with code-as-lines format, rejoin to
+        # strings before creating Pydantic models so the DB always stores strings.
+        flow_list = FlowListCreate(flows=[FlowCreate(**normalize_code_for_import(flow)) for flow in data["flows"]])
+    else:
+        raise HTTPException(status_code=400, detail="No flows found in the data")
+    # Generate unique names, tracking names already assigned within this batch
+    # to avoid collisions when multiple flows would get the same generated name
+    used_names_in_batch: set[str] = set()
+    for flow in flow_list.flows:
+        flow_name = await generate_unique_flow_name(flow.name, current_user.id, session)
+        # Ensure the name is also unique within the current batch;
+        # generate suffixed candidates and verify each against DB
+        base_name = flow_name
+        n = 1
+        while flow_name in used_names_in_batch:
+            candidate = f"{base_name} ({n})"
+            n += 1
+            flow_name = await generate_unique_flow_name(candidate, current_user.id, session)
+        used_names_in_batch.add(flow_name)
+        flow.name = flow_name
+        flow.user_id = current_user.id
+        flow.folder_id = new_project.id
+
+    return await create_flows(session=session, flow_list=flow_list, current_user=current_user)

@@ -0,0 +1,246 @@
+"""MCP server registration and management helpers for projects.
+
+Extracted from projects.py to reduce file size and isolate MCP concerns (SO1).
+"""
+
+from typing import cast
+from uuid import UUID
+
+from fastapi import HTTPException
+from lfx.log.logger import logger
+from lfx.services.mcp_composer.service import MCPComposerService
+
+from langflow.api.utils.mcp.config_utils import validate_mcp_server_for_project
+from langflow.api.v1.mcp_projects import get_project_streamable_http_url
+from langflow.api.v2.mcp import update_server
+from langflow.services.database.models.api_key.crud import create_api_key
+from langflow.services.database.models.api_key.model import ApiKeyCreate
+from langflow.services.deps import get_service, get_settings_service, get_storage_service
+from langflow.services.schema import ServiceType
+
+
+async def register_mcp_servers_for_project(
+    project,
+    default_auth: dict,
+    current_user,
+    session,
+) -> None:
+    """Register MCP servers for a newly created project.
+
+    This handles the full MCP auto-registration flow: building the transport URL,
+    creating API keys if needed, validating conflicts, and calling update_server.
+
+    Raises HTTPException on conflicts or unsupported auth types.
+    """
+    try:
+        streamable_http_url = await get_project_streamable_http_url(project.id)
+
+        if default_auth.get("auth_type", "none") == "apikey":
+            api_key_name = f"MCP Project {project.name} - default"
+            unmasked_api_key = await create_api_key(session, ApiKeyCreate(name=api_key_name), current_user.id)
+            command = "uvx"
+            args = [
+                "mcp-proxy",
+                "--transport",
+                "streamablehttp",
+                "--headers",
+                "x-api-key",
+                unmasked_api_key.api_key,
+                streamable_http_url,
+            ]
+        elif default_auth.get("auth_type", "none") == "oauth":
+            msg = "OAuth authentication is not yet implemented for MCP server creation during project creation."
+            await logger.awarning(msg)
+            return
+        else:
+            command = "uvx"
+            args = [
+                "mcp-proxy",
+                "--transport",
+                "streamablehttp",
+                streamable_http_url,
+            ]
+
+        server_config = {"command": command, "args": args}
+
+        validation_result = await validate_mcp_server_for_project(
+            project.id,
+            project.name,
+            current_user,
+            session,
+            get_storage_service(),
+            get_settings_service(),
+            operation="create",
+        )
+
+        if validation_result.has_conflict:
+            await logger.aerror(validation_result.conflict_message)
+            raise HTTPException(
+                status_code=409,
+                detail=validation_result.conflict_message,
+            )
+
+        if validation_result.should_skip:
+            await logger.adebug(
+                "MCP server '%s' already exists for project %s, updating",
+                validation_result.server_name,
+                project.id,
+            )
+
+        server_name = validation_result.server_name
+
+        await update_server(
+            server_name,
+            server_config,
+            current_user,
+            session,
+            get_storage_service(),
+            get_settings_service(),
+        )
+    except HTTPException:
+        raise
+    except Exception as e:  # noqa: BLE001
+        await logger.aexception("Failed to auto-register MCP server for project %s: %s", project.id, e)
+
+
+async def handle_mcp_server_rename(
+    existing_project,
+    old_project_name: str,
+    new_project_name: str,
+    current_user,
+    session,
+) -> None:
+    """Handle MCP server name update when a project is renamed.
+
+    Validates old and new server names, checks for conflicts, and performs
+    the rename (delete old + create new) if needed.
+
+    Raises HTTPException on name conflicts.
+    """
+    try:
+        old_validation = await validate_mcp_server_for_project(
+            existing_project.id,
+            old_project_name,
+            current_user,
+            session,
+            get_storage_service(),
+            get_settings_service(),
+            operation="update",
+        )
+
+        new_validation = await validate_mcp_server_for_project(
+            existing_project.id,
+            new_project_name,
+            current_user,
+            session,
+            get_storage_service(),
+            get_settings_service(),
+            operation="update",
+        )
+
+        if old_validation.server_name != new_validation.server_name:
+            if new_validation.has_conflict:
+                await logger.aerror(new_validation.conflict_message)
+                raise HTTPException(
+                    status_code=409,
+                    detail=new_validation.conflict_message,
+                )
+
+            if old_validation.server_exists and old_validation.project_id_matches:
+                await update_server(
+                    old_validation.server_name,
+                    {},
+                    current_user,
+                    session,
+                    get_storage_service(),
+                    get_settings_service(),
+                    delete=True,
+                )
+
+                await update_server(
+                    new_validation.server_name,
+                    old_validation.existing_config or {},
+                    current_user,
+                    session,
+                    get_storage_service(),
+                    get_settings_service(),
+                )
+
+                await logger.adebug(
+                    "Updated MCP server name from %s to %s",
+                    old_validation.server_name,
+                    new_validation.server_name,
+                )
+            else:
+                await logger.adebug(
+                    "Old MCP server '%s' not found for this project, skipping rename",
+                    old_validation.server_name,
+                )
+
+    except HTTPException:
+        raise
+    except Exception as e:  # noqa: BLE001
+        await logger.awarning("Failed to handle MCP server name update for project rename: %s", e)
+
+
+async def cleanup_mcp_on_delete(
+    project,
+    project_id: UUID,
+    current_user,
+    session,
+) -> None:
+    """Clean up MCP resources when a project is deleted.
+
+    Stops the MCP Composer if the project uses OAuth, and removes the
+    corresponding MCP server entry if auto-add was enabled.
+    """
+    # Stop MCP Composer if project used OAuth
+    if project.auth_settings and project.auth_settings.get("auth_type") == "oauth":
+        try:
+            mcp_composer_service: MCPComposerService = cast(
+                MCPComposerService, get_service(ServiceType.MCP_COMPOSER_SERVICE)
+            )
+            await mcp_composer_service.stop_project_composer(str(project_id))
+            await logger.adebug("Stopped MCP Composer for deleted OAuth project %s (%s)", project.name, project_id)
+        except Exception as e:  # noqa: BLE001
+            await logger.aerror("Failed to stop MCP Composer for deleted project %s: %s", project_id, e)
+
+    # Delete corresponding MCP server if auto-add was enabled
+    if get_settings_service().settings.add_projects_to_mcp_servers:
+        try:
+            validation_result = await validate_mcp_server_for_project(
+                project_id,
+                project.name,
+                current_user,
+                session,
+                get_storage_service(),
+                get_settings_service(),
+                operation="delete",
+            )
+
+            if validation_result.server_exists and validation_result.project_id_matches:
+                await update_server(
+                    validation_result.server_name,
+                    {},
+                    current_user,
+                    session,
+                    get_storage_service(),
+                    get_settings_service(),
+                    delete=True,
+                )
+                await logger.adebug(
+                    "Deleted MCP server %s for deleted project %s (%s)",
+                    validation_result.server_name,
+                    project.name,
+                    project_id,
+                )
+            elif validation_result.server_exists and not validation_result.project_id_matches:
+                await logger.adebug(
+                    "MCP server '%s' exists but belongs to different project, skipping deletion",
+                    validation_result.server_name,
+                )
+            else:
+                await logger.adebug("No MCP server found for deleted project %s (%s)", project.name, project_id)
+
+        except Exception as e:  # noqa: BLE001
+            await logger.awarning("Failed to handle MCP server cleanup for deleted project %s: %s", project_id, e)

@@ -1,15 +1,8 @@
-import io
-import json
-import zipfile
-from datetime import datetime, timezone
+import warnings
 from typing import Annotated, cast
-from urllib.parse import quote
 from uuid import UUID
 
-import orjson
 from fastapi import APIRouter, BackgroundTasks, Depends, File, HTTPException, Query, Response, UploadFile, status
-from fastapi.encoders import jsonable_encoder
-from fastapi.responses import StreamingResponse
 from fastapi_pagination import Params
 from fastapi_pagination.ext.sqlmodel import apaginate
 from lfx.log.logger import logger
@@ -18,25 +11,23 @@
 from sqlalchemy.orm import selectinload
 from sqlmodel import select
 
-from langflow.api.utils import CurrentActiveUser, DbSession, cascade_delete_flow, custom_params, remove_api_keys
-from langflow.api.utils.mcp.config_utils import validate_mcp_server_for_project
-from langflow.api.utils.zip_utils import extract_flows_from_zip
+from langflow.api.utils import (
+    CurrentActiveUser,
+    DbSession,
+    cascade_delete_flow,
+    custom_params,
+)
 from langflow.api.v1.auth_helpers import handle_auth_settings_update
-from langflow.api.v1.flows import create_flows
-from langflow.api.v1.mcp_projects import (
-    get_project_sse_url,  # noqa: F401
-    get_project_streamable_http_url,
-    register_project_with_composer,
+from langflow.api.v1.mcp_projects import register_project_with_composer
+from langflow.api.v1.projects_files import download_project_flows, upload_project_flows
+from langflow.api.v1.projects_mcp_helpers import (
+    cleanup_mcp_on_delete,
+    handle_mcp_server_rename,
+    register_mcp_servers_for_project,
 )
-from langflow.api.v1.schemas import FlowListCreate
-from langflow.api.v2.mcp import update_server
-from langflow.helpers.flow import generate_unique_flow_name
-from langflow.helpers.folders import generate_unique_folder_name
 from langflow.initial_setup.constants import ASSISTANT_FOLDER_NAME, STARTER_FOLDER_NAME
 from langflow.services.auth.mcp_encryption import encrypt_auth_settings
-from langflow.services.database.models.api_key.crud import create_api_key
-from langflow.services.database.models.api_key.model import ApiKeyCreate
-from langflow.services.database.models.flow.model import Flow, FlowCreate, FlowRead
+from langflow.services.database.models.flow.model import Flow, FlowRead
 from langflow.services.database.models.folder.constants import DEFAULT_FOLDER_NAME
 from langflow.services.database.models.folder.model import (
     Folder,
@@ -46,12 +37,17 @@
     FolderUpdate,
 )
 from langflow.services.database.models.folder.pagination_model import FolderWithPaginatedFlows
-from langflow.services.deps import get_service, get_settings_service, get_storage_service
+from langflow.services.deps import get_service, get_settings_service
 from langflow.services.schema import ServiceType
 
 router = APIRouter(prefix="/projects", tags=["Projects"])
 
 
+def _escape_like(value: str) -> str:
+    """Escape LIKE wildcards and the escape character itself."""
+    return value.replace("\\", "\\\\").replace("%", r"\%").replace("_", r"\_")
+
+
 @router.post("/", response_model=FolderRead, status_code=201)
 async def create_project(
     *,
@@ -72,31 +68,42 @@ async def create_project(
                 statement=select(Folder).where(Folder.name == new_project.name).where(Folder.user_id == current_user.id)
             )
         ).first():
+            escaped_project_name = _escape_like(new_project.name)
             project_results = await session.exec(
                 select(Folder).where(
-                    Folder.name.like(f"{new_project.name}%"),  # type: ignore[attr-defined]
+                    Folder.name.like(f"{escaped_project_name}%", escape="\\"),  # type: ignore[attr-defined]
                     Folder.user_id == current_user.id,
                 )
             )
             if project_results:
                 project_names = [project.name for project in project_results]
-                # TODO: this throws an error if the name contains non-numeric content in parentheses
-                project_numbers = [int(name.split("(")[-1].split(")")[0]) for name in project_names if "(" in name]
+                project_numbers = []
+                for name in project_names:
+                    if "(" not in name:
+                        continue
+                    try:
+                        project_numbers.append(int(name.split("(")[-1].split(")")[0]))
+                    except ValueError:
+                        continue
                 if project_numbers:
                     new_project.name = f"{new_project.name} ({max(project_numbers) + 1})"
                 else:
                     new_project.name = f"{new_project.name} (1)"
 
         settings_service = get_settings_service()
+        mcp_auth: dict = {"auth_type": "none"}
 
+        if project.auth_settings:
+            mcp_auth = project.auth_settings.copy()
+            new_project.auth_settings = encrypt_auth_settings(mcp_auth)
         # If AUTO_LOGIN is false, automatically enable API key authentication
-        default_auth = {"auth_type": "none"}
-        if not settings_service.auth_settings.AUTO_LOGIN and not new_project.auth_settings:
-            default_auth = {"auth_type": "apikey"}
-            new_project.auth_settings = encrypt_auth_settings(default_auth)
+        elif not settings_service.auth_settings.AUTO_LOGIN:
+            mcp_auth = {"auth_type": "apikey"}
+            new_project.auth_settings = encrypt_auth_settings(mcp_auth)
             await logger.adebug(
-                f"Auto-enabled API key authentication for project {new_project.name} "
-                f"({new_project.id}) due to AUTO_LOGIN=false"
+                "Auto-enabled API key authentication for project %s (%s) due to AUTO_LOGIN=false",
+                new_project.name,
+                new_project.id,
             )
 
         session.add(new_project)
@@ -105,91 +112,7 @@ async def create_project(
 
         # Auto-register MCP server for this project with configured default auth
         if get_settings_service().settings.add_projects_to_mcp_servers:
-            try:
-                # Build Streamable HTTP URL (preferred transport) and legacy SSE URL (for docs/errors)
-                streamable_http_url = await get_project_streamable_http_url(new_project.id)
-                # legacy SSE URL
-                # sse_url = await get_project_sse_url(new_project.id)
-
-                # Prepare server config based on auth type same as new project
-                if default_auth.get("auth_type", "none") == "apikey":
-                    # Create API key for API key authentication
-                    api_key_name = f"MCP Project {new_project.name} - default"
-                    unmasked_api_key = await create_api_key(session, ApiKeyCreate(name=api_key_name), current_user.id)
-                    # Starting v>=1.7.1, we use Streamable HTTP transport by default
-                    command = "uvx"
-                    args = [
-                        "mcp-proxy",
-                        "--transport",
-                        "streamablehttp",
-                        "--headers",
-                        "x-api-key",
-                        unmasked_api_key.api_key,
-                        streamable_http_url,
-                    ]
-                elif default_auth.get("auth_type", "none") == "oauth":
-                    msg = "OAuth authentication is not yet implemented for MCP server creation during project creation."
-                    logger.warning(msg)
-                    raise HTTPException(status_code=501, detail=msg)
-                else:  # default_auth_type == "none"
-                    # No authentication - direct connection
-                    command = "uvx"
-                    args = [
-                        "mcp-proxy",
-                        "--transport",
-                        "streamablehttp",
-                        streamable_http_url,
-                    ]
-
-                server_config = {"command": command, "args": args}
-
-                # Validate MCP server for this project
-                validation_result = await validate_mcp_server_for_project(
-                    new_project.id,
-                    new_project.name,
-                    current_user,
-                    session,
-                    get_storage_service(),
-                    get_settings_service(),
-                    operation="create",
-                )
-
-                # Handle conflicts
-                if validation_result.has_conflict:
-                    await logger.aerror(validation_result.conflict_message)
-                    raise HTTPException(
-                        status_code=409,  # Conflict - semantically correct for name conflicts
-                        detail=validation_result.conflict_message,
-                    )
-
-                # Log if updating existing server
-                if validation_result.should_skip:
-                    msg = (
-                        f"MCP server '{validation_result.server_name}' "
-                        f"already exists for project {new_project.id}, updating"
-                    )
-                    await logger.adebug(msg)
-
-                server_name = validation_result.server_name
-
-                await update_server(
-                    server_name,
-                    server_config,
-                    current_user,
-                    session,
-                    get_storage_service(),
-                    get_settings_service(),
-                )
-            except HTTPException:
-                # Re-raise HTTP validation errors (conflicts, etc.)
-                raise
-            except NotImplementedError:
-                msg = "OAuth as default MCP authentication type is not yet implemented"
-                await logger.aerror(msg)
-                raise
-            except Exception as e:  # noqa: BLE001
-                msg = f"Failed to auto-register MCP server for project {new_project.id}: {e}"
-                await logger.aexception(msg, exc_info=True)
+            await register_mcp_servers_for_project(new_project, mcp_auth, current_user, session)
 
         if project.components_list:
             update_statement_components = (
@@ -278,9 +201,8 @@ async def read_project(
             if is_flow:
                 stmt = stmt.where(Flow.is_component == False)  # noqa: E712
             if search:
-                stmt = stmt.where(Flow.name.like(f"%{search}%"))  # type: ignore[attr-defined]
-
-            import warnings
+                _search = _escape_like(search)
+                stmt = stmt.where(Flow.name.like(f"%{_search}%", escape="\\"))  # type: ignore[attr-defined]
 
             with warnings.catch_warnings():
                 warnings.filterwarnings(
@@ -343,88 +265,13 @@ async def update_project(
             should_start_mcp_composer = auth_result["should_start_composer"]
             should_stop_mcp_composer = auth_result["should_stop_composer"]
 
-        # Handle other updates
+        # Handle project rename and corresponding MCP server rename
         if project.name and project.name != existing_project.name:
             old_project_name = existing_project.name
             existing_project.name = project.name
 
-            # Update corresponding MCP server name if auto-add is enabled
             if get_settings_service().settings.add_projects_to_mcp_servers:
-                try:
-                    # Validate old server (for this specific project)
-                    old_validation = await validate_mcp_server_for_project(
-                        existing_project.id,
-                        old_project_name,
-                        current_user,
-                        session,
-                        get_storage_service(),
-                        get_settings_service(),
-                        operation="update",
-                    )
-
-                    # Validate new server name (check for conflicts)
-                    new_validation = await validate_mcp_server_for_project(
-                        existing_project.id,
-                        project.name,
-                        current_user,
-                        session,
-                        get_storage_service(),
-                        get_settings_service(),
-                        operation="update",
-                    )
-
-                    # Only proceed if server names would be different
-                    if old_validation.server_name != new_validation.server_name:
-                        # Check if new server name would conflict with different project
-                        if new_validation.has_conflict:
-                            await logger.aerror(new_validation.conflict_message)
-                            raise HTTPException(
-                                status_code=409,  # Conflict - semantically correct for name conflicts
-                                detail=new_validation.conflict_message,
-                            )
-
-                        # If old server exists and matches this project, proceed with rename
-                        if old_validation.server_exists and old_validation.project_id_matches:
-                            # Remove the old server
-                            await update_server(
-                                old_validation.server_name,
-                                {},  # Empty config for deletion
-                                current_user,
-                                session,
-                                get_storage_service(),
-                                get_settings_service(),
-                                delete=True,
-                            )
-
-                            # Add server with new name and existing config
-                            await update_server(
-                                new_validation.server_name,
-                                old_validation.existing_config or {},
-                                current_user,
-                                session,
-                                get_storage_service(),
-                                get_settings_service(),
-                            )
-
-                            msg = (
-                                f"Updated MCP server name from {old_validation.server_name} "
-                                f"to {new_validation.server_name}"
-                            )
-                            await logger.adebug(msg)
-                        else:
-                            msg = (
-                                f"Old MCP server '{old_validation.server_name}' "
-                                "not found for this project, skipping rename"
-                            )
-                            await logger.adebug(msg)
-
-                except HTTPException:
-                    # Re-raise HTTP validation errors (conflicts, etc.)
-                    raise
-                except Exception as e:  # noqa: BLE001
-                    # Log but don't fail the project update if MCP server handling fails
-                    msg = f"Failed to handle MCP server name update for project rename: {e}"
-                    await logger.awarning(msg)
+                await handle_mcp_server_rename(existing_project, old_project_name, project.name, current_user, session)
 
         if project.description is not None:
             existing_project.description = project.description
@@ -438,22 +285,21 @@ async def update_project(
 
         # Start MCP Composer if auth changed to OAuth
         if should_start_mcp_composer:
-            msg = (
-                f"Auth settings changed to OAuth for project {existing_project.name} ({existing_project.id}), "
-                "starting MCP Composer"
+            await logger.adebug(
+                "Auth settings changed to OAuth for project %s (%s), starting MCP Composer",
+                existing_project.name,
+                existing_project.id,
             )
-            await logger.adebug(msg)
             background_tasks.add_task(register_project_with_composer, existing_project)
 
         # Stop MCP Composer if auth changed FROM OAuth to something else
         elif should_stop_mcp_composer:
-            msg = (
-                f"Auth settings changed from OAuth for project {existing_project.name} ({existing_project.id}), "
-                "stopping MCP Composer"
+            await logger.ainfo(
+                "Auth settings changed from OAuth for project %s (%s), stopping MCP Composer",
+                existing_project.name,
+                existing_project.id,
             )
-            await logger.ainfo(msg)
 
-            # Get the MCP Composer service and stop the project's composer
             mcp_composer_service: MCPComposerService = cast(
                 MCPComposerService, get_service(ServiceType.MCP_COMPOSER_SERVICE)
             )
@@ -517,67 +363,13 @@ async def delete_project(
     # Prevent deletion of the Langflow Assistant folder
     if project.name == ASSISTANT_FOLDER_NAME:
         msg = f"Cannot delete the '{ASSISTANT_FOLDER_NAME}' folder, that contains pre-built flows."
-        await logger.adebug(msg)
+        await logger.adebug("Cannot delete the '%s' folder, that contains pre-built flows.", ASSISTANT_FOLDER_NAME)
         raise HTTPException(
             status_code=403,
             detail=msg,
         )
 
-    # Check if project has OAuth authentication and stop MCP Composer if needed
-    if project.auth_settings and project.auth_settings.get("auth_type") == "oauth":
-        try:
-            mcp_composer_service: MCPComposerService = cast(
-                MCPComposerService, get_service(ServiceType.MCP_COMPOSER_SERVICE)
-            )
-            await mcp_composer_service.stop_project_composer(str(project_id))
-            await logger.adebug(f"Stopped MCP Composer for deleted OAuth project {project.name} ({project_id})")
-        except Exception as e:  # noqa: BLE001
-            # Log but don't fail the deletion if MCP Composer cleanup fails
-            await logger.aerror(f"Failed to stop MCP Composer for deleted project {project_id}: {e}")
-
-    # Delete corresponding MCP server if auto-add was enabled
-    if get_settings_service().settings.add_projects_to_mcp_servers:
-        try:
-            # Validate MCP server for this specific project
-            validation_result = await validate_mcp_server_for_project(
-                project_id,
-                project.name,
-                current_user,
-                session,
-                get_storage_service(),
-                get_settings_service(),
-                operation="delete",
-            )
-
-            # Only delete if server exists and matches this project ID
-            if validation_result.server_exists and validation_result.project_id_matches:
-                await update_server(
-                    validation_result.server_name,
-                    {},  # Empty config for deletion
-                    current_user,
-                    session,
-                    get_storage_service(),
-                    get_settings_service(),
-                    delete=True,
-                )
-                msg = (
-                    f"Deleted MCP server {validation_result.server_name} for "
-                    f"deleted project {project.name} ({project_id})"
-                )
-                await logger.adebug(msg)
-            elif validation_result.server_exists and not validation_result.project_id_matches:
-                msg = (
-                    f"MCP server '{validation_result.server_name}' exists but belongs to different project, "
-                    "skipping deletion"
-                )
-                await logger.adebug(msg)
-            else:
-                msg = f"No MCP server found for deleted project {project.name} ({project_id})"
-                await logger.adebug(msg)
-
-        except Exception as e:  # noqa: BLE001
-            # Log but don't fail the project deletion if MCP server handling fails
-            await logger.awarning(f"Failed to handle MCP server cleanup for deleted project {project_id}: {e}")
+    await cleanup_mcp_on_delete(project, project_id, current_user, session)
 
     try:
         await session.delete(project)
@@ -594,47 +386,7 @@ async def download_file(
     current_user: CurrentActiveUser,
 ):
     """Download all flows from project as a zip file."""
-    try:
-        query = select(Folder).where(Folder.id == project_id, Folder.user_id == current_user.id)
-        result = await session.exec(query)
-        project = result.first()
-
-        if not project:
-            raise HTTPException(status_code=404, detail="Project not found")
-
-        flows_query = select(Flow).where(Flow.folder_id == project_id)
-        flows_result = await session.exec(flows_query)
-        flows = [FlowRead.model_validate(flow, from_attributes=True) for flow in flows_result.all()]
-
-        if not flows:
-            raise HTTPException(status_code=404, detail="No flows found in project")
-
-        flows_without_api_keys = [remove_api_keys(flow.model_dump()) for flow in flows]
-        zip_stream = io.BytesIO()
-
-        with zipfile.ZipFile(zip_stream, "w") as zip_file:
-            for flow in flows_without_api_keys:
-                flow_json = json.dumps(jsonable_encoder(flow))
-                zip_file.writestr(f"{flow['name']}.json", flow_json.encode("utf-8"))
-
-        zip_stream.seek(0)
-
-        current_time = datetime.now(tz=timezone.utc).astimezone().strftime("%Y%m%d_%H%M%S")
-        filename = f"{current_time}_{project.name}_flows.zip"
-
-        # URL encode filename handle non-ASCII (ex. Cyrillic)
-        encoded_filename = quote(filename)
-
-        return StreamingResponse(
-            zip_stream,
-            media_type="application/x-zip-compressed",
-            headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}"},
-        )
-
-    except Exception as e:
-        if "No result found" in str(e):
-            raise HTTPException(status_code=404, detail="Project not found") from e
-        raise HTTPException(status_code=500, detail=str(e)) from e
+    return await download_project_flows(session=session, project_id=project_id, current_user=current_user)
 
 
 @router.post("/upload/", response_model=list[FlowRead], status_code=201)
@@ -649,87 +401,4 @@ async def upload_file(
     Accepts either a JSON file with project metadata (folder_name, folder_description, flows)
     or a ZIP file containing individual flow JSON files (as produced by the download endpoint).
     """
-    if file is None:
-        raise HTTPException(status_code=400, detail="No file provided")
-
-    contents = await file.read()
-
-    if not contents:
-        raise HTTPException(status_code=400, detail="The uploaded file is empty")
-
-    # Detect ZIP files and extract flow data
-    if zipfile.is_zipfile(io.BytesIO(contents)):
-        try:
-            flows_data = await extract_flows_from_zip(contents)
-        except ValueError as e:
-            raise HTTPException(status_code=400, detail=str(e)) from e
-        if not flows_data:
-            raise HTTPException(status_code=400, detail="No valid flow JSON files found in the ZIP")
-
-        # Use the uploaded filename (without extension) as the project name
-        project_name_base = file.filename.rsplit(".", 1)[0] if file.filename else "Imported Project"
-        project_name_base = project_name_base or "Imported Project"
-        data: dict = {
-            "folder_name": project_name_base,
-            "folder_description": "",
-            "flows": flows_data,
-        }
-    else:
-        try:
-            data = orjson.loads(contents)
-        except orjson.JSONDecodeError as e:
-            raise HTTPException(status_code=400, detail=f"Invalid JSON file: {e}") from e
-
-    if not data:
-        raise HTTPException(status_code=400, detail="No flows found in the file")
-
-    project_name = await generate_unique_folder_name(data["folder_name"], current_user.id, session)
-
-    data["folder_name"] = project_name
-
-    project = FolderCreate(name=data["folder_name"], description=data.get("folder_description", ""))
-
-    new_project = Folder.model_validate(project, from_attributes=True)
-    new_project.id = None
-    new_project.user_id = current_user.id
-
-    settings_service = get_settings_service()
-
-    # If AUTO_LOGIN is false, automatically enable API key authentication
-    if not settings_service.auth_settings.AUTO_LOGIN and not new_project.auth_settings:
-        default_auth = {"auth_type": "apikey"}
-        new_project.auth_settings = encrypt_auth_settings(default_auth)
-        await logger.adebug(
-            f"Auto-enabled API key authentication for uploaded project {new_project.name} "
-            f"({new_project.id}) due to AUTO_LOGIN=false"
-        )
-
-    session.add(new_project)
-    await session.flush()
-    await session.refresh(new_project)
-    del data["folder_name"]
-    data.pop("folder_description", None)
-
-    if "flows" in data:
-        flow_list = FlowListCreate(flows=[FlowCreate(**flow) for flow in data["flows"]])
-    else:
-        raise HTTPException(status_code=400, detail="No flows found in the data")
-    # Generate unique names, tracking names already assigned within this batch
-    # to avoid collisions when multiple flows would get the same generated name
-    used_names_in_batch: set[str] = set()
-    for flow in flow_list.flows:
-        flow_name = await generate_unique_flow_name(flow.name, current_user.id, session)
-        # Ensure the name is also unique within the current batch;
-        # generate suffixed candidates and verify each against DB
-        base_name = flow_name
-        n = 1
-        while flow_name in used_names_in_batch:
-            candidate = f"{base_name} ({n})"
-            n += 1
-            flow_name = await generate_unique_flow_name(candidate, current_user.id, session)
-        used_names_in_batch.add(flow_name)
-        flow.name = flow_name
-        flow.user_id = current_user.id
-        flow.folder_id = new_project.id
-
-    return await create_flows(session=session, flow_list=flow_list, current_user=current_user)
+    return await upload_project_flows(session=session, file=file, current_user=current_user)

@@ -719,7 +719,7 @@ def create_new_project(
         gradient=project_gradient,
         tags=project_tags,
     )
-    db_flow = Flow.model_validate(new_project, from_attributes=True)
+    db_flow = Flow.model_validate(new_project.model_dump(exclude={"id"}))
     session.add(db_flow)
 
 
@@ -895,7 +895,7 @@ async def create_or_update_agentic_flows(session: AsyncSession, user_id: UUID) -
                         tags=flow_tags,
                         endpoint_name=flow_endpoint_name,  # Set endpoint_name from JSON
                     )
-                    db_flow = Flow.model_validate(new_project, from_attributes=True)
+                    db_flow = Flow.model_validate(new_project.model_dump(exclude={"id"}))
 
                     # Set the ID from JSON if provided
                     if flow_id:

@@ -8,7 +8,6 @@
 
 import emoji
 from emoji import purely_emoji
-from fastapi import HTTPException, status
 from lfx.log.logger import logger
 from pydantic import BaseModel, ValidationInfo, field_serializer, field_validator
 from sqlalchemy import Enum as SQLEnum
@@ -23,6 +22,24 @@
 
 HEX_COLOR_LENGTH = 7
 
+_ENDPOINT_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]+$")
+
+
+def _validate_endpoint_name_value(v: str | None) -> str | None:
+    """Validate that an endpoint name contains only safe URL characters.
+
+    Raises ``ValueError`` on invalid input — callers at the HTTP layer
+    should catch and translate to an appropriate HTTP response.
+    """
+    if v is not None:
+        if not isinstance(v, str):
+            msg = "Endpoint name must be a string"
+            raise ValueError(msg)
+        if not _ENDPOINT_NAME_RE.match(v):
+            msg = "Endpoint name must contain only letters, numbers, hyphens, and underscores"
+            raise ValueError(msg)
+    return v
+
 
 class AccessTypeEnum(str, Enum):
     PRIVATE = "PRIVATE"
@@ -70,19 +87,7 @@ class FlowBase(SQLModel):
     @field_validator("endpoint_name")
     @classmethod
     def validate_endpoint_name(cls, v):
-        # Endpoint name must be a string containing only letters, numbers, hyphens, and underscores
-        if v is not None:
-            if not isinstance(v, str):
-                raise HTTPException(
-                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-                    detail="Endpoint name must be a string",
-                )
-            if not re.match(r"^[a-zA-Z0-9_-]+$", v):
-                raise HTTPException(
-                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-                    detail="Endpoint name must contain only letters, numbers, hyphens, and underscores",
-                )
-        return v
+        return _validate_endpoint_name_value(v)
 
     @field_validator("icon_bg_color")
     @classmethod
@@ -122,7 +127,7 @@ def validate_icon_atr(cls, v):
 
         emoji_value = emoji.emojize(v, variant="emoji_type")
         if v == emoji_value:
-            logger.warning(f"Invalid emoji. {v} is not a valid emoji.")
+            logger.warning("Invalid emoji. %s is not a valid emoji.", v)
         icon = emoji_value
 
         if purely_emoji(icon):
@@ -214,6 +219,11 @@ def to_data(self):
 
 
 class FlowCreate(FlowBase):
+    # Optional stable ID.  When present on upload, the flow is upserted
+    # (created with that ID, or updated if the ID already belongs to the
+    # current user).  Flows without an id get a generated UUID — backward
+    # compatible with all existing import paths.
+    id: UUID | None = None
     user_id: UUID | None = None
     folder_id: UUID | None = None
     fs_path: str | None = None
@@ -269,16 +279,4 @@ class FlowUpdate(SQLModel):
     @field_validator("endpoint_name")
     @classmethod
     def validate_endpoint_name(cls, v):
-        # Endpoint name must be a string containing only letters, numbers, hyphens, and underscores
-        if v is not None:
-            if not isinstance(v, str):
-                raise HTTPException(
-                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-                    detail="Endpoint name must be a string",
-                )
-            if not re.match(r"^[a-zA-Z0-9_-]+$", v):
-                raise HTTPException(
-                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-                    detail="Endpoint name must contain only letters, numbers, hyphens, and underscores",
-                )
-        return v
+        return _validate_endpoint_name_value(v)

@@ -577,7 +577,7 @@ async def flow(
     loaded_json = json.loads(json_flow)
     flow_data = FlowCreate(name="test_flow", data=loaded_json.get("data"), user_id=active_user.id)
 
-    flow = Flow.model_validate(flow_data)
+    flow = Flow.model_validate(flow_data.model_dump(exclude={"id"}))
     async with session_scope() as session:
         session.add(flow)
         await session.flush()

@@ -0,0 +1,263 @@
+"""Unit tests for normalize_flow_for_export / normalize_code_for_import.
+
+These tests run entirely in-process — no database or HTTP server required.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+
+from langflow.api.utils.core import normalize_code_for_import, normalize_flow_for_export
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+# split("\n") keeps trailing newlines as a final empty string, making the
+# round-trip lossless.  splitlines() would drop the trailing "\n".
+_CODE_STRING = "def build(self):\n    return self.input\n"
+_CODE_LINES = ["def build(self):", "    return self.input", ""]
+
+
+def _make_flow(*, extra_top: dict | None = None, code_value: str | list | None = _CODE_STRING) -> dict:
+    """Return a minimal flow dict that exercises all normalisation paths."""
+    flow: dict = {
+        "id": "17050493-96dd-4dc9-ba42-4bd0075fd23d",
+        "name": "Test Flow",
+        "description": "A test flow",
+        "data": {
+            "nodes": [
+                {
+                    "id": "node-1",
+                    "position": {"x": 100, "y": 200},
+                    "positionAbsolute": {"x": 100, "y": 200},
+                    "dragging": False,
+                    "selected": True,
+                    "data": {
+                        "id": "node-1",
+                        "type": "PythonFunctionComponent",
+                        "node": {
+                            "template": {
+                                "code": {
+                                    "type": "code",
+                                    "value": code_value,
+                                    "name": "code",
+                                },
+                                "api_key": {
+                                    "type": "str",
+                                    "value": "sk-secret",
+                                    "name": "api_key",
+                                    "password": True,
+                                },
+                            }
+                        },
+                    },
+                }
+            ],
+            "edges": [],
+        },
+        "updated_at": "2024-01-01T12:00:00Z",
+        "created_at": "2023-06-15T08:00:00Z",
+        "user_id": "user-uuid-123",
+        "folder_id": "folder-uuid-456",
+        "access_type": "PRIVATE",
+        "gradient": "linear-gradient(to right, #f00, #00f)",
+        "is_component": False,
+        "endpoint_name": "test-flow",
+    }
+    if extra_top:
+        flow.update(extra_top)
+    return flow
+
+
+# ---------------------------------------------------------------------------
+# normalize_flow_for_export — volatile field stripping
+# ---------------------------------------------------------------------------
+
+
+class TestNormaliseVolatileFields:
+    def test_strips_updated_at(self):
+        result = normalize_flow_for_export(_make_flow())
+        assert "updated_at" not in result
+
+    def test_strips_created_at(self):
+        result = normalize_flow_for_export(_make_flow())
+        assert "created_at" not in result
+
+    def test_strips_user_id(self):
+        result = normalize_flow_for_export(_make_flow())
+        assert "user_id" not in result
+
+    def test_strips_folder_id(self):
+        result = normalize_flow_for_export(_make_flow())
+        assert "folder_id" not in result
+
+    def test_strips_access_type(self):
+        result = normalize_flow_for_export(_make_flow())
+        assert "access_type" not in result
+
+    def test_strips_gradient(self):
+        result = normalize_flow_for_export(_make_flow())
+        assert "gradient" not in result
+
+    def test_keeps_stable_fields(self):
+        result = normalize_flow_for_export(_make_flow())
+        assert result["id"] == "17050493-96dd-4dc9-ba42-4bd0075fd23d"
+        assert result["name"] == "Test Flow"
+        assert result["description"] == "A test flow"
+        assert result["endpoint_name"] == "test-flow"
+        assert result["is_component"] is False
+
+    def test_missing_volatile_fields_do_not_raise(self):
+        """Calling on a minimal flow without any volatile fields is safe."""
+        minimal = {"id": "abc", "name": "Minimal", "data": {"nodes": [], "edges": []}}
+        result = normalize_flow_for_export(minimal)
+        assert result["name"] == "Minimal"
+
+
+# ---------------------------------------------------------------------------
+# normalize_flow_for_export — node UI state stripping
+# ---------------------------------------------------------------------------
+
+
+class TestNormaliseNodeUiState:
+    def test_strips_position_absolute(self):
+        result = normalize_flow_for_export(_make_flow())
+        node = result["data"]["nodes"][0]
+        assert "positionAbsolute" not in node
+
+    def test_strips_dragging(self):
+        result = normalize_flow_for_export(_make_flow())
+        node = result["data"]["nodes"][0]
+        assert "dragging" not in node
+
+    def test_strips_selected(self):
+        result = normalize_flow_for_export(_make_flow())
+        node = result["data"]["nodes"][0]
+        assert "selected" not in node
+
+    def test_keeps_position(self):
+        """Position (canvas coords) is kept — only the derived positionAbsolute is stripped."""
+        result = normalize_flow_for_export(_make_flow())
+        node = result["data"]["nodes"][0]
+        assert node["position"] == {"x": 100, "y": 200}
+
+
+# ---------------------------------------------------------------------------
+# normalize_flow_for_export — code → list-of-lines
+# ---------------------------------------------------------------------------
+
+
+class TestNormaliseCodeToLines:
+    def test_code_field_becomes_list(self):
+        result = normalize_flow_for_export(_make_flow())
+        code_field = result["data"]["nodes"][0]["data"]["node"]["template"]["code"]
+        assert isinstance(code_field["value"], list)
+
+    def test_code_lines_content(self):
+        result = normalize_flow_for_export(_make_flow())
+        code_field = result["data"]["nodes"][0]["data"]["node"]["template"]["code"]
+        assert code_field["value"] == _CODE_LINES
+
+    def test_non_code_field_untouched(self):
+        result = normalize_flow_for_export(_make_flow())
+        api_key_field = result["data"]["nodes"][0]["data"]["node"]["template"]["api_key"]
+        assert api_key_field["value"] == "sk-secret"  # unchanged
+
+    def test_already_list_code_unchanged(self):
+        """If code is already a list (re-export scenario), it should stay as-is."""
+        result = normalize_flow_for_export(_make_flow(code_value=_CODE_LINES))
+        code_field = result["data"]["nodes"][0]["data"]["node"]["template"]["code"]
+        assert code_field["value"] == _CODE_LINES
+
+    def test_empty_code_string(self):
+        # "".split("\n") == [""], preserving the empty line
+        result = normalize_flow_for_export(_make_flow(code_value=""))
+        code_field = result["data"]["nodes"][0]["data"]["node"]["template"]["code"]
+        assert code_field["value"] == [""]  # split("\n") on "" gives [""]
+
+    def test_no_data_nodes_safe(self):
+        flow = {"id": "x", "name": "Empty", "data": {"nodes": [], "edges": []}}
+        # Must not raise
+        normalize_flow_for_export(flow)
+
+
+# ---------------------------------------------------------------------------
+# normalize_flow_for_export — does not mutate the original
+# ---------------------------------------------------------------------------
+
+
+class TestNormalisationIsNonMutating:
+    def test_original_unchanged(self):
+        original = _make_flow()
+        original_copy = copy.deepcopy(original)
+        normalize_flow_for_export(original)
+        assert original == original_copy
+
+
+# ---------------------------------------------------------------------------
+# normalize_code_for_import — list → string
+# ---------------------------------------------------------------------------
+
+
+class TestNormaliseCodeForImport:
+    def test_rejoins_list_to_string(self):
+        flow = _make_flow(code_value=_CODE_LINES)
+        result = normalize_code_for_import(flow)
+        code_field = result["data"]["nodes"][0]["data"]["node"]["template"]["code"]
+        assert isinstance(code_field["value"], str)
+        assert code_field["value"] == "\n".join(_CODE_LINES)
+
+    def test_string_format_passthrough(self):
+        """Legacy string format must survive import normalisation unchanged."""
+        flow = _make_flow(code_value=_CODE_STRING)
+        result = normalize_code_for_import(flow)
+        code_field = result["data"]["nodes"][0]["data"]["node"]["template"]["code"]
+        assert code_field["value"] == _CODE_STRING
+
+    def test_none_code_value_passthrough(self):
+        flow = _make_flow(code_value=None)
+        result = normalize_code_for_import(flow)
+        code_field = result["data"]["nodes"][0]["data"]["node"]["template"]["code"]
+        assert code_field["value"] is None
+
+    def test_original_unchanged(self):
+        flow = _make_flow(code_value=_CODE_LINES)
+        original_copy = copy.deepcopy(flow)
+        normalize_code_for_import(flow)
+        assert flow == original_copy
+
+
+# ---------------------------------------------------------------------------
+# Round-trip: export → import restores code strings
+# ---------------------------------------------------------------------------
+
+
+class TestRoundTrip:
+    def test_code_round_trips_to_identical_string(self):
+        original = _make_flow()
+        exported = normalize_flow_for_export(original)
+        imported = normalize_code_for_import(exported)
+        code_orig = original["data"]["nodes"][0]["data"]["node"]["template"]["code"]["value"]
+        code_trip = imported["data"]["nodes"][0]["data"]["node"]["template"]["code"]["value"]
+        assert code_trip == code_orig
+
+    def test_two_exports_byte_identical(self):
+        """Consecutive exports of the same flow must produce identical JSON bytes."""
+        import orjson
+
+        flow = _make_flow()
+        export1 = orjson.dumps(normalize_flow_for_export(flow), option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2)
+        export2 = orjson.dumps(normalize_flow_for_export(flow), option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2)
+        assert export1 == export2
+
+    def test_sorted_keys_top_level(self):
+        """All top-level keys must be in lexicographic order after export."""
+        import orjson
+
+        flow = _make_flow()
+        raw = orjson.dumps(normalize_flow_for_export(flow), option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2).decode()
+        parsed = json.loads(raw)
+        top_keys = list(parsed.keys())
+        assert top_keys == sorted(top_keys)

@@ -0,0 +1,208 @@
+"""Unit tests for stable flow-ID preservation on import/upload.
+
+Tests cover the FlowCreate schema change and the three upsert branches
+in the upload endpoint logic.  No database or HTTP server is required
+for the schema tests; the endpoint-logic tests use unittest.mock to
+stand in for the async DB session and storage service.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import UUID, uuid4
+
+import pytest
+from langflow.services.database.models.flow.model import FlowCreate
+
+# ---------------------------------------------------------------------------
+# FlowCreate schema
+# ---------------------------------------------------------------------------
+
+
+class TestFlowCreateSchema:
+    def test_id_field_is_optional(self):
+        """FlowCreate must accept construction without an id."""
+        flow = FlowCreate(name="No ID Flow")
+        assert flow.id is None
+
+    def test_id_field_accepts_uuid(self):
+        """FlowCreate must store a provided UUID verbatim."""
+        stable_id = uuid4()
+        flow = FlowCreate(name="With ID", id=stable_id)
+        assert flow.id == stable_id
+
+    def test_id_field_accepts_uuid_string(self):
+        """FlowCreate must coerce a UUID string to a UUID object."""
+        stable_id = uuid4()
+        flow = FlowCreate(name="With String ID", id=str(stable_id))
+        assert flow.id == stable_id
+
+    def test_id_from_dict_roundtrip(self):
+        """Simulates parsing a downloaded flow JSON that contains an id."""
+        stable_id = uuid4()
+        raw = {
+            "id": str(stable_id),
+            "name": "Exported Flow",
+            "data": {"nodes": [], "edges": []},
+        }
+        flow = FlowCreate(**raw)
+        assert flow.id == stable_id
+        assert flow.name == "Exported Flow"
+
+    def test_no_id_in_dict_gives_none(self):
+        """Existing flow dicts without id must still parse cleanly."""
+        raw = {"name": "Legacy Flow", "data": {"nodes": [], "edges": []}}
+        flow = FlowCreate(**raw)
+        assert flow.id is None
+
+
+# ---------------------------------------------------------------------------
+# Upload endpoint — upsert branch selection
+# ---------------------------------------------------------------------------
+# We test the branching logic by directly calling the three paths as the
+# endpoint itself would, verifying correct helper selection without spinning
+# up a real server or database.
+
+
+def _make_flow_create(flow_id: UUID | None = None, name: str = "Test Flow") -> FlowCreate:
+    return FlowCreate(id=flow_id, name=name, data={"nodes": [], "edges": []})
+
+
+def _make_existing_flow(flow_id: UUID, user_id: UUID) -> MagicMock:
+    """Return a mock Flow ORM object."""
+    m = MagicMock()
+    m.id = flow_id
+    m.user_id = user_id
+    return m
+
+
+class TestUploadUpsertBranches:
+    """Test the three upsert branches introduced in the upload endpoint."""
+
+    @pytest.mark.asyncio
+    async def test_no_id_calls_new_flow(self):
+        """Flow without id → _new_flow called with no flow_id kwarg."""
+        flow = _make_flow_create(flow_id=None)
+        current_user = MagicMock()
+        current_user.id = uuid4()
+
+        new_flow_mock = AsyncMock(return_value=MagicMock())
+
+        # Simulate the branch directly
+        if flow.id is not None:
+            pytest.fail("Should have taken the no-id branch")
+
+        with patch("langflow.api.v1.flows._new_flow", new_flow_mock):
+            await new_flow_mock(session=None, flow=flow, user_id=current_user.id, storage_service=None)
+
+        new_flow_mock.assert_awaited_once()
+        call_kwargs = new_flow_mock.call_args.kwargs
+        # No flow_id should be passed (will generate a UUID)
+        assert "flow_id" not in call_kwargs or call_kwargs["flow_id"] is None
+
+    @pytest.mark.asyncio
+    async def test_new_id_not_in_db_calls_new_flow_with_id(self):
+        """Flow with id that doesn't exist → _new_flow called with flow_id."""
+        stable_id = uuid4()
+        flow = _make_flow_create(flow_id=stable_id)
+        current_user = MagicMock()
+        current_user.id = uuid4()
+
+        new_flow_mock = AsyncMock(return_value=MagicMock())
+
+        # existing = None (not in DB)
+        existing = None
+
+        # Branch: id present, existing is None → CREATE with stable id
+        assert flow.id is not None
+        assert existing is None
+
+        with patch("langflow.api.v1.flows._new_flow", new_flow_mock):
+            await new_flow_mock(
+                session=None,
+                flow=flow,
+                user_id=current_user.id,
+                storage_service=None,
+                flow_id=flow.id,
+            )
+
+        call_kwargs = new_flow_mock.call_args.kwargs
+        assert call_kwargs["flow_id"] == stable_id
+
+    @pytest.mark.asyncio
+    async def test_id_owned_by_same_user_calls_update(self):
+        """Flow id exists and belongs to current user → update path."""
+        stable_id = uuid4()
+        user_id = uuid4()
+        flow = _make_flow_create(flow_id=stable_id)
+        existing = _make_existing_flow(stable_id, user_id)
+
+        current_user = MagicMock()
+        current_user.id = user_id
+
+        update_mock = AsyncMock(return_value=MagicMock())
+
+        # Branch: id present, existing owned by current user → UPDATE
+        assert flow.id is not None
+        assert existing is not None
+        assert existing.user_id == current_user.id
+
+        with patch("langflow.api.v1.flows._update_existing_flow", update_mock):
+            await update_mock(
+                session=None,
+                existing_flow=existing,
+                flow=flow,
+                current_user=current_user,
+                storage_service=None,
+            )
+
+        update_mock.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_id_owned_by_other_user_clears_id(self):
+        """Flow id exists but belongs to another user → id cleared, new UUID minted."""
+        stable_id = uuid4()
+        owner_id = uuid4()
+        requester_id = uuid4()
+
+        flow = _make_flow_create(flow_id=stable_id)
+        existing = _make_existing_flow(stable_id, owner_id)
+
+        current_user = MagicMock()
+        current_user.id = requester_id
+
+        # Branch: id present, existing owned by different user → clear id, new_flow
+        assert flow.id is not None
+        assert existing is not None
+        assert existing.user_id != current_user.id
+
+        # Simulate the branch behaviour
+        flow.id = None  # as the endpoint does
+
+        new_flow_mock = AsyncMock(return_value=MagicMock())
+        with patch("langflow.api.v1.flows._new_flow", new_flow_mock):
+            await new_flow_mock(session=None, flow=flow, user_id=current_user.id, storage_service=None)
+
+        # After clearing, flow.id must be None so a fresh UUID is generated
+        assert flow.id is None
+        call_kwargs = new_flow_mock.call_args.kwargs
+        assert "flow_id" not in call_kwargs or call_kwargs.get("flow_id") is None
+
+
+# ---------------------------------------------------------------------------
+# PUT /flows/{id} endpoint schema visibility
+# ---------------------------------------------------------------------------
+
+
+class TestUpsertEndpointVisibility:
+    def test_upsert_route_in_openapi_schema(self):
+        """PUT /flows/{flow_id} must be visible in the OpenAPI schema."""
+        from langflow.api.v1.flows import router
+
+        put_routes = [r for r in router.routes if "PUT" in getattr(r, "methods", set())]
+        upsert_routes = [r for r in put_routes if "{flow_id}" in getattr(r, "path", "")]
+
+        assert upsert_routes, "No PUT /{flow_id} route found"
+        upsert_route = upsert_routes[0]
+        # include_in_schema defaults to True; check it is not False
+        assert getattr(upsert_route, "include_in_schema", True) is True

@@ -136,7 +136,7 @@ async def files_flow(
 ):
     loaded_json = json.loads(json_flow)
     flow_data = FlowCreate(name="test_flow", data=loaded_json.get("data"), user_id=files_active_user.id)
-    flow = Flow.model_validate(flow_data)
+    flow = Flow.model_validate(flow_data.model_dump(exclude={"id"}))
     async with session_scope() as session:
         session.add(flow)
         await session.flush()

@@ -0,0 +1,81 @@
+"""Focused unit tests for flow helper branches that are hard to force via HTTP."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import anyio
+import pytest
+from fastapi import HTTPException
+from langflow.api.v1.flows_helpers import _new_flow, _save_flow_to_fs
+from langflow.services.database.models.flow.model import Flow, FlowCreate
+from langflow.services.database.models.user.model import User
+from langflow.services.storage.service import StorageService
+
+
+@pytest.fixture
+def storage_service(tmp_path):
+    """Create a mock storage service with a temporary data directory."""
+    service = MagicMock(spec=StorageService)
+    service.data_dir = anyio.Path(tmp_path)
+    return service
+
+
+@pytest.fixture
+async def current_user(async_session):
+    """Create a user that can own flows in helper-level tests."""
+    password = f"password-{uuid4()}"
+    user = User(
+        username=f"flow-helper-{uuid4()}",
+        password=password,
+        is_active=True,
+    )
+    async_session.add(user)
+    await async_session.commit()
+    await async_session.refresh(user)
+    return user
+
+
+@pytest.mark.asyncio
+async def test_new_flow_with_validate_folder_rejects_unknown_folder(async_session, current_user, storage_service):
+    """Test that validate_folder rejects a folder that does not belong to the user."""
+    flow = FlowCreate(
+        name="flow-with-bad-folder",
+        data={},
+        folder_id=uuid4(),
+    )
+
+    with (
+        patch("langflow.api.v1.flows_helpers.get_default_folder_id", new=AsyncMock()) as mock_default_folder_id,
+        pytest.raises(HTTPException) as exc_info,
+    ):
+        await _new_flow(
+            session=async_session,
+            flow=flow,
+            user_id=current_user.id,
+            storage_service=storage_service,
+            validate_folder=True,
+        )
+
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.detail == "Folder not found"
+    mock_default_folder_id.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_save_flow_to_fs_returns_500_on_os_error(current_user, storage_service):
+    """Test that filesystem write errors surface as an HTTP 500."""
+    flow = Flow(
+        name="flow-write-error",
+        data={},
+        user_id=current_user.id,
+        fs_path="nested/flow.json",
+    )
+
+    with (
+        patch("langflow.api.v1.flows_helpers.async_open", side_effect=OSError("disk full")),
+        pytest.raises(HTTPException) as exc_info,
+    ):
+        await _save_flow_to_fs(flow, current_user.id, storage_service)
+
+    assert exc_info.value.status_code == 500
+    assert "disk full" in exc_info.value.detail

@@ -151,6 +151,75 @@ async def test_update_flow(client: AsyncClient, logged_in_headers):
     assert result["name"] == updated_name, "The name must be updated"
 
 
+async def test_patch_flow_keeps_existing_endpoint_when_not_provided(client: AsyncClient, logged_in_headers):
+    """Test that PATCH preserves endpoint_name when the field is omitted."""
+    initial_flow = {
+        "name": "patch_endpoint_flow",
+        "endpoint_name": "keep_patch_endpoint",
+        "data": {},
+    }
+    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
+    assert create_response.status_code == status.HTTP_201_CREATED
+    flow_id = create_response.json()["id"]
+
+    response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={"name": "patch_endpoint_flow_updated"},
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["endpoint_name"] == "keep_patch_endpoint"
+
+
+async def test_patch_flow_allows_clearing_endpoint_with_null(client: AsyncClient, logged_in_headers):
+    """Test that PATCH clears endpoint_name when it is explicitly set to null."""
+    initial_flow = {
+        "name": "patch_clear_endpoint_flow",
+        "endpoint_name": "clear_patch_endpoint",
+        "data": {},
+    }
+    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
+    assert create_response.status_code == status.HTTP_201_CREATED
+    flow_id = create_response.json()["id"]
+
+    response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={"endpoint_name": None},
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["endpoint_name"] is None
+
+
+async def test_patch_flow_updates_access_and_action_fields(client: AsyncClient, logged_in_headers):
+    """PATCH should persist public-sharing and MCP action metadata fields."""
+    create_response = await client.post(
+        "api/v1/flows/",
+        json={"name": "patch_access_type_flow", "data": {}},
+        headers=logged_in_headers,
+    )
+    assert create_response.status_code == status.HTTP_201_CREATED
+    flow_id = create_response.json()["id"]
+
+    response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={
+            "access_type": "PUBLIC",
+            "action_name": "shared_action",
+            "action_description": "Shared flow action",
+        },
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    result = response.json()
+    assert result["access_type"] == "PUBLIC"
+    assert result["action_name"] == "shared_action"
+    assert result["action_description"] == "Shared flow action"
+
+
 async def test_create_flows(client: AsyncClient, logged_in_headers):
     amount_flows = 10
     basic_case = {
@@ -718,6 +787,48 @@ async def test_upsert_flow_keeps_existing_folder_on_update_when_not_provided(cli
     assert result["folder_id"] == original_folder_id  # Folder unchanged
 
 
+async def test_upsert_flow_keeps_existing_endpoint_when_not_provided(client: AsyncClient, logged_in_headers):
+    """Test that PUT preserves endpoint_name when it is omitted during update."""
+    initial_flow = {
+        "name": "upsert_endpoint_flow",
+        "endpoint_name": "keep_upsert_endpoint",
+        "data": {},
+    }
+    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
+    assert create_response.status_code == status.HTTP_201_CREATED
+    flow_id = create_response.json()["id"]
+
+    response = await client.put(
+        f"api/v1/flows/{flow_id}",
+        json={"name": "upsert_endpoint_flow_updated", "data": {}},
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["endpoint_name"] == "keep_upsert_endpoint"
+
+
+async def test_upsert_flow_allows_clearing_endpoint_with_null(client: AsyncClient, logged_in_headers):
+    """Test that PUT clears endpoint_name when it is explicitly set to null."""
+    initial_flow = {
+        "name": "upsert_clear_endpoint_flow",
+        "endpoint_name": "clear_upsert_endpoint",
+        "data": {},
+    }
+    create_response = await client.post("api/v1/flows/", json=initial_flow, headers=logged_in_headers)
+    assert create_response.status_code == status.HTTP_201_CREATED
+    flow_id = create_response.json()["id"]
+
+    response = await client.put(
+        f"api/v1/flows/{flow_id}",
+        json={"name": "upsert_clear_endpoint_flow", "endpoint_name": None, "data": {}},
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["endpoint_name"] is None
+
+
 async def test_upsert_flow_ignores_user_id_from_body(client: AsyncClient, logged_in_headers, active_user):
     """Test that PUT ignores user_id from body and uses current user."""
     specified_id = str(uuid.uuid4())

@@ -37,6 +37,32 @@ async def test_create_project(client: AsyncClient, logged_in_headers, basic_case
     assert "parent_id" in result, "The dictionary must contain a key called 'parent_id'"
 
 
+async def test_create_project_duplicate_name_escapes_like_wildcards(client: AsyncClient, logged_in_headers):
+    unrelated = {
+        "name": "proj_a (7)",
+        "description": "",
+        "flows_list": [],
+        "components_list": [],
+    }
+    wildcard = {
+        "name": "proj_%",
+        "description": "",
+        "flows_list": [],
+        "components_list": [],
+    }
+
+    response = await client.post("api/v1/projects/", json=unrelated, headers=logged_in_headers)
+    assert response.status_code == status.HTTP_201_CREATED
+
+    response = await client.post("api/v1/projects/", json=wildcard, headers=logged_in_headers)
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["name"] == "proj_%"
+
+    response = await client.post("api/v1/projects/", json=wildcard, headers=logged_in_headers)
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["name"] == "proj_% (1)"
+
+
 async def test_read_projects(client: AsyncClient, logged_in_headers):
     response = await client.get("api/v1/projects/", headers=logged_in_headers)
     result = response.json()
@@ -464,11 +490,11 @@ async def test_create_project_with_mcp_auto_add_enabled_success(
         """Test successful project creation with MCP server auto-add."""
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.create_api_key") as mock_create_api_key,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key") as mock_create_api_key,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks
             mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable"
@@ -514,12 +540,12 @@ async def test_create_project_with_mcp_auto_add_enabled_success_legacy_sse(
         """Legacy SSE test for project creation with MCP server auto-add."""
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.create_api_key") as mock_create_api_key,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key") as mock_create_api_key,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks
             mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse"
@@ -566,9 +592,9 @@ async def test_create_project_with_mcp_server_conflict(
         """Test project creation failure due to MCP server name conflict."""
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks
             mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable"
@@ -611,10 +637,10 @@ async def test_create_project_with_mcp_server_conflict_legacy_sse(
         """Legacy SSE test verifying project creation failure due to MCP server name conflict."""
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks
             mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse"
@@ -659,9 +685,9 @@ async def test_create_project_oauth_not_implemented(
         oauth_case["auth_settings"] = {"auth_type": "oauth"}
 
         with (
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks to trigger OAuth path
             mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable"
@@ -691,10 +717,10 @@ async def test_create_project_oauth_not_implemented_legacy_sse(
         oauth_case["auth_settings"] = {"auth_type": "oauth"}
 
         with (
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks to trigger OAuth path
             mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse"
@@ -724,11 +750,11 @@ async def test_update_project_name_with_mcp_server_update(
         # First create a project
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url"),
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"),
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -750,9 +776,9 @@ async def test_update_project_name_with_mcp_server_update(
 
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -797,12 +823,12 @@ async def test_update_project_name_with_mcp_server_update_legacy_sse(
         # First create a project
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -827,9 +853,9 @@ async def test_update_project_name_with_mcp_server_update_legacy_sse(
 
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -874,11 +900,11 @@ async def test_update_project_name_with_mcp_conflict(
         # Create project first
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url"),
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"),
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -900,8 +926,8 @@ async def test_update_project_name_with_mcp_conflict(
 
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -940,12 +966,12 @@ async def test_update_project_name_with_mcp_conflict_legacy_sse(
         # Create project first
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -970,8 +996,8 @@ async def test_update_project_name_with_mcp_conflict_legacy_sse(
 
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1010,11 +1036,11 @@ async def test_delete_project_with_mcp_server_cleanup(
         # Create project first
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url"),
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"),
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1034,9 +1060,9 @@ async def test_delete_project_with_mcp_server_cleanup(
         # Delete the project
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1074,12 +1100,12 @@ async def test_delete_project_with_mcp_server_cleanup_legacy_sse(
         # Create project first
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1102,9 +1128,9 @@ async def test_delete_project_with_mcp_server_cleanup_legacy_sse(
         # Delete the project
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1142,11 +1168,11 @@ async def test_delete_project_mcp_server_different_project(
         # Create project first
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url"),
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"),
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1166,9 +1192,9 @@ async def test_delete_project_mcp_server_different_project(
         # Delete the project
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1203,12 +1229,12 @@ async def test_delete_project_mcp_server_different_project_legacy_sse(
         # Create project first
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate_create,
-            patch("langflow.api.v1.projects.update_server"),
-            patch("langflow.api.v1.projects.create_api_key"),
-            patch("langflow.api.v1.projects.get_storage_service"),
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server"),
+            patch("langflow.api.v1.projects_mcp_helpers.create_api_key"),
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"),
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1231,9 +1257,9 @@ async def test_delete_project_mcp_server_different_project_legacy_sse(
         # Delete the project
         with (
             patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.update_server") as mock_update_server,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Mock settings to enable MCP auto-add
             mock_settings = MagicMock()
@@ -1285,9 +1311,9 @@ async def test_project_mcp_exception_handling(
     ):
         """Test that MCP exceptions during project creation don't prevent project creation."""
         with (
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks
             mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable"
@@ -1313,10 +1339,10 @@ async def test_project_mcp_exception_handling_legacy_sse(
     ):
         """Legacy SSE test ensuring MCP exceptions don't block project creation."""
         with (
-            patch("langflow.api.v1.projects.get_project_sse_url") as mock_sse_url,
-            patch("langflow.api.v1.projects.get_project_streamable_http_url") as mock_streamable_url,
-            patch("langflow.api.v1.projects.validate_mcp_server_for_project") as mock_validate,
-            patch("langflow.api.v1.projects.get_storage_service") as mock_storage,
+            patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url,
+            patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url,
+            patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate,
+            patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage,
         ):
             # Setup mocks
             mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse"
@@ -1620,7 +1646,7 @@ async def test_download_file_starter_project(client: AsyncClient, logged_in_head
                 folder_id=starter_project_id,
                 user_id=active_user.id,
             )
-            flow = Flow.model_validate(flow_create, from_attributes=True)
+            flow = Flow.model_validate(flow_create.model_dump(exclude={"id"}))
             session.add(flow)
             flows_created.append(flow)
 
@@ -1632,7 +1658,7 @@ async def test_download_file_starter_project(client: AsyncClient, logged_in_head
             folder_id=starter_project_id,
             user_id=active_user.id,
         )
-        flow_note = Flow.model_validate(flow_create_note, from_attributes=True)
+        flow_note = Flow.model_validate(flow_create_note.model_dump(exclude={"id"}))
         session.add(flow_note)
         flows_created.append(flow_note)
 
@@ -1692,7 +1718,6 @@ async def test_download_file_starter_project(client: AsyncClient, logged_in_head
         assert "data" in note_flow_json
         assert "nodes" in note_flow_json["data"]
         assert len(note_flow_json["data"]["nodes"]) == 2
-
         # Find the API node and verify API key was removed
         api_node = None
         note_node = None
@@ -1719,3 +1744,54 @@ async def test_download_file_starter_project(client: AsyncClient, logged_in_head
     # Clean up: delete the project (which will cascade delete flows)
     delete_response = await client.delete(f"api/v1/projects/{starter_project_id}", headers=logged_in_headers)
     assert delete_response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_download_project_missing_returns_404(client: AsyncClient, logged_in_headers):
+    response = await client.get(f"api/v1/projects/download/{uuid4()}", headers=logged_in_headers)
+
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json()["detail"] == "Project not found"
+
+
+async def test_download_project_with_no_flows_returns_404(client: AsyncClient, logged_in_headers, basic_case):
+    response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers)
+    assert response.status_code == status.HTTP_201_CREATED
+    project_id = response.json()["id"]
+
+    download_response = await client.get(f"api/v1/projects/download/{project_id}", headers=logged_in_headers)
+
+    assert download_response.status_code == status.HTTP_404_NOT_FOUND
+    assert download_response.json()["detail"] == "No flows found in project"
+
+
+async def test_download_project_sanitizes_windows_path_characters(
+    client: AsyncClient, logged_in_headers, basic_case, active_user
+):
+    create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers)
+    assert create_response.status_code == status.HTTP_201_CREATED
+    project_id = create_response.json()["id"]
+
+    async with session_scope() as session:
+        flow_create = FlowCreate(
+            name=r"..\evil\flow",
+            description="Flow with unsafe filename characters",
+            data={"nodes": [], "edges": []},
+            folder_id=project_id,
+            user_id=active_user.id,
+        )
+        flow = Flow.model_validate(flow_create.model_dump(exclude={"id"}))
+        session.add(flow)
+        await session.flush()
+        await session.refresh(flow)
+        await session.commit()
+
+    response = await client.get(f"api/v1/projects/download/{project_id}", headers=logged_in_headers)
+    assert response.status_code == status.HTTP_200_OK
+
+    with zipfile.ZipFile(io.BytesIO(response.content), "r") as zip_file:
+        file_names = zip_file.namelist()
+        assert len(file_names) == 1
+        assert "/" not in file_names[0]
+        assert "\\" not in file_names[0]
+        assert ".." not in file_names[0]
+        assert file_names[0].endswith(".json")

@@ -545,7 +545,7 @@ async def test_download_file(
         saved_flows = []
         for flow in flow_list.flows:
             flow.user_id = active_user.id
-            db_flow = Flow.model_validate(flow, from_attributes=True)
+            db_flow = Flow.model_validate(flow.model_dump(exclude={"id"}))
             _session.add(db_flow)
             saved_flows.append(db_flow)
         await _session.commit()
@@ -565,6 +565,56 @@ async def test_download_file(
     assert "attachment; filename=" in response.headers["Content-Disposition"]
 
 
+@pytest.mark.usefixtures("session")
+async def test_download_single_flow_returns_normalized_json(client: AsyncClient, logged_in_headers):
+    """Downloading a single flow returns normalized JSON rather than a ZIP archive."""
+    code_value = "print('hello')\nprint('world')"
+    flow_payload = FlowCreate(
+        name=str(uuid4()),
+        description="single flow export",
+        data={
+            "nodes": [
+                {
+                    "id": "node-1",
+                    "data": {
+                        "node": {
+                            "template": {
+                                "code": {"type": "code", "value": code_value},
+                                "api_key": {"name": "api_key", "password": True, "value": "super-secret"},
+                            }
+                        }
+                    },
+                }
+            ],
+            "edges": [],
+        },
+    )
+
+    create_response = await client.post("api/v1/flows/", json=flow_payload.model_dump(), headers=logged_in_headers)
+    assert create_response.status_code == 201
+    flow_id = create_response.json()["id"]
+
+    download_response = await client.post(
+        "api/v1/flows/download/",
+        data=json.dumps([flow_id]),
+        headers={**logged_in_headers, "Content-Type": "application/json"},
+    )
+    assert download_response.status_code == 200
+    assert download_response.headers["Content-Type"].startswith("application/json")
+
+    downloaded = download_response.json()
+    assert downloaded["name"] == flow_payload.name
+    assert "updated_at" not in downloaded
+    assert "user_id" not in downloaded
+    assert "folder_id" not in downloaded
+    assert "access_type" not in downloaded
+    assert downloaded["data"]["nodes"][0]["data"]["node"]["template"]["code"]["value"] == [
+        "print('hello')",
+        "print('world')",
+    ]
+    assert downloaded["data"]["nodes"][0]["data"]["node"]["template"]["api_key"]["value"] is None
+
+
 @pytest.mark.usefixtures("session")
 async def test_upload_zip_file_to_flows(client: AsyncClient, json_flow: str, logged_in_headers):
     """Test uploading a ZIP file containing flow JSONs to the flows upload endpoint."""
@@ -671,6 +721,7 @@ async def test_download_then_upload_roundtrip(client: AsyncClient, json_flow: st
         saved_flows = []
         for f in flow_list.flows:
             f.user_id = active_user.id
+            f.id = uuid4()
             db_flow = Flow.model_validate(f, from_attributes=True)
             _session.add(db_flow)
             saved_flows.append(db_flow)
@@ -843,6 +894,90 @@ async def test_upload_zip_to_projects_filename_none(client: AsyncClient, json_fl
     assert project_response.json()["name"].startswith("Imported Project")
 
 
+@pytest.mark.usefixtures("session")
+async def test_upload_json_file_to_projects_rejoins_code_lines(client: AsyncClient, logged_in_headers):
+    """Project JSON upload accepts exported code-as-lines format and stores code as a string."""
+    project_name = f"JSON Project {uuid4()}"
+    code_lines = ["print('alpha')", "print('beta')"]
+    payload = {
+        "folder_name": project_name,
+        "folder_description": "json upload",
+        "flows": [
+            {
+                "name": f"Flow {uuid4()}",
+                "description": "imported from project json",
+                "data": {
+                    "nodes": [
+                        {
+                            "id": "node-1",
+                            "data": {
+                                "node": {
+                                    "template": {
+                                        "code": {"type": "code", "value": code_lines},
+                                    }
+                                }
+                            },
+                        }
+                    ],
+                    "edges": [],
+                },
+            }
+        ],
+    }
+
+    response = await client.post(
+        "api/v1/projects/upload/",
+        files={"file": ("project.json", json.dumps(payload).encode("utf-8"), "application/json")},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == 201, response.text
+    response_data = response.json()
+    assert len(response_data) == 1
+    assert response_data[0]["folder_id"] is not None
+    assert (
+        response_data[0]["data"]["nodes"][0]["data"]["node"]["template"]["code"]["value"]
+        == "print('alpha')\nprint('beta')"
+    )
+
+    project_response = await client.get(f"api/v1/projects/{response_data[0]['folder_id']}", headers=logged_in_headers)
+    assert project_response.status_code == 200
+    assert project_response.json()["name"].startswith(project_name)
+
+
+@pytest.mark.usefixtures("session")
+async def test_download_project_zip_sanitizes_flow_names(client: AsyncClient, json_flow: str, logged_in_headers):
+    """Project ZIP downloads must sanitize flow names to prevent Zip Slip paths."""
+    project_response = await client.post(
+        "api/v1/projects/",
+        json={"name": f"Download Project {uuid4()}", "description": "", "flows_list": [], "components_list": []},
+        headers=logged_in_headers,
+    )
+    assert project_response.status_code == 201
+    project_id = project_response.json()["id"]
+
+    flow = orjson.loads(json_flow)
+    create_response = await client.post(
+        "api/v1/flows/",
+        json={
+            "name": "../escaped-flow",
+            "description": "path traversal test",
+            "folder_id": project_id,
+            "data": flow["data"],
+            "is_component": False,
+        },
+        headers=logged_in_headers,
+    )
+    assert create_response.status_code == 201
+
+    download_response = await client.get(f"api/v1/projects/download/{project_id}", headers=logged_in_headers)
+    assert download_response.status_code == 200
+
+    with zipfile.ZipFile(io.BytesIO(download_response.content), "r") as zip_file:
+        file_names = zip_file.namelist()
+        assert "__escaped-flow.json" in file_names
+        assert all("/" not in name and ".." not in name for name in file_names)
+
+
 @pytest.mark.usefixtures("session")
 async def test_upload_bad_zip_file_returns_400(client: AsyncClient, logged_in_headers):
     """Uploading a corrupt/invalid ZIP file returns 400 with a descriptive error."""

@@ -24,9 +24,11 @@ ENV UV_LINK_MODE=copy
 # Workspace root metadata + lockfile
 COPY pyproject.toml uv.lock ./
 
-# Member's pyproject so uv knows about 'lfx' (no source yet, better cache)
+# Member pyproject files so uv knows about workspace packages (no source yet, better cache)
 COPY src/lfx/pyproject.toml /app/src/lfx/pyproject.toml
 COPY src/lfx/README.md /app/src/lfx/README.md
+COPY src/sdk/pyproject.toml /app/src/sdk/pyproject.toml
+COPY src/sdk/README.md /app/src/sdk/README.md
 
 # Create the venv and install *only* what lfx needs (no dev)
 # We expect some packages to be built from source, so we mount the cache
@@ -35,6 +37,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
 
 # --- Now copy the source (doesn't bust the deps layer) ---
 COPY src/lfx/src /app/src/lfx/src
+COPY src/sdk/src /app/src/sdk/src
 
 # Install the LFX package into the virtual environment (non-editable)
 RUN --mount=type=cache,target=/root/.cache/uv \

@@ -27,9 +27,11 @@ RUN apt-get update \
 # Workspace root metadata + lockfile
 COPY pyproject.toml uv.lock ./
 
-# Member's pyproject so uv knows about workspace packages (no source yet, better cache)
+# Member pyproject files so uv knows about workspace packages (no source yet, better cache)
 COPY src/lfx/pyproject.toml /app/src/lfx/pyproject.toml
 COPY src/lfx/README.md /app/src/lfx/README.md
+COPY src/sdk/pyproject.toml /app/src/sdk/pyproject.toml
+COPY src/sdk/README.md /app/src/sdk/README.md
 COPY src/backend/base/pyproject.toml /app/src/backend/base/pyproject.toml
 COPY src/backend/base/README.md /app/src/backend/base/README.md
 
@@ -40,6 +42,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
 # --- Now copy the source and tests (doesn't bust the deps layer) ---
 COPY src/lfx/src /app/src/lfx/src
 COPY src/lfx/tests /app/src/lfx/tests
+COPY src/sdk/src /app/src/sdk/src
 
 # Install the LFX package into the virtual environment (editable for dev)
 RUN --mount=type=cache,target=/root/.cache/uv \

@@ -33,6 +33,7 @@ dependencies = [
     "passlib>=1.7.4,<2.0.0",
     "pydantic-settings>=2.10.1,<3.0.0",
     "tomli>=2.2.1,<3.0.0",
+    "pyyaml>=6.0.0,<7.0.0",
     "orjson>=3.11.6,<4.0.0",
     "asyncer>=0.0.8,<1.0.0",
     "structlog>=25.4.0,<26.0.0",
@@ -44,6 +45,7 @@ dependencies = [
     "pypdf>=6.9.0,<7.0.0",
     "cryptography>=46.0.6",
     "ag-ui-protocol>=0.1.10",
+    "langflow-sdk>=0.1.0",
     "markitdown>=0.1.4,<2.0.0",
     "setuptools>=80.0.0,<81.0.0",
     "wheel>=0.46.2,<1.0.0",
@@ -54,6 +56,9 @@ dependencies = [
 lfx = "lfx.__main__:main"
 lfx-mcp = "lfx.mcp.__main__:main"
 
+[project.entry-points.pytest11]
+lfx = "lfx.testing"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"
@@ -72,7 +77,9 @@ markers = [
     "unit: Unit tests",
     "integration: Integration tests",
     "slow: Slow-running tests",
-    "asyncio: Async tests"
+    "asyncio: Async tests",
+    "lfx_env_file(path): path to a .env file loaded before this test's flow execution",
+    "lfx_timeout(seconds): timeout in seconds for this test's flow execution"
 ]
 
 [dependency-groups]

@@ -0,0 +1,207 @@
+"""Authoring commands: create, requirements, validate."""
+
+import typer
+
+
+def register(app: typer.Typer) -> None:
+    """Register authoring-stage commands on *app*."""
+
+    @app.command(name="create", help="Create a new flow JSON from a built-in template", rich_help_panel="Authoring")
+    def create_command_wrapper(
+        name: str = typer.Argument(help="Display name for the new flow (also used as the filename)."),
+        template: str = typer.Option(
+            "hello-world",
+            "--template",
+            "-t",
+            help="Template to use. Run with --list to see all available templates.",
+        ),
+        output_dir: str = typer.Option(
+            "flows",
+            "--output-dir",
+            "-o",
+            help="Directory to write the new flow JSON into (created if absent; default: flows/).",
+        ),
+        *,
+        list_templates: bool = typer.Option(
+            False,
+            "--list",
+            "-l",
+            help="Print available templates and exit.",
+            is_eager=True,
+        ),
+        overwrite: bool = typer.Option(
+            False,
+            "--overwrite",
+            help="Overwrite the destination file if it already exists.",
+        ),
+    ) -> None:
+        """Scaffold a new Langflow flow JSON from a built-in template (lazy-loaded)."""
+        from pathlib import Path
+
+        from lfx.cli.create import create_command, print_templates
+
+        if list_templates:
+            print_templates()
+            raise typer.Exit(0)
+
+        create_command(
+            name=name,
+            template=template,
+            output_dir=Path(output_dir),
+            overwrite=overwrite,
+        )
+
+    @app.command(
+        name="requirements",
+        help="Generate requirements.txt for a flow",
+        no_args_is_help=True,
+        rich_help_panel="Authoring",
+    )
+    def requirements_command_wrapper(
+        flow_path: str = typer.Argument(help="Path to the Langflow flow JSON file"),
+        output: str | None = typer.Option(
+            None,
+            "--output",
+            "-o",
+            help="Output file path (default: stdout)",
+        ),
+        lfx_package: str = typer.Option(
+            "lfx",
+            "--lfx-package",
+            help="Name of the LFX package (default: lfx)",
+        ),
+        *,
+        no_lfx: bool = typer.Option(
+            False,
+            "--no-lfx",
+            help="Exclude the LFX package from output",
+        ),
+        no_pin: bool = typer.Option(
+            False,
+            "--no-pin",
+            help="Do not pin package versions (default: pin to currently installed versions)",
+        ),
+    ) -> None:
+        """Generate requirements.txt from a Langflow flow JSON (lazy-loaded)."""
+        import json
+        from pathlib import Path
+
+        from lfx.utils.flow_requirements import generate_requirements_txt
+
+        path = Path(flow_path)
+        if not path.is_file():
+            typer.echo(f"Error: File not found: {path}", err=True)
+            raise typer.Exit(1)
+
+        try:
+            flow = json.loads(path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError) as e:
+            typer.echo(f"Error: Could not read flow JSON: {e}", err=True)
+            raise typer.Exit(1) from e
+
+        content = generate_requirements_txt(
+            flow,
+            lfx_package=lfx_package,
+            include_lfx=not no_lfx,
+            pin_versions=not no_pin,
+        )
+
+        if output:
+            try:
+                Path(output).write_text(content, encoding="utf-8")
+            except OSError as e:
+                typer.echo(f"Error: Could not write to {output}: {e}", err=True)
+                raise typer.Exit(1) from e
+            typer.echo(f"Requirements written to {output}")
+        else:
+            typer.echo(content, nl=False)
+
+    @app.command(name="validate", help="Validate one or more flow JSON files", rich_help_panel="Authoring")
+    def validate_command_wrapper(
+        flow_paths: list[str] = typer.Argument(
+            default=None,
+            help="Path(s) to Langflow flow JSON file(s) or directories to validate. Defaults to flows/.",
+        ),
+        dir_path: str | None = typer.Option(
+            None,
+            "--dir",
+            "-d",
+            help="Directory of flow JSON files to validate (validates all *.json files). Defaults to flows/.",
+        ),
+        level: int = typer.Option(
+            4,
+            "--level",
+            "-l",
+            min=1,
+            max=4,
+            help=(
+                "Validation depth: "
+                "1=structural JSON, "
+                "2=+component existence, "
+                "3=+edge type compatibility, "
+                "4=+required inputs connected"
+            ),
+        ),
+        skip_components: bool = typer.Option(
+            False,
+            "--skip-components",
+            help="Skip component existence checks (level 2)",
+        ),
+        skip_edge_types: bool = typer.Option(
+            False,
+            "--skip-edge-types",
+            help="Skip edge type compatibility checks (level 3)",
+        ),
+        skip_required_inputs: bool = typer.Option(
+            False,
+            "--skip-required-inputs",
+            help="Skip required-inputs checks (level 4)",
+        ),
+        skip_version_check: bool = typer.Option(
+            False,
+            "--skip-version-check",
+            help="Skip version-mismatch / outdated-component warnings",
+        ),
+        skip_credentials: bool = typer.Option(
+            False,
+            "--skip-credentials",
+            help="Skip missing-credentials warnings for password/secret fields",
+        ),
+        strict: bool = typer.Option(
+            False,
+            "--strict",
+            help="Treat warnings as errors (exit 1 if any warnings are found)",
+        ),
+        verbose: bool = typer.Option(
+            False,
+            "--verbose",
+            "-v",
+            help="Print all issues including warnings for passing flows",
+        ),
+        output_format: str = typer.Option(
+            "text",
+            "--format",
+            "-f",
+            help="Output format: text (default) or json",
+        ),
+    ) -> None:
+        """Validate Langflow flow JSON files without executing them (lazy-loaded)."""
+        from lfx.cli.validate import validate_command
+
+        # Merge --dir into positional paths for a consistent interface with push
+        effective_paths = list(flow_paths or [])
+        if dir_path is not None:
+            effective_paths.append(dir_path)
+
+        validate_command(
+            flow_paths=effective_paths,
+            level=level,
+            skip_components=skip_components,
+            skip_edge_types=skip_edge_types,
+            skip_required_inputs=skip_required_inputs,
+            skip_version_check=skip_version_check,
+            skip_credentials=skip_credentials,
+            strict=strict,
+            verbose=verbose,
+            output_format=output_format,
+        )

@@ -17,7 +17,7 @@
 from io import StringIO
 from pathlib import Path
 from shutil import which
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 from urllib.parse import urlparse
 
 import httpx
@@ -106,8 +106,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("LANGFLOW_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("LANGFLOW_API_KEY") or os.getenv("LFX_API_KEY")
     if not api_key:
         msg = "LANGFLOW_API_KEY environment variable is not set"
         raise ValueError(msg)
@@ -613,6 +619,31 @@ def download_and_extract_repo(url: str, verbose_print, *, timeout: float = 60.0)
         return root_path
 
 
+def load_sdk(command_name: str) -> Any:
+    """Lazily import ``langflow_sdk`` to keep CLI startup fast.
+
+    Raises :class:`typer.BadParameter` with install guidance when the package
+    is not available.
+
+    Args:
+        command_name: Name of the CLI command requesting the SDK (used in
+            the error message).
+    """
+    try:
+        import langflow_sdk  # type: ignore[import-untyped]
+    except ImportError as exc:
+        msg = f"langflow-sdk is required for lfx {command_name}. Install it with: pip install langflow-sdk"
+        raise typer.BadParameter(msg) from exc
+    else:
+        return langflow_sdk
+
+
+def safe_filename(name: str) -> str:
+    """Convert a flow name to a safe filesystem basename (no extension)."""
+    safe = "".join(c if c.isalnum() or c in "-_ " else "_" for c in name)
+    return safe.strip().replace(" ", "_")
+
+
 def extract_script_docstring(script_path: Path) -> str | None:
     """Extract the module-level docstring from a Python script.
 

@@ -0,0 +1,157 @@
+"""lfx create -- scaffold a new flow JSON from a built-in template.
+
+Writes a ready-to-edit flow JSON file into the target directory so teams
+can start from a known-good structure rather than an empty file.
+
+Examples::
+
+    lfx create my-chatbot
+    lfx create my-rag --template hello-world
+    lfx create my-flow --output-dir ./flows --overwrite
+    lfx create --list
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from pathlib import Path
+from typing import Any
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+console = Console()
+
+_FLOWS_TEMPLATE_DIR = Path(__file__).parent.parent / "templates" / "flows"
+
+# Descriptions shown in ``lfx create --list``
+_TEMPLATE_DESCRIPTIONS: dict[str, str] = {
+    "hello-world": "ChatInput → ChatOutput — minimal echo flow, no LLM required",
+}
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def list_templates() -> list[str]:
+    """Return the names of all available flow templates (sorted)."""
+    if not _FLOWS_TEMPLATE_DIR.exists():
+        return []
+    return sorted(p.stem for p in _FLOWS_TEMPLATE_DIR.glob("*.json"))
+
+
+def _load_template(name: str) -> dict[str, Any]:
+    """Load and parse a template JSON by name.  Raises FileNotFoundError if missing."""
+    path = _FLOWS_TEMPLATE_DIR / f"{name}.json"
+    if not path.exists():
+        available = list_templates()
+        hint = f"  Available templates: {', '.join(available)}" if available else "  No templates found."
+        msg = f"Template '{name}' not found.\n{hint}"
+        raise FileNotFoundError(msg)
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _slugify(name: str) -> str:
+    """Convert a flow name to a safe filename stem (lowercase, hyphens)."""
+    return name.lower().replace(" ", "-").replace("_", "-")
+
+
+# ---------------------------------------------------------------------------
+# Core command (importable for testing and for init seeding)
+# ---------------------------------------------------------------------------
+
+
+def _is_lfx_project() -> bool:
+    """Return True if cwd (or a parent up to .git) contains .lfx/environments.yaml."""
+    cwd = Path.cwd()
+    for directory in (cwd, *cwd.parents):
+        if (directory / ".lfx" / "environments.yaml").is_file():
+            return True
+        if (directory / ".lfx" / "environments.yml").is_file():
+            return True
+        if (directory / ".git").is_dir() or directory.parent == directory:
+            break
+    return False
+
+
+def create_command(
+    name: str,
+    *,
+    template: str = "hello-world",
+    output_dir: Path = Path("flows"),
+    overwrite: bool = False,
+) -> Path:
+    """Create a new flow JSON from *template* and write it to *output_dir/<slug>.json*.
+
+    Returns the path of the written file.
+    Raises ``typer.Exit`` on user-facing errors so the CLI reports them cleanly.
+    """
+    if not _is_lfx_project():
+        console.print(
+            "[yellow]Warning:[/yellow] No .lfx/environments.yaml found in this project. "
+            "Run [bold]lfx init[/bold] first to scaffold a project."
+        )
+
+    available = list_templates()
+    if not available:
+        console.print("[red]Error:[/red] No flow templates found in the lfx package.")
+        raise typer.Exit(1)
+
+    if template not in available:
+        console.print(
+            f"[red]Error:[/red] Unknown template [bold]{template!r}[/bold]. Available: {', '.join(available)}"
+        )
+        raise typer.Exit(1)
+
+    slug = _slugify(name)
+    output_dir = output_dir.resolve()
+    dest = output_dir / f"{slug}.json"
+
+    if dest.exists() and not overwrite:
+        console.print(f"[red]Error:[/red] {dest} already exists. Use [bold]--overwrite[/bold] to replace it.")
+        raise typer.Exit(1)
+
+    # Load template and stamp with a fresh UUID + the requested name
+    flow = _load_template(template)
+    flow["id"] = str(uuid.uuid4())
+    flow["name"] = name
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    dest.write_text(json.dumps(flow, indent=2), encoding="utf-8")
+
+    console.print(f"[bold green]✓[/bold green] Created [bold]{dest}[/bold]")
+    console.print(f"  Template : {template}")
+    console.print(f"  Flow ID  : {flow['id']}")
+    console.print()
+    console.print("Next steps:")
+    console.print(f"  [bold]lfx validate {dest}[/bold]")
+    console.print(f"  [bold]lfx serve {dest}[/bold]")
+
+    return dest
+
+
+# ---------------------------------------------------------------------------
+# Listing helper (also used by the CLI --list flag)
+# ---------------------------------------------------------------------------
+
+
+def print_templates() -> None:
+    """Print available templates as a Rich table."""
+    available = list_templates()
+    if not available:
+        console.print("[yellow]No flow templates found.[/yellow]")
+        return
+
+    table = Table(title="Available flow templates", show_header=True, header_style="bold")
+    table.add_column("Name", style="cyan", no_wrap=True)
+    table.add_column("Description")
+
+    for name in available:
+        desc = _TEMPLATE_DESCRIPTIONS.get(name, "")
+        table.add_row(name, desc)
+
+    console.print(table)

@@ -0,0 +1,188 @@
+"""lfx export -- serialize flows to git-friendly JSON.
+
+Two modes of operation
+----------------------
+Local (default)
+    Read one or more flow JSON files from disk, normalize them, and write the
+    result back to disk (or stdout).  No network connection required.
+
+Remote (--env / --flow-id / --project-id)
+    Pull flows directly from a running Langflow instance using the
+    ``langflow-sdk`` HTTP client, normalize them, and write to disk.
+
+Examples:
+--------
+Normalize a local file in-place::
+
+    lfx export my_flow.json --in-place
+
+Normalize and write to a new file::
+
+    lfx export my_flow.json -o my_flow.normalized.json
+
+Pull a single flow from staging and write to the current directory::
+
+    lfx export --flow-id <uuid> --env staging
+
+Export an entire project from staging into ./flows/::
+
+    lfx export --project-id <uuid> --env staging --output-dir ./flows/
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import typer
+from rich.console import Console
+
+from lfx.cli.common import load_sdk, safe_filename
+
+console = Console(stderr=True)
+
+
+def _write_flow(
+    flow: dict[str, Any],
+    *,
+    sdk: Any,
+    output: Path | None,
+    in_place: bool,
+    source_path: Path | None,
+    indent: int,
+) -> Path | None:
+    """Serialise *flow* and write it to the appropriate destination.
+
+    Returns the path written to, or ``None`` if writing to stdout.
+    """
+    content = sdk.flow_to_json(flow, indent=indent)
+
+    if in_place and source_path:
+        source_path.write_text(content, encoding="utf-8")
+        return source_path
+
+    if output:
+        output.write_text(content, encoding="utf-8")
+        return output
+
+    sys.stdout.write(content)
+    return None
+
+
+def export_command(
+    flow_paths: list[str],
+    *,
+    output: str | None,
+    output_dir: str | None,
+    env: str | None,
+    flow_id: str | None,
+    project_id: str | None,
+    environments_file: str | None,
+    target: str | None = None,
+    api_key: str | None = None,
+    in_place: bool,
+    strip_volatile: bool,
+    strip_secrets: bool,
+    code_as_lines: bool,
+    strip_node_volatile: bool,
+    indent: int,
+) -> None:
+    sdk = load_sdk("export")
+
+    normalize_kwargs = {
+        "strip_volatile": strip_volatile,
+        "strip_secrets": strip_secrets,
+        "sort_keys": True,
+        "code_as_lines": code_as_lines,
+        "strip_node_volatile": strip_node_volatile,
+    }
+
+    # ------------------------------------------------------------------
+    # Remote mode: pull from a live Langflow instance
+    # ------------------------------------------------------------------
+    if flow_id or project_id:
+        if not env and not target:
+            console.print("[red]Error:[/red] --env or --target is required for remote export.")
+            raise typer.Exit(1)
+
+        from lfx.config import ConfigError, resolve_environment
+
+        try:
+            env_cfg = resolve_environment(
+                env,
+                target=target,
+                api_key=api_key,
+                environments_file=environments_file,
+            )
+        except ConfigError as exc:
+            console.print(f"[red]Error:[/red] {exc}")
+            raise typer.Exit(1) from exc
+
+        client = sdk.Client(base_url=env_cfg.url, api_key=env_cfg.api_key)
+
+        if flow_id:
+            flow_obj = client.get_flow(UUID(flow_id))
+            normalized = sdk.normalize_flow(flow_obj.model_dump(mode="json"), **normalize_kwargs)
+            dest_dir = Path(output_dir) if output_dir else Path.cwd()
+            dest_dir.mkdir(parents=True, exist_ok=True)
+            safe_name = safe_filename(flow_obj.name)
+            out_path = dest_dir / f"{safe_name}.json"
+            out_path.write_text(sdk.flow_to_json(normalized, indent=indent), encoding="utf-8")
+            console.print(f"[green]Exported[/green] {flow_obj.name!r} → {out_path}")
+            return
+
+        # Project mode: export all flows
+        project = client.get_project(UUID(project_id))
+        dest_dir = Path(output_dir) if output_dir else Path.cwd() / safe_filename(project.name)
+        dest_dir.mkdir(parents=True, exist_ok=True)
+
+        exported = 0
+        for flow_obj in project.flows:
+            normalized = sdk.normalize_flow(flow_obj.model_dump(mode="json"), **normalize_kwargs)
+            safe_name = safe_filename(flow_obj.name)
+            out_path = dest_dir / f"{safe_name}.json"
+            out_path.write_text(sdk.flow_to_json(normalized, indent=indent), encoding="utf-8")
+            console.print(f"[green]Exported[/green] {flow_obj.name!r} → {out_path}")
+            exported += 1
+
+        console.print(f"\n[bold]{exported}[/bold] flow(s) exported to {dest_dir}")
+        return
+
+    # ------------------------------------------------------------------
+    # Local mode: normalize files already on disk
+    # ------------------------------------------------------------------
+    if not flow_paths:
+        console.print("[red]Error:[/red] Provide at least one flow JSON file, or use --flow-id / --project-id.")
+        raise typer.Exit(1)
+
+    if output and len(flow_paths) > 1:
+        console.print("[red]Error:[/red] --output can only be used with a single input file.")
+        raise typer.Exit(1)
+
+    out_path_obj = Path(output) if output else None
+
+    for raw_path in flow_paths:
+        src = Path(raw_path)
+        if not src.exists():
+            console.print(f"[red]Error:[/red] File not found: {src}")
+            raise typer.Exit(1)
+
+        try:
+            normalized = sdk.normalize_flow_file(src, **normalize_kwargs)
+        except (json.JSONDecodeError, OSError, ValueError) as exc:
+            console.print(f"[red]Error:[/red] Could not process {src}: {exc}")
+            raise typer.Exit(1) from exc
+
+        dest = _write_flow(
+            normalized,
+            sdk=sdk,
+            output=out_path_obj,
+            in_place=in_place,
+            source_path=src,
+            indent=indent,
+        )
+        if dest:
+            console.print(f"[green]Exported[/green] {src} → {dest}")

@@ -0,0 +1,257 @@
+"""lfx init -- scaffold a new Flow DevOps project.
+
+Creates the standard directory layout, an environments config stub,
+example tests, and (optionally) GitHub Actions CI workflows -- everything
+a team needs to start treating flows as code.
+
+Examples::
+
+    lfx init my-rag-project
+    lfx init .                   # scaffold into the current directory
+    lfx init my-project --no-github-actions
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import typer
+from rich.console import Console
+from rich.tree import Tree
+
+console = Console()
+
+# ---------------------------------------------------------------------------
+# Templates embedded as strings (environments config and test stubs)
+# ---------------------------------------------------------------------------
+
+_ENVIRONMENTS_YAML = """\
+# .lfx/environments.yaml
+#
+# Configure your Langflow instances here.
+# Safe to commit — API keys are NEVER stored in this file.
+# The api_key_env value is the NAME of an environment variable that holds
+# the actual API key; set that variable in your shell or CI secrets.
+#
+# Quick start:
+#   1. Open Langflow → Settings → API Keys → Create a new key
+#   2. export LANGFLOW_LOCAL_API_KEY=<your key>
+#   3. lfx export --env local --flow-id <uuid> --output-dir flows/
+
+environments:
+  local:
+    url: http://localhost:7860
+    api_key_env: LANGFLOW_LOCAL_API_KEY
+
+  staging:
+    url: https://staging.langflow.example.com
+    api_key_env: LANGFLOW_STAGING_API_KEY
+
+  production:
+    url: https://langflow.example.com
+    api_key_env: LANGFLOW_PROD_API_KEY
+
+defaults:
+  environment: local
+"""
+
+_TEST_FLOWS_PY = '''\
+"""Integration tests for Langflow flows.
+
+Run against a local instance (started with ``lfx serve``):
+
+    pytest tests/ --langflow-url http://localhost:8000
+
+Run against a named environment (staging, production, etc.):
+
+    pytest tests/ --langflow-env staging -m integration
+
+The flow_runner fixture auto-skips when no connection is configured,
+so these tests are safe to include in any CI pipeline.
+"""
+
+import pytest
+
+
+@pytest.mark.integration
+def test_flow_responds(flow_runner):
+    """Smoke test: every flow should return a non-empty response."""
+    # TODO: replace "my-flow-endpoint" with your flow\'s endpoint name or UUID
+    result = flow_runner("my-flow-endpoint", "Hello!")
+    assert result.first_text_output() is not None, "Flow returned no output"
+
+
+@pytest.mark.integration
+def test_flow_output_quality(flow_runner):
+    """Example: assert on the content of the response."""
+    result = flow_runner("my-flow-endpoint", "What is Langflow?")
+    text = result.first_text_output()
+    assert text is not None
+    assert len(text) > 20, f"Response seems too short: {text!r}"
+'''
+
+_GITIGNORE = """\
+# Langflow credentials -- never commit API keys
+# (langflow-environments.toml may contain literal keys; .lfx/environments.yaml is safe to commit)
+langflow-environments.toml
+"""
+
+# Templates bundled inside the Python package
+_TEMPLATES_DIR = Path(__file__).parent.parent / "templates"
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _write(
+    path: Path,
+    content: str,
+    label: str,
+    created: list[tuple[str, str]],
+    *,
+    target: Path,
+    overwrite: bool,
+) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if path.exists() and not overwrite:
+        return
+    path.write_text(content, encoding="utf-8")
+    created.append((str(path.relative_to(target)), label))
+
+
+def _copy_template(
+    src: Path, dest: Path, label: str, created: list[tuple[str, str]], *, target: Path, overwrite: bool
+) -> None:
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    if dest.exists() and not overwrite:
+        return
+    dest.write_text(src.read_text(encoding="utf-8"), encoding="utf-8")
+    created.append((str(dest.relative_to(target)), label))
+
+
+def _render_tree(target: Path, created: list[tuple[str, str]]) -> None:
+    label = f"[bold]{target.name}/[/bold]" if target != Path.cwd() else "[bold].[/bold] (current directory)"
+    tree = Tree(label)
+    branch_nodes: dict[str, Any] = {}
+
+    for rel_path, annotation in sorted(created, key=lambda x: x[0]):
+        parts = Path(rel_path).parts
+        node = tree
+        for i, part in enumerate(parts[:-1]):
+            key = "/".join(parts[: i + 1])
+            if key not in branch_nodes:
+                branch_nodes[key] = node.add(f"[bold blue]{part}/[/bold blue]")
+            node = branch_nodes[key]
+        suffix = f"  [dim]{annotation}[/dim]" if annotation else ""
+        node.add(f"[green]{parts[-1]}[/green]{suffix}")
+
+    console.print()
+    console.print(tree)
+
+
+# ---------------------------------------------------------------------------
+# Command
+# ---------------------------------------------------------------------------
+
+
+def init_command(
+    project_dir: Path,
+    *,
+    github_actions: bool,
+    overwrite: bool,
+    example: bool = True,
+) -> None:
+    """Scaffold a Flow DevOps project at *project_dir*."""
+    target = project_dir.resolve()
+
+    if target.exists() and not overwrite:
+        existing = [p for p in target.iterdir() if p.name != ".git"]
+        if existing:
+            msg = f"{target} already exists and is not empty. Use [bold]--overwrite[/bold] to scaffold into it anyway."
+            console.print(f"[red]Error:[/red] {msg}")
+            raise typer.Exit(1)
+
+    target.mkdir(parents=True, exist_ok=True)
+    created: list[tuple[str, str]] = []
+
+    kw: dict[str, Any] = {"target": target, "overwrite": overwrite, "created": created}
+
+    # flows/
+    (target / "flows").mkdir(exist_ok=True)
+    if example:
+        from lfx.cli.create import create_command as _create
+
+        try:
+            _create(
+                "hello-world",
+                template="hello-world",
+                output_dir=target / "flows",
+                overwrite=overwrite,
+            )
+            created.append(("flows/hello-world.json", "starter flow — edit or replace"))
+        except (OSError, ValueError, TypeError, RuntimeError) as exc:
+            # Don't let a template failure block the rest of init
+            console.print(f"[yellow]Warning:[/yellow] Could not seed starter flow: {exc}")
+    else:
+        _write(target / "flows" / ".gitkeep", "", "versioned empty directory", **kw)
+
+    # tests/
+    _write(target / "tests" / "__init__.py", "", "", **kw)
+    _write(target / "tests" / "test_flows.py", _TEST_FLOWS_PY, "flow_runner example tests", **kw)
+
+    # .lfx/environments.yaml
+    _write(
+        target / ".lfx" / "environments.yaml",
+        _ENVIRONMENTS_YAML,
+        "edit with your instance URLs + API key env var names (safe to commit)",
+        **kw,
+    )
+
+    # .gitignore — keep langflow-environments.toml ignored for backward compat
+    gitignore = target / ".gitignore"
+    if gitignore.exists():
+        existing_content = gitignore.read_text(encoding="utf-8")
+        if "langflow-environments.toml" not in existing_content:
+            gitignore.write_text(existing_content.rstrip() + "\n\n" + _GITIGNORE, encoding="utf-8")
+            created.append((".gitignore", "appended credentials ignore rule"))
+    else:
+        _write(gitignore, _GITIGNORE, "ignores legacy credentials file", **kw)
+
+    # GitHub Actions CI workflows
+    if github_actions:
+        gha_src = _TEMPLATES_DIR / "github-actions"
+        if gha_src.exists():
+            for tmpl in sorted(gha_src.glob("*.yml")):
+                dest = target / ".github" / "workflows" / tmpl.name
+                _copy_template(tmpl, dest, "CI workflow", created, target=target, overwrite=overwrite)
+        else:
+            console.print("[yellow]Warning:[/yellow] GitHub Actions templates not found; skipping.")
+
+    # Generic shell CI scripts (always scaffolded — work with any CI system)
+    shell_src = _TEMPLATES_DIR / "shell"
+    if shell_src.exists():
+        for tmpl in sorted(shell_src.glob("*.sh")):
+            dest = target / "ci" / tmpl.name
+            _copy_template(tmpl, dest, "generic CI script", created, target=target, overwrite=overwrite)
+            dest.chmod(dest.stat().st_mode | 0o111)  # ensure executable bit
+
+    # Print the created-files tree
+    _render_tree(target, created)
+
+    # Next-steps guide
+    console.print()
+    console.print("[bold green]✓ Project scaffolded.[/bold green]  Next steps:\n")
+    console.print("  1. Edit [bold].lfx/environments.yaml[/bold] with your instance URL")
+    console.print("  2. [bold]export LANGFLOW_LOCAL_API_KEY=<key>[/bold]   (Settings → API Keys)")
+    if example:
+        console.print("  3. [bold]lfx validate flows/hello-world.json[/bold]  (check the starter flow)")
+        console.print("  4. [bold]lfx serve flows/hello-world.json[/bold]     (run it locally)")
+        console.print("  5. [bold]lfx push --dir flows/ --env local[/bold]    (deploy to Langflow)")
+    else:
+        console.print("  3. [bold]lfx create my-flow --template hello-world[/bold]")
+        console.print("  4. [bold]lfx push --dir flows/ --env local[/bold]")
+    console.print(f"  {'6' if example else '5'}. [bold]pytest tests/ --langflow-env local[/bold]")
+    console.print()

@@ -0,0 +1,215 @@
+"""lfx login -- validate credentials against a remote Langflow instance.
+
+Tests whether the configured URL and API key are reachable and accepted,
+then prints a summary with guidance on how to fix any problems.
+
+Usage examples
+--------------
+Validate a named environment::
+
+    lfx login --env staging
+
+Validate inline credentials::
+
+    lfx login --target https://langflow.example.com --api-key sk-abc123
+
+Validate whichever environment is the default in your config::
+
+    lfx login
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+import typer
+from rich.console import Console
+from rich.panel import Panel
+
+from lfx.cli.common import load_sdk
+
+console = Console()
+err_console = Console(stderr=True)
+
+_KEY_MASK = 8  # characters shown before "..."
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _mask_key(key: str) -> str:
+    if len(key) <= _KEY_MASK:
+        return "***"
+    return key[:_KEY_MASK] + "..."
+
+
+def _api_key_env_name(env_name: str, environments_file: str | None) -> str | None:
+    """Return the ``api_key_env`` var name for *env_name* from the config file, or ``None``."""
+    try:
+        from lfx.config.environments import _find_config_file, _load_config
+
+        override = Path(environments_file) if environments_file else None
+        config_path = _find_config_file(override)
+        if config_path is None:
+            return None
+        _load_config(config_path)
+
+        # Re-read raw to get api_key_env
+        from lfx.config.environments import _load_raw
+
+        raw = _load_raw(config_path)
+        block = (raw.get("environments") or {}).get(env_name, {})
+        return block.get("api_key_env") if isinstance(block, dict) else None
+    except Exception:  # noqa: BLE001
+        return None
+
+
+def _probe_connection(client: Any, sdk: Any) -> tuple[bool, str, int]:
+    """Attempt a lightweight API call. Return ``(ok, message, flow_count)``."""
+    try:
+        flows = client.list_flows(page=1, size=1)
+        return True, "OK", len(flows)
+    except sdk.LangflowAuthError:
+        return False, "auth", 0
+    except sdk.LangflowConnectionError as exc:
+        return False, f"connection:{exc}", 0
+    except sdk.LangflowHTTPError as exc:
+        return False, f"http:{exc}", 0
+    except Exception as exc:  # noqa: BLE001
+        # A Pydantic ValidationError means the HTTP request completed and was
+        # authenticated (auth is fine) but the SDK's schema doesn't match the
+        # local Langflow version exactly.  Treat this as a successful probe.
+        if "ValidationError" in type(exc).__qualname__:
+            return True, "OK", 0
+        return False, f"error:{exc}", 0
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
+
+
+def login_command(
+    *,
+    env: str | None,
+    environments_file: str | None,
+    target: str | None,
+    api_key: str | None,
+) -> None:
+    sdk = load_sdk("login")
+
+    from lfx.config import ConfigError, resolve_environment
+
+    try:
+        env_cfg = resolve_environment(
+            env,
+            target=target,
+            api_key=api_key,
+            environments_file=environments_file,
+        )
+    except ConfigError as exc:
+        err_console.print(f"[red]Configuration error:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+    # ------------------------------------------------------------------
+    # Key guidance (before probing)
+    # ------------------------------------------------------------------
+    key_env_name: str | None = None
+    if env_cfg.api_key is None and env_cfg.name not in ("__inline__", "__env__"):
+        key_env_name = _api_key_env_name(env_cfg.name, environments_file)
+
+    if env_cfg.api_key is None:
+        warning_parts = ["[yellow]Warning:[/yellow] No API key configured."]
+        if key_env_name:
+            warning_parts.append(f"  Set [bold]export {key_env_name}=<your-key>[/bold] then retry.")
+        else:
+            warning_parts.append("  Add [bold]api_key_env: LANGFLOW_<ENV>_API_KEY[/bold] to your config,")
+            warning_parts.append("  then set that environment variable to your API key.")
+        for line in warning_parts:
+            err_console.print(line)
+
+    # ------------------------------------------------------------------
+    # Probe the connection
+    # ------------------------------------------------------------------
+    client = sdk.Client(base_url=env_cfg.url, api_key=env_cfg.api_key)
+
+    console.print(f"[dim]Connecting to[/dim] {env_cfg.url} …")
+    ok, msg, _flow_count = _probe_connection(client, sdk)
+
+    if not ok:
+        if msg == "auth":
+            err_console.print()
+            err_console.print("[red bold]✗ Authentication failed.[/red bold]")
+            err_console.print(f"  URL:  {env_cfg.url}")
+            if env_cfg.api_key:
+                err_console.print(f"  Key:  {_mask_key(env_cfg.api_key)}")
+            err_console.print()
+            err_console.print("[bold]How to fix:[/bold]")
+            if key_env_name:
+                err_console.print("  1. Open Langflow → Settings → API Keys → Create a new key")
+                err_console.print(f"  2. [bold]export {key_env_name}=<your-new-key>[/bold]")
+            elif env_cfg.name not in ("__inline__", "__env__"):
+                err_console.print("  1. Open Langflow → Settings → API Keys → Create a new key")
+                err_console.print("  2. Pass [bold]--api-key <key>[/bold] or configure api_key_env in your YAML")
+            else:
+                err_console.print("  1. Open Langflow → Settings → API Keys → Create a new key")
+                err_console.print("  2. Pass [bold]--api-key <key>[/bold]")
+            raise typer.Exit(1)
+
+        if msg.startswith("connection:"):
+            err_console.print()
+            err_console.print("[red bold]✗ Cannot connect.[/red bold]")
+            err_console.print(f"  URL: {env_cfg.url}")
+            err_console.print()
+            err_console.print("[bold]How to fix:[/bold]")
+            err_console.print("  • Make sure your Langflow instance is running")
+            err_console.print("  • Check the URL in your .lfx/environments.yaml")
+            err_console.print("  • If running locally: [bold]langflow run[/bold] or [bold]lfx serve <flow.json>[/bold]")
+            raise typer.Exit(1)
+
+        # Generic HTTP or other error
+        err_console.print(f"\n[red bold]✗ Request failed:[/red bold] {msg.split(':', 1)[-1]}")
+        raise typer.Exit(1)
+
+    # ------------------------------------------------------------------
+    # Success
+    # ------------------------------------------------------------------
+    masked_key = _mask_key(env_cfg.api_key) if env_cfg.api_key else "[dim](none)[/dim]"
+
+    key_source = ""
+    if key_env_name and env_cfg.api_key:
+        key_source = f"  [dim]from env var {key_env_name}[/dim]"
+    elif os.environ.get("LANGFLOW_API_KEY") == env_cfg.api_key and env_cfg.api_key:
+        key_source = "  [dim]from LANGFLOW_API_KEY[/dim]"
+
+    env_label = env_cfg.name if env_cfg.name not in ("__inline__", "__env__") else "(inline)"
+
+    console.print()
+    console.print(
+        Panel.fit(
+            f"[bold green]✓ Connected successfully![/bold green]\n\n"
+            f"[bold]Environment:[/bold] {env_label}\n"
+            f"[bold]URL:[/bold]         {env_cfg.url}\n"
+            f"[bold]API key:[/bold]     {masked_key}{key_source}",
+            title="[bold blue]lfx login[/bold blue]",
+            border_style="green",
+        )
+    )
+
+    if not env_cfg.api_key:
+        console.print()
+        console.print("[dim]Note: connected without an API key (anonymous access).[/dim]")
+        console.print("[dim]Some operations may fail. Add api_key_env to your config.[/dim]")
+
+    if env_cfg.name in ("__inline__", "__env__") and env_cfg.api_key:
+        console.print()
+        console.print("[dim]Tip: to avoid passing credentials each time, add to .lfx/environments.yaml:[/dim]")
+        env_display = env or "myenv"
+        console.print("[dim]  environments:[/dim]")
+        console.print(f"[dim]    {env_display}:[/dim]")
+        console.print(f"[dim]      url: {env_cfg.url}[/dim]")
+        console.print(f"[dim]      api_key_env: LANGFLOW_{env_display.upper()}_API_KEY[/dim]")

@@ -0,0 +1,252 @@
+"""lfx pull -- fetch flows from a remote Langflow instance to local files.
+
+Downloads flows from a live Langflow instance, normalizes them for version
+control, and writes them to a local directory.  Repeated pulls are safe:
+existing files are overwritten with the latest remote state.
+
+Usage examples
+--------------
+Pull all flows from staging to ``flows/``::
+
+    lfx pull --env staging
+
+Pull all flows in a named project::
+
+    lfx pull --env staging --project "My RAG Pipeline"
+
+Pull a single flow by UUID::
+
+    lfx pull --env staging --flow-id <uuid>
+
+Pull to a custom directory::
+
+    lfx pull --env production --output-dir ./local-flows/
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from lfx.cli.common import load_sdk, safe_filename
+from lfx.log.logger import logger
+
+console = Console(stderr=True)
+ok_console = Console()
+
+
+# ---------------------------------------------------------------------------
+# Result types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class PullResult:
+    flow_id: UUID
+    flow_name: str
+    path: Path
+    status: str  # "unchanged" | "updated" | "created" | "error"
+    error: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.status != "error"
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _write_flow(
+    flow_obj: Any,
+    *,
+    sdk: Any,
+    dest_dir: Path,
+    strip_secrets: bool,
+    indent: int,
+) -> PullResult:
+    """Normalize and write a single flow to *dest_dir*."""
+    flow_id = flow_obj.id
+    flow_name = flow_obj.name
+
+    try:
+        normalized = sdk.normalize_flow(
+            flow_obj.model_dump(mode="json"),
+            strip_volatile=True,
+            strip_secrets=strip_secrets,
+            sort_keys=True,
+        )
+        safe_name = safe_filename(flow_name)
+        out_path = dest_dir / f"{safe_name}.json"
+        new_content = sdk.flow_to_json(normalized, indent=indent)
+        if out_path.exists() and out_path.read_text(encoding="utf-8") == new_content:
+            status = "unchanged"
+        else:
+            status = "created" if not out_path.exists() else "updated"
+            out_path.write_text(new_content, encoding="utf-8")
+        return PullResult(flow_id=flow_id, flow_name=flow_name, path=out_path, status=status)
+    except Exception as exc:  # noqa: BLE001
+        logger.debug("Failed to write flow %s", flow_id, exc_info=True)
+        dummy_path = dest_dir / f"{flow_id}.json"
+        return PullResult(flow_id=flow_id, flow_name=flow_name, path=dummy_path, status="error", error=str(exc))
+
+
+def _render_results(results: list[PullResult]) -> None:
+    table = Table(show_header=True, header_style="bold")
+    table.add_column("Name")
+    table.add_column("ID")
+    table.add_column("File")
+    table.add_column("Status")
+
+    status_style = {
+        "unchanged": ("dim", "UNCHANGED"),
+        "updated": ("yellow", "UPDATED"),
+        "created": ("green", "CREATED"),
+        "error": ("red", "ERROR"),
+    }
+
+    for r in results:
+        color, label = status_style.get(r.status, ("white", r.status.upper()))
+        if r.error:
+            label += f": {r.error}"
+        table.add_row(r.flow_name, str(r.flow_id), str(r.path), f"[{color}]{label}[/{color}]")
+
+    ok_console.print()
+    ok_console.print(table)
+
+    errors = [r for r in results if not r.ok]
+    n_changed = sum(1 for r in results if r.status in ("created", "updated"))
+    n_unchanged = sum(1 for r in results if r.status == "unchanged")
+    if errors:
+        console.print(f"\n[red]{len(errors)} pull(s) failed.[/red]")
+    else:
+        parts = []
+        if n_changed:
+            parts.append(f"[green]{n_changed} updated[/green]")
+        if n_unchanged:
+            parts.append(f"[dim]{n_unchanged} unchanged[/dim]")
+        ok_console.print("\n" + ", ".join(parts) + ".")
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
+
+
+def pull_command(
+    *,
+    env: str | None,
+    output_dir: str | None,
+    flow_id: str | None,
+    project: str | None,
+    project_id: str | None,
+    environments_file: str | None,
+    target: str | None = None,
+    api_key: str | None = None,
+    strip_secrets: bool,
+    indent: int,
+) -> None:
+    sdk = load_sdk("pull")
+
+    from lfx.config import ConfigError, resolve_environment
+
+    try:
+        env_cfg = resolve_environment(
+            env,
+            target=target,
+            api_key=api_key,
+            environments_file=environments_file,
+        )
+    except ConfigError as exc:
+        console.print(f"[red]Error:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+    client = sdk.Client(base_url=env_cfg.url, api_key=env_cfg.api_key)
+
+    dest_dir = Path(output_dir) if output_dir else Path("flows")
+    dest_dir.mkdir(parents=True, exist_ok=True)
+
+    results: list[PullResult] = []
+
+    # ---- Single flow by ID -----------------------------------------------
+    if flow_id:
+        try:
+            flow_obj = client.get_flow(UUID(flow_id))
+        except Exception as exc:
+            console.print(f"[red]Error:[/red] Could not fetch flow {flow_id}: {exc}")
+            raise typer.Exit(1) from exc
+
+        result = _write_flow(flow_obj, sdk=sdk, dest_dir=dest_dir, strip_secrets=strip_secrets, indent=indent)
+        results.append(result)
+        if result.ok:
+            console.print(f"[green]Pulled[/green] {result.flow_name!r} → {result.path}")
+        else:
+            console.print(f"[red]Failed[/red] {result.flow_name!r}: {result.error}")
+
+    # ---- All flows in a named project ------------------------------------
+    elif project or project_id:
+        if project_id:
+            try:
+                proj = client.get_project(UUID(project_id))
+            except Exception as exc:
+                console.print(f"[red]Error:[/red] Could not fetch project {project_id}: {exc}")
+                raise typer.Exit(1) from exc
+        else:
+            projects = client.list_projects()
+            matched = [p for p in projects if p.name == project]
+            if not matched:
+                names = ", ".join(repr(p.name) for p in projects) or "(none)"
+                console.print(f"[red]Error:[/red] Project {project!r} not found. Available: {names}")
+                raise typer.Exit(1)
+            try:
+                proj = client.get_project(matched[0].id)
+            except Exception as exc:
+                console.print(f"[red]Error:[/red] Could not fetch project {project!r}: {exc}")
+                raise typer.Exit(1) from exc
+
+        console.print(f"[dim]Pulling from project[/dim] {proj.name!r} (id={proj.id})")
+
+        for flow_obj in proj.flows:
+            result = _write_flow(flow_obj, sdk=sdk, dest_dir=dest_dir, strip_secrets=strip_secrets, indent=indent)
+            results.append(result)
+            if result.status == "unchanged":
+                console.print(f"[dim]Unchanged[/dim] {result.flow_name!r}")
+            elif result.ok:
+                console.print(f"[green]Pulled[/green] {result.flow_name!r} → {result.path}")
+            else:
+                console.print(f"[red]Failed[/red] {result.flow_name!r}: {result.error}")
+
+    # ---- All flows in the environment ------------------------------------
+    else:
+        console.print(f"[dim]Pulling all flows from[/dim] {env_cfg.url}")
+        try:
+            flows = client.list_flows(get_all=True, remove_example_flows=True)
+        except Exception as exc:
+            console.print(f"[red]Error:[/red] Could not list flows: {exc}")
+            raise typer.Exit(1) from exc
+
+        if not flows:
+            console.print("[yellow]Warning:[/yellow] No flows found on the remote instance.")
+            return
+
+        for flow_obj in flows:
+            result = _write_flow(flow_obj, sdk=sdk, dest_dir=dest_dir, strip_secrets=strip_secrets, indent=indent)
+            results.append(result)
+            if result.status == "unchanged":
+                console.print(f"[dim]Unchanged[/dim] {result.flow_name!r}")
+            elif result.ok:
+                console.print(f"[green]Pulled[/green] {result.flow_name!r} → {result.path}")
+            else:
+                console.print(f"[red]Failed[/red] {result.flow_name!r}: {result.error}")
+
+    _render_results(results)
+
+    if any(not r.ok for r in results):
+        raise typer.Exit(1)

@@ -0,0 +1,419 @@
+"""lfx push -- push normalized flow JSON to a remote Langflow instance.
+
+Uses stable flow IDs for upsert (PUT /api/v1/flows/{id}), so repeated pushes
+are idempotent: the first push creates the flow, subsequent ones update it in
+place without changing its ID on the remote instance.
+
+Usage examples
+--------------
+Push a single flow to staging::
+
+    lfx push my_flow.json --env staging
+
+Push several flows at once::
+
+    lfx push flows/*.json --env staging
+
+Push all flows in a directory and place them in a named project::
+
+    lfx push --dir ./flows/ --env staging --project "My RAG Pipeline"
+
+Dry-run to see what would happen::
+
+    lfx push ./flows/ --env production --dry-run
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from lfx.cli.common import load_sdk
+
+console = Console(stderr=True)
+ok_console = Console()
+
+
+# ---------------------------------------------------------------------------
+# Result types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class PushResult:
+    path: Path
+    flow_id: UUID
+    flow_name: str
+    status: str  # "created" | "updated" | "unchanged" | "error" | "dry-run"
+    error: str | None = None
+    flow_url: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.status in ("created", "updated", "unchanged", "dry-run")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _load_flow_file(path: Path) -> dict[str, Any]:
+    """Read and parse a flow JSON file; raise Exit on error."""
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        console.print(f"[red]Error:[/red] Cannot read {path}: {exc}")
+        raise typer.Exit(1) from exc
+
+
+def _extract_flow_id(flow: dict[str, Any], path: Path) -> UUID:
+    """Extract and validate the flow's stable ID from the JSON."""
+    raw_id = flow.get("id")
+    if not raw_id:
+        console.print(f"[red]Error:[/red] {path} has no 'id' field. Run [bold]lfx export[/bold] first.")
+        raise typer.Exit(1)
+    try:
+        return UUID(str(raw_id))
+    except ValueError:
+        console.print(f"[red]Error:[/red] {path} has an invalid 'id': {raw_id!r}")
+        raise typer.Exit(1)  # noqa: B904
+
+
+def _flow_to_create(sdk: Any, flow: dict[str, Any], folder_id: UUID | None) -> Any:
+    """Build a FlowCreate from a normalized flow dict."""
+    return sdk.FlowCreate(
+        name=flow.get("name", "Untitled"),
+        description=flow.get("description"),
+        data=flow.get("data"),
+        is_component=flow.get("is_component", False),
+        endpoint_name=flow.get("endpoint_name"),
+        tags=flow.get("tags"),
+        folder_id=folder_id or (UUID(flow["folder_id"]) if flow.get("folder_id") else None),
+        icon=flow.get("icon"),
+        icon_bg_color=flow.get("icon_bg_color"),
+        locked=flow.get("locked", False),
+        mcp_enabled=flow.get("mcp_enabled", False),
+    )
+
+
+def _upsert_single(
+    client: Any,
+    sdk: Any,
+    path: Path,
+    flow_id: UUID,
+    flow_create: Any,
+    *,
+    dry_run: bool,
+    flow_name: str,
+    base_url: str,
+    local_file_content: str | None = None,
+    strip_secrets: bool = True,
+) -> PushResult:
+    flow_url = f"{base_url.rstrip('/')}/flow/{flow_id}"
+
+    if dry_run:
+        return PushResult(path=path, flow_id=flow_id, flow_name=flow_name, status="dry-run", flow_url=flow_url)
+
+    # Compare normalized remote against local file to detect unchanged flows,
+    # avoiding a spurious PUT when nothing has actually changed.
+    # Import directly from serialization so this internal comparison is not
+    # counted as a call to the public sdk.normalize_flow (keeps tests clean).
+    if local_file_content is not None:
+        try:
+            # Use direct module imports (not sdk.*) so mock call-counts in tests
+            # stay accurate and so the except clause uses a real exception class.
+            from langflow_sdk.exceptions import LangflowNotFoundError
+            from langflow_sdk.serialization import flow_to_json, normalize_flow
+
+            remote = client.get_flow(flow_id)
+            remote_normalized = normalize_flow(
+                remote.model_dump(mode="json"),
+                strip_volatile=True,
+                strip_secrets=strip_secrets,
+                sort_keys=True,
+            )
+            if flow_to_json(remote_normalized) == local_file_content:
+                return PushResult(
+                    path=path,
+                    flow_id=flow_id,
+                    flow_name=flow_name,
+                    status="unchanged",
+                    flow_url=flow_url,
+                )
+        except LangflowNotFoundError:
+            pass  # Flow doesn't exist yet — fall through to create it
+        except Exception:  # noqa: BLE001
+            import logging
+
+            logging.getLogger(__name__).debug("Remote comparison failed; proceeding with push", exc_info=True)
+
+    try:
+        _, created = client.upsert_flow(flow_id, flow_create)
+        status = "created" if created else "updated"
+        return PushResult(path=path, flow_id=flow_id, flow_name=flow_name, status=status, flow_url=flow_url)
+    except sdk.LangflowHTTPError as exc:
+        return PushResult(
+            path=path,
+            flow_id=flow_id,
+            flow_name=flow_name,
+            status="error",
+            error=str(exc),
+            flow_url=flow_url,
+        )
+
+
+def _find_or_create_project(
+    client: Any,
+    sdk: Any,
+    project_name: str,
+    *,
+    dry_run: bool,
+) -> UUID | None:
+    """Return the UUID of a project with *project_name*, creating if needed.
+
+    Returns ``None`` in dry-run mode (project may not exist yet).
+    """
+    projects = client.list_projects()
+    for p in projects:
+        if p.name == project_name:
+            console.print(f"[dim]Project[/dim] {project_name!r} found (id={p.id})")
+            return p.id
+
+    if dry_run:
+        console.print(f"[dim]Project[/dim] {project_name!r} would be created (dry-run)")
+        return None
+
+    project = client.create_project(sdk.ProjectCreate(name=project_name))
+    console.print(f"[green]Created project[/green] {project_name!r} (id={project.id})")
+    return project.id
+
+
+def _find_project_root() -> Path | None:
+    """Return the lfx project root (directory containing .lfx/), or None.
+
+    Only the ``.lfx`` marker is used — ``.git`` is intentionally excluded so
+    that the containment check doesn't reject paths when running inside a
+    larger monorepo or outside any lfx project.
+    """
+    cwd = Path.cwd()
+    for directory in (cwd, *cwd.parents):
+        if (directory / ".lfx").is_dir():
+            return directory
+        # Stop at a filesystem root
+        if directory.parent == directory:
+            break
+    return None
+
+
+def _check_path_containment(p: Path, root: Path | None) -> None:
+    """Ensure *p* is inside *root*; skip check when no project root is found."""
+    if root is None:
+        return
+    try:
+        p.resolve().relative_to(root.resolve())
+    except ValueError:
+        console.print(
+            f"[red]Error:[/red] Path {p} is outside the project root ({root}). "
+            "Refusing to push files from outside the project."
+        )
+        raise typer.Exit(1)  # noqa: B904
+
+
+def _collect_flow_files(sources: list[str], dir_path: str | None) -> list[Path]:
+    """Resolve the set of flow JSON files to push.
+
+    When neither explicit file paths nor ``--dir`` are given, defaults to
+    ``flows/`` — mirroring the behaviour of ``lfx pull``.
+    """
+    paths: list[Path] = []
+    root = _find_project_root()
+
+    # Default to flows/ when nothing is specified, just like lfx pull does.
+    effective_dir = dir_path or (None if sources else "flows")
+
+    if effective_dir:
+        d = Path(effective_dir)
+        _check_path_containment(d, root)
+        if not d.is_dir():
+            console.print(f"[red]Error:[/red] Directory not found: {d}")
+            raise typer.Exit(1)
+        paths.extend(sorted(d.glob("*.json")))
+        if not paths:
+            console.print(f"[yellow]Warning:[/yellow] No *.json files found in {d}")
+
+    for s in sources:
+        p = Path(s)
+        _check_path_containment(p, root)
+        if not p.exists():
+            console.print(f"[red]Error:[/red] File not found: {p}")
+            raise typer.Exit(1)
+        if p.is_dir():
+            dir_jsons = sorted(p.glob("*.json"))
+            if not dir_jsons:
+                console.print(f"[yellow]Warning:[/yellow] No *.json files found in {p}")
+            paths.extend(dir_jsons)
+        else:
+            paths.append(p)
+
+    return paths
+
+
+def _render_results(results: list[PushResult], *, dry_run: bool) -> None:
+    table = Table(show_header=True, header_style="bold")
+    table.add_column("File")
+    table.add_column("Name")
+    table.add_column("ID")
+    table.add_column("Status")
+    table.add_column("URL")
+
+    status_colors = {
+        "created": "green",
+        "updated": "cyan",
+        "unchanged": "dim",
+        "dry-run": "yellow",
+        "error": "red",
+    }
+
+    for r in results:
+        color = status_colors.get(r.status, "white")
+        label = r.status.upper() + (f": {r.error}" if r.error else "")
+        url_cell = f"[blue]{r.flow_url}[/blue]" if r.flow_url and r.ok else (r.flow_url or "")
+        table.add_row(
+            str(r.path),
+            r.flow_name,
+            str(r.flow_id),
+            f"[{color}]{label}[/{color}]",
+            url_cell,
+        )
+
+    ok_console.print()
+    ok_console.print(table)
+
+    errors = [r for r in results if not r.ok]
+    if errors:
+        console.print(f"\n[red]{len(errors)} push(es) failed.[/red]")
+    elif dry_run:
+        ok_console.print(f"\n[yellow]{len(results)} flow(s) would be pushed (dry-run).[/yellow]")
+    else:
+        created = sum(1 for r in results if r.status == "created")
+        updated = sum(1 for r in results if r.status == "updated")
+        unchanged = sum(1 for r in results if r.status == "unchanged")
+        parts = []
+        if created:
+            parts.append(f"[green]{created} created[/green]")
+        if updated:
+            parts.append(f"[cyan]{updated} updated[/cyan]")
+        if unchanged:
+            parts.append(f"[dim]{unchanged} unchanged[/dim]")
+        ok_console.print("\n" + ", ".join(parts) + ".")
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
+
+
+def push_command(
+    flow_paths: list[str],
+    *,
+    env: str | None,
+    dir_path: str | None,
+    project: str | None,
+    project_id: str | None,
+    environments_file: str | None,
+    target: str | None = None,
+    api_key: str | None = None,
+    dry_run: bool,
+    normalize: bool,
+    strip_secrets: bool,
+) -> None:
+    sdk = load_sdk("push")
+
+    from lfx.config import ConfigError, resolve_environment
+
+    try:
+        env_cfg = resolve_environment(
+            env,
+            target=target,
+            api_key=api_key,
+            environments_file=environments_file,
+        )
+    except ConfigError as exc:
+        console.print(f"[red]Error:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+    client = sdk.Client(base_url=env_cfg.url, api_key=env_cfg.api_key)
+
+    paths = _collect_flow_files(flow_paths, dir_path)
+    if not paths:
+        console.print(
+            "[red]Error:[/red] No *.json flow files found. "
+            "Run [bold]lfx pull[/bold] first, or pass explicit file paths."
+        )
+        raise typer.Exit(1)
+
+    # Resolve target project folder_id
+    target_folder_id: UUID | None = None
+    if project_id:
+        target_folder_id = UUID(project_id)
+    elif project:
+        target_folder_id = _find_or_create_project(client, sdk, project, dry_run=dry_run)
+
+    results: list[PushResult] = []
+
+    for path in paths:
+        raw_flow = _load_flow_file(path)
+
+        if normalize:
+            raw_flow = sdk.normalize_flow(
+                raw_flow,
+                strip_volatile=True,
+                strip_secrets=strip_secrets,
+                sort_keys=True,
+            )
+
+        flow_id = _extract_flow_id(raw_flow, path)
+        flow_name = raw_flow.get("name", path.stem)
+        flow_create = _flow_to_create(sdk, raw_flow, target_folder_id)
+        # Capture normalized content now so _upsert_single can compare against remote.
+        local_file_content = sdk.flow_to_json(raw_flow) if normalize else None
+
+        result = _upsert_single(
+            client,
+            sdk,
+            path,
+            flow_id,
+            flow_create,
+            dry_run=dry_run,
+            flow_name=flow_name,
+            base_url=env_cfg.url,
+            local_file_content=local_file_content,
+            strip_secrets=strip_secrets,
+        )
+        results.append(result)
+
+        if dry_run:
+            console.print(f"[yellow]DRY-RUN[/yellow] Would push {flow_name!r} ({flow_id})")
+        elif result.status == "unchanged":
+            console.print(f"[dim]Unchanged[/dim] {flow_name!r}")
+        elif result.status == "created":
+            url_hint = f"  [dim]{result.flow_url}[/dim]" if result.flow_url else ""
+            console.print(f"[green]Created[/green]  {flow_name!r} ({flow_id}){url_hint}")
+        elif result.status == "updated":
+            url_hint = f"  [dim]{result.flow_url}[/dim]" if result.flow_url else ""
+            console.print(f"[cyan]Updated[/cyan]  {flow_name!r} ({flow_id}){url_hint}")
+        else:
+            console.print(f"[red]Failed[/red]   {flow_name!r} ({flow_id}): {result.error}")
+
+    _render_results(results, dry_run=dry_run)
+
+    if any(not r.ok for r in results):
+        raise typer.Exit(1)

@@ -0,0 +1,322 @@
+"""Remote commands: status, push, pull, export."""
+
+import typer
+
+
+def register(app: typer.Typer) -> None:
+    """Register remote-stage commands on *app*."""
+
+    @app.command(
+        name="status", help="Compare local flow files against a remote Langflow instance", rich_help_panel="Remote"
+    )
+    def status_command_wrapper(
+        flow_paths: list[str] = typer.Argument(
+            default=None,
+            help="Specific flow JSON file(s) to check. Omit to scan --dir (default: flows/).",
+        ),
+        env: str | None = typer.Option(
+            None,
+            "--env",
+            "-e",
+            help="Environment name from .lfx/environments.yaml. Uses [defaults] if omitted.",
+        ),
+        dir_path: str | None = typer.Option(
+            None,
+            "--dir",
+            "-d",
+            help="Directory of flow JSON files to compare (default: flows/ in cwd).",
+        ),
+        environments_file: str | None = typer.Option(
+            None,
+            "--environments-file",
+            help="Path to environments config file (.yaml or .toml; overrides default lookup).",
+        ),
+        target: str | None = typer.Option(
+            None,
+            "--target",
+            help="Langflow instance URL (inline override — skips config file lookup).",
+        ),
+        api_key: str | None = typer.Option(
+            None,
+            "--api-key",
+            help="API key for the Langflow instance (used with --target or to override config).",
+        ),
+        show_remote_only: bool = typer.Option(
+            False,
+            "--remote-only",
+            help="Also list flows that exist on the server but have no local file.",
+        ),
+    ) -> None:
+        """Show whether local flow files are in sync, ahead, or missing vs the remote instance."""
+        from lfx.cli.status import status_command
+
+        status_command(
+            dir_path=dir_path,
+            flow_paths=flow_paths or [],
+            env=env,
+            environments_file=environments_file,
+            target=target,
+            api_key=api_key,
+            show_remote_only=show_remote_only,
+        )
+
+    @app.command(
+        name="push",
+        help="Push flow JSON to a remote Langflow instance (upsert by stable ID)",
+        rich_help_panel="Remote",
+    )
+    def push_command_wrapper(
+        flow_paths: list[str] = typer.Argument(
+            default=None,
+            help="Path(s) to flow JSON file(s) to push. Use --dir for a whole directory.",
+        ),
+        env: str | None = typer.Option(
+            None,
+            "--env",
+            "-e",
+            help="Environment name from .lfx/environments.yaml. Use --target for inline configuration.",
+        ),
+        dir_path: str | None = typer.Option(
+            None,
+            "--dir",
+            "-d",
+            help="Directory of flow JSON files to push (pushes all *.json files). Defaults to flows/.",
+        ),
+        project: str | None = typer.Option(
+            None,
+            "--project",
+            "-p",
+            help="Target project name on the remote instance. Created if it does not exist.",
+        ),
+        project_id: str | None = typer.Option(
+            None,
+            "--project-id",
+            help="Target project UUID (alternative to --project).",
+        ),
+        environments_file: str | None = typer.Option(
+            None,
+            "--environments-file",
+            help="Path to environments config file (.yaml or .toml; overrides default lookup).",
+        ),
+        target: str | None = typer.Option(
+            None,
+            "--target",
+            help="Langflow instance URL (inline override — skips config file lookup).",
+        ),
+        api_key: str | None = typer.Option(
+            None,
+            "--api-key",
+            help="API key for the Langflow instance (used with --target or to override config).",
+        ),
+        dry_run: bool = typer.Option(
+            False,
+            "--dry-run",
+            help="Show what would be pushed without making any changes.",
+        ),
+        normalize: bool = typer.Option(
+            True,
+            "--normalize/--no-normalize",
+            help="Normalize (strip volatile fields, sort keys) before pushing.",
+        ),
+        strip_secrets: bool = typer.Option(
+            True,
+            "--strip-secrets/--keep-secrets",
+            help="Clear password/load_from_db field values before pushing.",
+        ),
+    ) -> None:
+        """Push Langflow flows to a remote instance using stable IDs for upsert (lazy-loaded)."""
+        from lfx.cli.push import push_command
+
+        push_command(
+            flow_paths=flow_paths or [],
+            env=env,
+            dir_path=dir_path,
+            project=project,
+            project_id=project_id,
+            environments_file=environments_file,
+            target=target,
+            api_key=api_key,
+            dry_run=dry_run,
+            normalize=normalize,
+            strip_secrets=strip_secrets,
+        )
+
+    @app.command(
+        name="pull", help="Pull flows from a remote Langflow instance to local files", rich_help_panel="Remote"
+    )
+    def pull_command_wrapper(
+        env: str | None = typer.Option(
+            None,
+            "--env",
+            "-e",
+            help="Environment name from .lfx/environments.yaml. Uses [defaults] if omitted.",
+        ),
+        output_dir: str | None = typer.Option(
+            None,
+            "--output-dir",
+            "-d",
+            help="Directory to write pulled flows into (default: flows/).",
+        ),
+        flow_id: str | None = typer.Option(
+            None,
+            "--flow-id",
+            help="Pull a single flow by UUID.",
+        ),
+        project: str | None = typer.Option(
+            None,
+            "--project",
+            "-p",
+            help="Pull all flows in a named project.",
+        ),
+        project_id: str | None = typer.Option(
+            None,
+            "--project-id",
+            help="Pull all flows in a project by UUID.",
+        ),
+        environments_file: str | None = typer.Option(
+            None,
+            "--environments-file",
+            help="Path to environments config file (.yaml or .toml; overrides default lookup).",
+        ),
+        target: str | None = typer.Option(
+            None,
+            "--target",
+            help="Langflow instance URL (inline override — skips config file lookup).",
+        ),
+        api_key: str | None = typer.Option(
+            None,
+            "--api-key",
+            help="API key for the Langflow instance (used with --target or to override config).",
+        ),
+        strip_secrets: bool = typer.Option(
+            True,
+            "--strip-secrets/--keep-secrets",
+            help="Clear password/load_from_db field values (default: strip).",
+        ),
+        indent: int = typer.Option(
+            2,
+            "--indent",
+            help="JSON indentation level.",
+        ),
+    ) -> None:
+        """Pull and normalize flows from a remote Langflow instance (lazy-loaded)."""
+        from lfx.cli.pull import pull_command
+
+        pull_command(
+            env=env,
+            output_dir=output_dir,
+            flow_id=flow_id,
+            project=project,
+            project_id=project_id,
+            environments_file=environments_file,
+            target=target,
+            api_key=api_key,
+            strip_secrets=strip_secrets,
+            indent=indent,
+        )
+
+    @app.command(
+        name="export",
+        help="Normalize flow JSON for git (local) or pull from a remote instance",
+        rich_help_panel="Remote",
+    )
+    def export_command_wrapper(
+        flow_paths: list[str] = typer.Argument(
+            default=None,
+            help="Path(s) to local flow JSON file(s) to normalize. Omit when using --flow-id or --project-id.",
+        ),
+        output: str | None = typer.Option(
+            None,
+            "--output",
+            "-o",
+            help="Output file path (single-file local mode only).",
+        ),
+        output_dir: str | None = typer.Option(
+            None,
+            "--output-dir",
+            "-d",
+            help="Directory to write exported flows into (remote mode or multi-file).",
+        ),
+        env: str | None = typer.Option(
+            None,
+            "--env",
+            "-e",
+            help="Environment name from .lfx/environments.yaml (required for remote mode unless --target is used).",
+        ),
+        flow_id: str | None = typer.Option(
+            None,
+            "--flow-id",
+            help="Pull and export a single flow by UUID from the remote instance.",
+        ),
+        project_id: str | None = typer.Option(
+            None,
+            "--project-id",
+            help="Pull and export all flows in a project by UUID from the remote instance.",
+        ),
+        environments_file: str | None = typer.Option(
+            None,
+            "--environments-file",
+            help="Path to environments config file (.yaml or .toml; overrides default lookup).",
+        ),
+        target: str | None = typer.Option(
+            None,
+            "--target",
+            help="Langflow instance URL (inline override — skips config file lookup).",
+        ),
+        api_key: str | None = typer.Option(
+            None,
+            "--api-key",
+            help="API key for the Langflow instance (used with --target or to override config).",
+        ),
+        in_place: bool = typer.Option(
+            False,
+            "--in-place",
+            "-i",
+            help="Overwrite each input file with its normalized version.",
+        ),
+        strip_volatile: bool = typer.Option(
+            True,
+            "--strip-volatile/--keep-volatile",
+            help="Strip instance-specific fields (updated_at, user_id, folder_id).",
+        ),
+        strip_secrets: bool = typer.Option(
+            True,
+            "--strip-secrets/--keep-secrets",
+            help="Clear values of password/load_from_db template fields.",
+        ),
+        code_as_lines: bool = typer.Option(
+            False,
+            "--code-as-lines",
+            help="Convert code-type template field values to a list of lines.",
+        ),
+        strip_node_volatile: bool = typer.Option(
+            True,
+            "--strip-node-volatile/--keep-node-volatile",
+            help="Strip transient node keys (positionAbsolute, dragging, selected).",
+        ),
+        indent: int = typer.Option(
+            2,
+            "--indent",
+            help="JSON indentation level.",
+        ),
+    ) -> None:
+        """Export and normalize Langflow flow JSON for version control (lazy-loaded)."""
+        from lfx.cli.export import export_command
+
+        export_command(
+            flow_paths=flow_paths or [],
+            output=output,
+            output_dir=output_dir,
+            env=env,
+            flow_id=flow_id,
+            project_id=project_id,
+            environments_file=environments_file,
+            target=target,
+            api_key=api_key,
+            in_place=in_place,
+            strip_volatile=strip_volatile,
+            strip_secrets=strip_secrets,
+            code_as_lines=code_as_lines,
+            strip_node_volatile=strip_node_volatile,
+            indent=indent,
+        )

@@ -0,0 +1,143 @@
+"""Running commands: run, serve."""
+
+import typer
+
+
+def register(app: typer.Typer) -> None:
+    """Register running-stage commands on *app*."""
+
+    @app.command(name="run", help="Run a flow directly", no_args_is_help=True, rich_help_panel="Running")
+    def run_command_wrapper(
+        script_path: str | None = typer.Argument(
+            None, help="Path to the Python script (.py) or JSON flow (.json) containing a graph"
+        ),
+        input_value: str | None = typer.Argument(None, help="Input value to pass to the graph"),
+        input_value_option: str | None = typer.Option(
+            None,
+            "--input-value",
+            help="Input value to pass to the graph (alternative to positional argument)",
+        ),
+        output_format: str = typer.Option(
+            "json",
+            "--format",
+            "-f",
+            help="Output format: json, text, message, or result",
+        ),
+        flow_json: str | None = typer.Option(
+            None,
+            "--flow-json",
+            help="Inline JSON flow content as a string (alternative to script_path)",
+        ),
+        *,
+        stdin: bool = typer.Option(
+            default=False,
+            show_default=True,
+            help="Read JSON flow content from stdin (alternative to script_path)",
+        ),
+        check_variables: bool = typer.Option(
+            default=True,
+            show_default=True,
+            help="Check global variables for environment compatibility",
+        ),
+        verbose: bool = typer.Option(
+            False,
+            "-v",
+            "--verbose",
+            help="Show basic progress information",
+        ),
+        verbose_detailed: bool = typer.Option(
+            False,
+            "-vv",
+            help="Show detailed progress and debug information",
+        ),
+        verbose_full: bool = typer.Option(
+            False,
+            "-vvv",
+            help="Show full debugging output including component logs",
+        ),
+        timing: bool = typer.Option(
+            default=False,
+            show_default=True,
+            help="Include detailed timing information in output",
+        ),
+    ) -> None:
+        """Run a flow directly (lazy-loaded)."""
+        from pathlib import Path
+
+        from lfx.cli.run import run
+
+        # Convert script_path string to Path if provided
+        script_path_obj = Path(script_path) if script_path else None
+
+        run(
+            script_path=script_path_obj,
+            input_value=input_value,
+            input_value_option=input_value_option,
+            output_format=output_format,
+            flow_json=flow_json,
+            stdin=stdin,
+            check_variables=check_variables,
+            verbose=verbose,
+            verbose_detailed=verbose_detailed,
+            verbose_full=verbose_full,
+            timing=timing,
+        )
+
+    @app.command(name="serve", help="Serve a flow as an API", no_args_is_help=True, rich_help_panel="Running")
+    def serve_command_wrapper(
+        script_path: str | None = typer.Argument(
+            None,
+            help=(
+                "Path to JSON flow (.json) or Python script (.py) file or stdin input. "
+                "Optional when using --flow-json or --stdin."
+            ),
+        ),
+        host: str = typer.Option("127.0.0.1", "--host", "-h", help="Host to bind the server to"),
+        port: int = typer.Option(8000, "--port", "-p", help="Port to bind the server to"),
+        verbose: bool = typer.Option(False, "--verbose", "-v", help="Show diagnostic output and execution details"),
+        env_file: str | None = typer.Option(
+            None,
+            "--env-file",
+            help="Path to the .env file containing environment variables",
+        ),
+        log_level: str = typer.Option(
+            "warning",
+            "--log-level",
+            help="Logging level. One of: debug, info, warning, error, critical",
+        ),
+        flow_json: str | None = typer.Option(
+            None,
+            "--flow-json",
+            help="Inline JSON flow content as a string (alternative to script_path)",
+        ),
+        *,
+        stdin: bool = typer.Option(
+            False,
+            "--stdin",
+            help="Read JSON flow content from stdin (alternative to script_path)",
+        ),
+        check_variables: bool = typer.Option(
+            True,
+            "--check-variables/--no-check-variables",
+            help="Check global variables for environment compatibility",
+        ),
+    ) -> None:
+        """Serve LFX flows as a web API (lazy-loaded)."""
+        from pathlib import Path
+
+        from lfx.cli.commands import serve_command
+
+        # Convert env_file string to Path if provided
+        env_file_path = Path(env_file) if env_file else None
+
+        serve_command(
+            script_path=script_path,
+            host=host,
+            port=port,
+            verbose=verbose,
+            env_file=env_file_path,
+            log_level=log_level,
+            flow_json=flow_json,
+            stdin=stdin,
+            check_variables=check_variables,
+        )

@@ -0,0 +1,75 @@
+"""Setup commands: init, login."""
+
+import typer
+
+
+def register(app: typer.Typer) -> None:
+    """Register setup-stage commands on *app*."""
+
+    @app.command(name="init", help="Scaffold a new Flow DevOps project", rich_help_panel="Setup")
+    def init_command_wrapper(
+        project_dir: str = typer.Argument(
+            ".",
+            help="Directory to scaffold (created if it does not exist; default: current directory).",
+        ),
+        github_actions: bool = typer.Option(
+            True,
+            "--github-actions/--no-github-actions",
+            help="Copy GitHub Actions workflow templates into .github/workflows/.",
+        ),
+        overwrite: bool = typer.Option(
+            False,
+            "--overwrite",
+            help="Write files even if the target directory already contains files.",
+        ),
+        example: bool = typer.Option(
+            True,
+            "--example/--no-example",
+            help="Seed flows/ with a hello-world.json starter flow (default: true).",
+        ),
+    ) -> None:
+        """Scaffold a Flow DevOps project: flows/, tests/, environments config, and CI templates."""
+        from pathlib import Path
+
+        from lfx.cli.init import init_command
+
+        init_command(
+            project_dir=Path(project_dir),
+            github_actions=github_actions,
+            overwrite=overwrite,
+            example=example,
+        )
+
+    @app.command(name="login", help="Validate credentials against a remote Langflow instance", rich_help_panel="Setup")
+    def login_command_wrapper(
+        env: str | None = typer.Option(
+            None,
+            "--env",
+            "-e",
+            help="Environment name from .lfx/environments.yaml. Uses [defaults] if omitted.",
+        ),
+        environments_file: str | None = typer.Option(
+            None,
+            "--environments-file",
+            help="Path to environments config file (.yaml or .toml; overrides default lookup).",
+        ),
+        target: str | None = typer.Option(
+            None,
+            "--target",
+            help="Langflow instance URL (inline override — skips config file lookup).",
+        ),
+        api_key: str | None = typer.Option(
+            None,
+            "--api-key",
+            help="API key for the Langflow instance (used with --target or to override config).",
+        ),
+    ) -> None:
+        """Test connectivity and authentication for a Langflow environment (lazy-loaded)."""
+        from lfx.cli.login import login_command
+
+        login_command(
+            env=env,
+            environments_file=environments_file,
+            target=target,
+            api_key=api_key,
+        )

@@ -0,0 +1,288 @@
+"""lfx status -- compare local flow files against a remote Langflow instance.
+
+Shows, for each local flow JSON, whether it is in sync with the remote,
+ahead (locally modified), brand new (not yet pushed), or missing entirely.
+Optionally surfaces flows that exist on the server but have no local file.
+
+Examples::
+
+    lfx status                          # scans flows/ in cwd, uses [defaults] env
+    lfx status --env staging            # compare against staging
+    lfx status --dir ./my-flows/        # specify a custom flows directory
+    lfx status --env prod --remote-only # also show server flows not tracked locally
+"""
+
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import json
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import UUID
+
+import typer
+from rich import box
+from rich.console import Console
+from rich.table import Table
+
+from lfx.cli.common import load_sdk
+
+console = Console()
+
+# ---------------------------------------------------------------------------
+# Data model
+# ---------------------------------------------------------------------------
+
+_STATUS_SYNCED = "synced"
+_STATUS_AHEAD = "ahead"
+_STATUS_BEHIND = "behind"
+_STATUS_NEW = "new"
+_STATUS_REMOTE_ONLY = "remote-only"
+_STATUS_NO_ID = "no-id"
+_STATUS_ERROR = "error"
+
+_STATUS_STYLE: dict[str, tuple[str, str, str]] = {
+    _STATUS_SYNCED: ("✓", "green", "synced"),
+    _STATUS_AHEAD: ("↑", "yellow", "ahead"),
+    _STATUS_BEHIND: ("↓", "yellow", "behind"),
+    _STATUS_NEW: ("+", "cyan", "new"),
+    _STATUS_REMOTE_ONLY: ("○", "blue", "remote only"),
+    _STATUS_NO_ID: ("?", "dim", "no id"),
+    _STATUS_ERROR: ("✗", "red", "error"),
+}
+
+
+@dataclass
+class FlowStatus:
+    name: str
+    status: str
+    path: Path | None = None
+    flow_id: UUID | None = None
+    detail: str = field(default="")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _load_sdk() -> tuple[object, object, object, type]:
+    """Return (normalize_flow, flow_to_json, Client, LangflowNotFoundError) from langflow_sdk."""
+    sdk = load_sdk("status")
+    from langflow_sdk.exceptions import LangflowNotFoundError
+    from langflow_sdk.serialization import flow_to_json, normalize_flow
+
+    return normalize_flow, flow_to_json, sdk.Client, LangflowNotFoundError
+
+
+def _flow_hash(flow_dict: dict, normalize_flow: object, flow_to_json: object) -> str:
+    """Return a short deterministic hash of the normalized flow content."""
+    normalized = normalize_flow(flow_dict)  # type: ignore[operator]
+    content = flow_to_json(normalized)  # type: ignore[operator]
+    return hashlib.sha256(content.encode()).hexdigest()[:12]
+
+
+def _collect_files(dir_path: str | None, flow_paths: list[str]) -> list[Path]:
+    """Resolve the set of local flow files to examine."""
+    files: list[Path] = []
+
+    if dir_path:
+        d = Path(dir_path)
+        if not d.is_dir():
+            msg = f"Directory not found: {d}"
+            console.print(f"[red]Error:[/red] {msg}")
+            raise typer.Exit(1)
+        files.extend(sorted(d.glob("*.json")))
+    elif flow_paths:
+        files.extend(Path(p) for p in flow_paths)
+    else:
+        # Default: flows/ in cwd
+        default = Path.cwd() / "flows"
+        if default.is_dir():
+            files.extend(sorted(default.glob("*.json")))
+
+    return files
+
+
+def _render_table(statuses: list[FlowStatus], env_label: str) -> None:
+    table = Table(
+        box=box.SIMPLE,
+        show_header=True,
+        header_style="bold",
+        title=f"Flow status vs [bold]{env_label}[/bold]",
+        title_justify="left",
+    )
+    table.add_column("Flow", min_width=24)
+    table.add_column("ID", style="dim", min_width=10)
+    table.add_column("File", style="dim")
+    table.add_column("Status", min_width=14)
+
+    for s in statuses:
+        icon, color, label = _STATUS_STYLE.get(s.status, ("?", "dim", s.status))
+        detail_str = f"  [dim]({s.detail})[/dim]" if s.detail else ""
+        id_str = str(s.flow_id)[:8] + "…" if s.flow_id else "—"
+        file_str = s.path.name if s.path else "—"
+        table.add_row(
+            s.name,
+            id_str,
+            file_str,
+            f"[{color}]{icon} {label}[/{color}]{detail_str}",
+        )
+
+    console.print(table)
+
+    # Summary line
+    counts: dict[str, int] = {}
+    for s in statuses:
+        counts[s.status] = counts.get(s.status, 0) + 1
+
+    parts = []
+    for status, (_, color, label) in _STATUS_STYLE.items():
+        if counts.get(status):
+            parts.append(f"[{color}]{counts[status]} {label}[/{color}]")
+
+    if parts:
+        console.print("  " + "  ·  ".join(parts))
+        console.print()
+
+
+# ---------------------------------------------------------------------------
+# Command
+# ---------------------------------------------------------------------------
+
+
+def status_command(
+    dir_path: str | None,
+    flow_paths: list[str],
+    env: str | None,
+    environments_file: str | None,
+    *,
+    target: str | None = None,
+    api_key: str | None = None,
+    show_remote_only: bool,
+) -> None:
+    """Compare local flow files against the remote instance and render a status table."""
+    normalize_flow, flow_to_json, client_cls, not_found_error = _load_sdk()
+
+    from lfx.config import ConfigError, resolve_environment
+
+    try:
+        env_cfg = resolve_environment(
+            env,
+            target=target,
+            api_key=api_key,
+            environments_file=environments_file,
+        )
+    except ConfigError as exc:
+        console.print(f"[red]Error:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+    try:
+        client = client_cls(base_url=env_cfg.url, api_key=env_cfg.api_key)
+    except Exception as exc:
+        console.print(f"[red]Error:[/red] Could not create client for {env_cfg.url!r}: {exc}")
+        raise typer.Exit(1) from exc
+
+    try:
+        env_label = env_cfg.name
+        local_files = _collect_files(dir_path, flow_paths)
+
+        if not local_files and not show_remote_only:
+            console.print("[yellow]No flow files found.[/yellow] Use [bold]--dir[/bold] to specify a directory.")
+            raise typer.Exit(0)
+
+        statuses: list[FlowStatus] = []
+        seen_ids: set[UUID] = set()
+
+        # ------------------------------------------------------------------ #
+        # Check each local file                                               #
+        # ------------------------------------------------------------------ #
+        for path in local_files:
+            if not path.exists():
+                statuses.append(FlowStatus(name=path.name, status=_STATUS_ERROR, path=path, detail="file not found"))
+                continue
+
+            try:
+                raw: dict = json.loads(path.read_text(encoding="utf-8"))
+            except (json.JSONDecodeError, OSError) as exc:
+                statuses.append(FlowStatus(name=path.name, status=_STATUS_ERROR, path=path, detail=str(exc)))
+                continue
+
+            name: str = raw.get("name", path.stem)
+            raw_id = raw.get("id")
+
+            if not raw_id:
+                statuses.append(
+                    FlowStatus(
+                        name=name,
+                        status=_STATUS_NO_ID,
+                        path=path,
+                        detail="run lfx export --env <env> first to assign a stable id",
+                    )
+                )
+                continue
+
+            try:
+                flow_id = UUID(str(raw_id))
+            except ValueError:
+                statuses.append(
+                    FlowStatus(name=name, status=_STATUS_ERROR, path=path, detail=f"invalid id: {raw_id!r}")
+                )
+                continue
+
+            seen_ids.add(flow_id)
+
+            try:
+                remote_flow = client.get_flow(flow_id)
+            except not_found_error:
+                statuses.append(FlowStatus(name=name, status=_STATUS_NEW, path=path, flow_id=flow_id))
+                continue
+            except Exception as exc:  # noqa: BLE001
+                statuses.append(
+                    FlowStatus(name=name, status=_STATUS_ERROR, path=path, flow_id=flow_id, detail=str(exc))
+                )
+                continue
+
+            local_hash = _flow_hash(raw, normalize_flow, flow_to_json)
+            remote_hash = _flow_hash(remote_flow.model_dump(mode="json"), normalize_flow, flow_to_json)
+
+            if local_hash == remote_hash:
+                statuses.append(FlowStatus(name=name, status=_STATUS_SYNCED, path=path, flow_id=flow_id))
+            else:
+                remote_updated_at: datetime | None = remote_flow.updated_at
+                local_mtime = datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc)
+                if remote_updated_at and remote_updated_at.tzinfo is None:
+                    remote_updated_at = remote_updated_at.replace(tzinfo=timezone.utc)
+
+                status = _STATUS_BEHIND if remote_updated_at and remote_updated_at > local_mtime else _STATUS_AHEAD
+                statuses.append(FlowStatus(name=name, status=status, path=path, flow_id=flow_id))
+
+        # ------------------------------------------------------------------ #
+        # Remote-only flows                                                   #
+        # ------------------------------------------------------------------ #
+        if show_remote_only:
+            try:
+                all_remote = client.list_flows(get_all=True)
+                statuses.extend(
+                    FlowStatus(
+                        name=remote_flow.name,
+                        status=_STATUS_REMOTE_ONLY,
+                        flow_id=remote_flow.id,
+                    )
+                    for remote_flow in all_remote
+                    if remote_flow.id not in seen_ids
+                )
+            except Exception as exc:  # noqa: BLE001
+                console.print(f"[yellow]Warning:[/yellow] Could not list remote flows: {exc}")
+    finally:
+        with contextlib.suppress(OSError):
+            client.close()
+
+    _render_table(statuses, env_label)
+
+    # Exit 1 when anything is out of sync so CI pipelines can detect drift
+    not_clean = [s for s in statuses if s.status not in (_STATUS_SYNCED,)]
+    if not_clean:
+        raise typer.Exit(1)

@@ -0,0 +1,60 @@
+"""lfx validate -- structural and semantic validation of Langflow flow JSON.
+
+Validation levels (each level implies all levels below it):
+
+    Level 1 - structural
+        The file parses as valid JSON and contains the expected top-level keys
+        (``id``, ``name``, ``data``, ``data.nodes``, ``data.edges``).
+        Also checks for orphaned nodes (no edges at all) and unused nodes
+        (not reachable from any output node), and warns about version mismatches
+        (nodes built with a different Langflow version than the one installed).
+
+    Level 2 - components
+        Every node's ``data.type`` references a component type that exists in
+        the lfx component registry.
+
+    Level 3 - edge types
+        Connected ports carry compatible types (source output type must be
+        assignable to target input type).
+
+    Level 4 - required inputs
+        Every required input field on every component has a value or an
+        incoming edge connected to it.  Also checks that password/secret fields
+        have a value or a matching environment variable set.
+
+Use ``--level`` to select how deep to go, or ``--skip-*`` flags to opt out of
+individual checks while still running the others.
+
+Pass ``--strict`` to treat warnings as errors (exit code 1).
+
+This module is a thin wrapper that re-exports symbols from the
+``lfx.cli.validation`` subpackage so that all existing imports continue
+to work unchanged.
+"""
+
+from __future__ import annotations
+
+# Re-export everything from the validation subpackage.
+# This keeps ``from lfx.cli.validate import ...`` working for all consumers
+# (the CLI entry point in __main__.py, tests, etc.).
+from lfx.cli.validation import (  # noqa: F401
+    ValidationIssue,
+    ValidationResult,
+    _check_missing_credentials,
+    _check_orphaned_nodes,
+    _check_unused_nodes,
+    _check_version_mismatch,
+    _expand_paths,
+    _get_lf_version,
+    _node_display_name,
+    validate_command,
+    validate_flow_file,
+)
+
+# Level constants (re-exported for backwards compatibility)
+from lfx.cli.validation.core import (  # noqa: F401
+    _LEVEL_COMPONENTS,
+    _LEVEL_EDGE_TYPES,
+    _LEVEL_REQUIRED_INPUTS,
+    _LEVEL_STRUCTURAL,
+)

@@ -0,0 +1,300 @@
+"""Core validation types, constants, and orchestrator for flow validation.
+
+Contains the ValidationIssue and ValidationResult dataclasses, level constants,
+the main validate_flow_file orchestrator, and CLI rendering / path helpers.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import typer
+from rich.console import Console
+
+from lfx.cli.validation.semantic import (
+    _check_component_existence,
+    _check_edge_type_compatibility,
+    _check_missing_credentials,
+    _check_required_inputs,
+)
+from lfx.cli.validation.structural import (
+    _check_orphaned_nodes,
+    _check_structural,
+    _check_unused_nodes,
+    _check_version_mismatch,
+)
+
+console = Console(stderr=True)
+ok_console = Console()
+
+# Validation level constants
+LEVEL_STRUCTURAL = 1
+LEVEL_COMPONENTS = 2
+LEVEL_EDGE_TYPES = 3
+LEVEL_REQUIRED_INPUTS = 4
+
+# Keep underscore-prefixed aliases for backwards compatibility
+_LEVEL_STRUCTURAL = LEVEL_STRUCTURAL
+_LEVEL_COMPONENTS = LEVEL_COMPONENTS
+_LEVEL_EDGE_TYPES = LEVEL_EDGE_TYPES
+_LEVEL_REQUIRED_INPUTS = LEVEL_REQUIRED_INPUTS
+
+
+# ---------------------------------------------------------------------------
+# Result types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class ValidationIssue:
+    level: int
+    severity: str  # "error" | "warning"
+    node_id: str | None
+    node_name: str | None
+    message: str
+
+
+@dataclass
+class ValidationResult:
+    path: Path
+    issues: list[ValidationIssue] = field(default_factory=list)
+
+    @property
+    def errors(self) -> list[ValidationIssue]:
+        return [i for i in self.issues if i.severity == "error"]
+
+    @property
+    def warnings(self) -> list[ValidationIssue]:
+        return [i for i in self.issues if i.severity == "warning"]
+
+    @property
+    def ok(self) -> bool:
+        return not self.errors
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _node_display_name(node: dict[str, Any]) -> str | None:
+    return node.get("data", {}).get("node", {}).get("display_name") or node.get("data", {}).get("id") or node.get("id")
+
+
+def _get_lf_version() -> str | None:
+    """Return the installed Langflow version string, or *None* if not installed.
+
+    Tries the four known package names in order of preference so the check
+    works with released builds, nightly builds, and editable installs.
+    """
+    from importlib.metadata import PackageNotFoundError, version
+
+    for pkg in ("langflow-base", "langflow", "langflow-base-nightly", "langflow-nightly"):
+        try:
+            return version(pkg)
+        except PackageNotFoundError:
+            continue
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Orchestrator
+# ---------------------------------------------------------------------------
+
+
+def validate_flow_file(
+    path: Path,
+    *,
+    level: int = LEVEL_REQUIRED_INPUTS,
+    skip_components: bool = False,
+    skip_edge_types: bool = False,
+    skip_required_inputs: bool = False,
+    skip_version_check: bool = False,
+    skip_credentials: bool = False,
+) -> ValidationResult:
+    result = ValidationResult(path=path)
+
+    try:
+        raw = path.read_text(encoding="utf-8")
+        flow: dict[str, Any] = json.loads(raw)
+    except OSError as exc:
+        result.issues.append(
+            ValidationIssue(
+                level=LEVEL_STRUCTURAL,
+                severity="error",
+                node_id=None,
+                node_name=None,
+                message=f"Cannot read file: {exc}",
+            )
+        )
+        return result
+    except json.JSONDecodeError as exc:
+        result.issues.append(
+            ValidationIssue(
+                level=LEVEL_STRUCTURAL,
+                severity="error",
+                node_id=None,
+                node_name=None,
+                message=f"Invalid JSON: {exc}",
+            )
+        )
+        return result
+
+    # Level 1 - structural (JSON shape + orphaned/unused node checks)
+    can_continue = _check_structural(flow, result)
+    if can_continue:
+        _check_orphaned_nodes(flow, result)
+        _check_unused_nodes(flow, result)
+        # Extended: version mismatch / outdated components
+        if not skip_version_check:
+            _check_version_mismatch(flow, result)
+    if not can_continue or level < LEVEL_COMPONENTS:
+        return result
+
+    # Level 2 - component existence
+    if not skip_components:
+        _check_component_existence(flow, result)
+    if level < LEVEL_EDGE_TYPES:
+        return result
+
+    # Level 3 - edge type compatibility
+    if not skip_edge_types:
+        _check_edge_type_compatibility(flow, result)
+    if level < LEVEL_REQUIRED_INPUTS:
+        return result
+
+    # Level 4 - required inputs + extended: missing credentials
+    if not skip_required_inputs:
+        _check_required_inputs(flow, result)
+    if not skip_credentials:
+        _check_missing_credentials(flow, result)
+
+    return result
+
+
+# ---------------------------------------------------------------------------
+# CLI rendering and path helpers
+# ---------------------------------------------------------------------------
+
+
+def _render_result(
+    result: ValidationResult,
+    *,
+    index: int,
+    total: int,
+    verbose: bool,
+    strict: bool = False,
+) -> None:
+    counter = f"[dim][{index}/{total}][/dim] " if total > 1 else ""
+    label = f"[bold]{result.path}[/bold]"
+    passes = result.ok and not (strict and result.warnings)
+    if passes:
+        ok_console.print(f"{counter}[green]\u2713[/green] {label}")
+    else:
+        console.print(f"{counter}[red]\u2717[/red] {label}")
+
+    show_issues = verbose or not passes
+    if show_issues:
+        for issue in result.issues:
+            effective_severity = "error" if (strict and issue.severity == "warning") else issue.severity
+            color = "red" if effective_severity == "error" else "yellow"
+            loc = f" [{issue.node_name or issue.node_id}]" if (issue.node_id or issue.node_name) else ""
+            console.print(f"  [{color}][L{issue.level} {effective_severity.upper()}][/{color}]{loc} {issue.message}")
+
+
+def _expand_paths(raw_paths: list[str]) -> list[Path]:
+    """Expand each entry to a list of .json files.
+
+    * If the path is a directory, collect every ``*.json`` file recursively.
+    * If the path is a file, return it as-is.
+    * If the path does not exist, print an error and exit 2.
+    """
+    paths: list[Path] = []
+    for raw in raw_paths:
+        p = Path(raw)
+        if not p.exists():
+            console.print(f"[red]Error:[/red] Path not found: {p}")
+            raise typer.Exit(2)
+        if p.is_dir():
+            found = sorted(p.rglob("*.json"))
+            if not found:
+                console.print(f"[yellow]Warning:[/yellow] No .json files found in {p}")
+            paths.extend(found)
+        else:
+            paths.append(p)
+    return paths
+
+
+_DEFAULT_FLOWS_DIR = "flows"
+
+
+def validate_command(
+    flow_paths: list[str],
+    level: int,
+    *,
+    skip_components: bool,
+    skip_edge_types: bool,
+    skip_required_inputs: bool,
+    skip_version_check: bool,
+    skip_credentials: bool,
+    strict: bool,
+    verbose: bool,
+    output_format: str,
+) -> None:
+    if not flow_paths:
+        flow_paths = [_DEFAULT_FLOWS_DIR]
+
+    paths = _expand_paths(flow_paths)
+
+    if not paths:
+        console.print("[yellow]No flow files to validate.[/yellow]")
+        raise typer.Exit(0)
+
+    results: list[ValidationResult] = []
+    for i, p in enumerate(paths, start=1):
+        result = validate_flow_file(
+            p,
+            level=level,
+            skip_components=skip_components,
+            skip_edge_types=skip_edge_types,
+            skip_required_inputs=skip_required_inputs,
+            skip_version_check=skip_version_check,
+            skip_credentials=skip_credentials,
+        )
+        results.append(result)
+        if output_format != "json":
+            _render_result(result, index=i, total=len(paths), verbose=verbose, strict=strict)
+
+    if output_format == "json":
+        import json as _json
+
+        out = [
+            {
+                "path": str(r.path),
+                "ok": r.ok if not strict else (not r.errors and not r.warnings),
+                "issues": [
+                    {
+                        "level": i.level,
+                        "severity": i.severity,
+                        "node_id": i.node_id,
+                        "node_name": i.node_name,
+                        "message": i.message,
+                    }
+                    for i in r.issues
+                ],
+            }
+            for r in results
+        ]
+        sys.stdout.write(_json.dumps(out, indent=2) + "\n")
+    elif len(paths) > 1:
+        passed = sum(1 for r in results if r.ok and not (strict and r.warnings))
+        failed = len(results) - passed
+        color = "green" if failed == 0 else "red"
+        ok_console.print(f"\n[{color}]Validated {len(paths)} flows: {passed} passed, {failed} failed.[/{color}]")
+
+    if any((not r.ok) or (strict and r.warnings) for r in results):
+        raise typer.Exit(1)