From 6eaae2ca4df3caac3f655c623fc76aaf0f2fe517 Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Tue, 23 Jun 2026 14:06:34 -0300 Subject: [PATCH] feat(lfx serve): v2 workflow endpoints (sync + stream) on lfx serve Gives the production runtime (lfx serve) the same v2 contract as the langflow backend's POST /api/v2/workflows, built on the shared lfx.workflow layer. - New POST /workflows endpoint: WorkflowRunRequest in, WorkflowExecutionResponse (sync) or an SSE stream (langflow/agui protocols) out. flow_id resolves against the serve registry; per-request deepcopy+stamp mirrors the run/stream endpoints. - sync runs via run_graph_internal (the same primitive the backend sync path uses) so the converter sees the aggregated RunOutputs shape. - stream drives the run with a token-stream EventManager wired into execute_graph_with_capture and feeds queue events through the shared StreamAdapter; emits a terminal end so the adapter closes the run cleanly. - background and public modes are rejected (422); tweaks/data/files/globals and partial-run boundaries are rejected too (no per-request graph rebuild yet). - execute_graph_with_capture gains an optional event_manager param. Tests build a real ChatInput->ChatOutput flow (no mocks) and exercise sync, both stream protocols, and the guards. 329 lfx serve + contract tests pass. --- .secrets.baseline | 4 +- src/lfx/src/lfx/cli/common.py | 13 +- src/lfx/src/lfx/cli/serve_app.py | 4 + src/lfx/src/lfx/cli/serve_workflow.py | 301 ++++++++++++++++++ src/lfx/tests/unit/cli/test_serve_workflow.py | 130 ++++++++ 5 files changed, 448 insertions(+), 4 deletions(-) create mode 100644 src/lfx/src/lfx/cli/serve_workflow.py create mode 100644 src/lfx/tests/unit/cli/test_serve_workflow.py diff --git a/.secrets.baseline b/.secrets.baseline index 8c0c458bf5..4aa716a215 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -8889,7 +8889,7 @@ "filename": "src/lfx/src/lfx/cli/serve_app.py", "hashed_secret": "b894b81be94cf8fa8d7536475aaec876addf05c8", "is_verified": false, - "line_number": 49, + "line_number": 51, "is_secret": false } ], @@ -9287,5 +9287,5 @@ } ] }, - "generated_at": "2026-06-10T08:36:23Z" + "generated_at": "2026-06-23T17:06:15Z" } diff --git a/src/lfx/src/lfx/cli/common.py b/src/lfx/src/lfx/cli/common.py index 9f3cb67ac9..3a00db4059 100644 --- a/src/lfx/src/lfx/cli/common.py +++ b/src/lfx/src/lfx/cli/common.py @@ -307,7 +307,7 @@ def prepare_graph(graph, verbose_print): raise typer.Exit(1) from e -async def execute_graph_with_capture(graph, input_value: str | None, session_id: str | None = None): +async def execute_graph_with_capture(graph, input_value: str | None, session_id: str | None = None, event_manager=None): """Execute a graph and capture output. Args: @@ -317,6 +317,10 @@ async def execute_graph_with_capture(graph, input_value: str | None, session_id: message-store paths (which validate session_id) succeed; an empty or whitespace-only string is rejected with ``ValueError`` to surface shell/env-var typos (see ``lfx.run._defaults.validate_provided_id``). + event_manager: Optional ``EventManager``. When provided it is threaded + into ``graph.async_start`` so components emit token/message/error + events to its queue as the run progresses (used by the streaming + workflow endpoint). ``None`` keeps the non-streaming behavior. Returns: Tuple of (results, captured_logs) @@ -349,7 +353,12 @@ async def execute_graph_with_capture(graph, input_value: str | None, session_id: try: sys.stdout = captured_stdout sys.stderr = captured_stderr - results = [result async for result in graph.async_start(inputs, fallback_to_env_vars=fallback_to_env_vars)] + results = [ + result + async for result in graph.async_start( + inputs, fallback_to_env_vars=fallback_to_env_vars, event_manager=event_manager + ) + ] except Exception as exc: # Capture any error output that was written to stderr error_output = captured_stderr.getvalue() diff --git a/src/lfx/src/lfx/cli/serve_app.py b/src/lfx/src/lfx/cli/serve_app.py index 8e7fc00cab..81cc94fb26 100644 --- a/src/lfx/src/lfx/cli/serve_app.py +++ b/src/lfx/src/lfx/cli/serve_app.py @@ -36,6 +36,7 @@ from lfx.cli.common import ( get_api_key, ) from lfx.cli.runtime_variables import apply_global_vars_to_graph +from lfx.cli.serve_workflow import add_v2_workflow_routes from lfx.load import load_flow_from_json from lfx.log.logger import logger from lfx.utils.flow_validation import validate_flow_for_current_settings @@ -781,6 +782,9 @@ def create_multi_serve_app( return StreamingResponse(error_stream(), media_type="text/event-stream") + # V2 workflow contract endpoints (sync + stream), shared with the langflow backend. + add_v2_workflow_routes(app, registry, api_key_dependency=verify_api_key) + return app diff --git a/src/lfx/src/lfx/cli/serve_workflow.py b/src/lfx/src/lfx/cli/serve_workflow.py new file mode 100644 index 0000000000..3beeb99a85 --- /dev/null +++ b/src/lfx/src/lfx/cli/serve_workflow.py @@ -0,0 +1,301 @@ +"""V2 workflow-shaped endpoints for ``lfx serve``. + +Gives the standalone lfx runtime the same request/response contract as the +langflow backend ``POST /api/v2/workflows`` for the ``sync`` and ``stream`` +execution modes, so a client integrates against one contract regardless of which +runtime serves it. + +Background and public modes stay backend-only: they need a database, job queue, +and auth model that stateless ``lfx serve`` does not have. + +Built on the shared contract layer in ``lfx.workflow`` (adapters + converters) +and the native ``lfx`` ``Graph.async_start`` event stream. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import time +from copy import deepcopy +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any +from uuid import uuid4 + +from fastapi import Depends, HTTPException, status +from fastapi.responses import StreamingResponse + +from lfx.cli.common import execute_graph_with_capture +from lfx.events.event_manager import create_stream_tokens_event_manager +from lfx.processing.process import run_graph_internal +from lfx.schema.schema import InputValueRequest + +# WorkflowRunRequest stays a runtime import: FastAPI resolves the route's body +# annotation at request-model build time, so it cannot live under TYPE_CHECKING. +from lfx.schema.workflow import WorkflowRunRequest # noqa: TC001 +from lfx.utils.flow_validation import validate_flow_for_current_settings +from lfx.workflow.adapters import ( + STREAM_ADAPTERS, + StreamAdapterContext, + available_protocols, + get_stream_adapter, +) +from lfx.workflow.converters import ( + create_error_response, + parse_workflow_run_request, + run_response_to_workflow_response, +) + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from fastapi import FastAPI + + from lfx.graph.graph.base import Graph + from lfx.schema.workflow import WorkflowExecutionResponse + from lfx.workflow.adapters import StreamAdapter + from lfx.workflow.converters import ParsedWorkflowRun + + +# Bounded queue between the graph run and the SSE consumer, mirroring the backend +# stream loop: a slow client applies backpressure instead of letting frames +# accumulate without bound. +_STREAM_QUEUE_MAX_SIZE = 256 + + +@dataclass +class _RunResponse: + """Minimal ``RunResponseLike``: the two attributes the converter reads.""" + + outputs: list[Any] | None + session_id: str | None + + +def _reject_unsupported_fields(parsed: ParsedWorkflowRun) -> None: + """Reject request fields lfx serve does not execute yet. + + lfx serve runs a pre-registered, prepared graph and has no per-request graph + rebuild, so live ``data`` overrides, ``tweaks``, ``files``, partial-run + boundaries, and request-level ``globals`` are not supported here yet (they + remain available on the langflow backend). Reject explicitly rather than + silently ignore so a caller is never surprised. + """ + unsupported: list[str] = [] + if parsed.tweaks: + unsupported.append("tweaks") + if parsed.data is not None: + unsupported.append("data") + if parsed.files: + unsupported.append("files") + if parsed.globals: + unsupported.append("globals") + if parsed.start_component_id is not None: + unsupported.append("start_component_id") + if parsed.stop_component_id is not None: + unsupported.append("stop_component_id") + if unsupported: + fields = ", ".join(unsupported) + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail={ + "error": "Unsupported request fields", + "code": "LFX_SERVE_UNSUPPORTED_FIELDS", + "message": f"lfx serve does not support these v2 fields yet: {fields}. Use the langflow " + "backend for live-data overrides, tweaks, files, partial-run boundaries, or " + "request-level globals.", + "fields": unsupported, + }, + ) + + +def _build_inputs(parsed: ParsedWorkflowRun) -> list[InputValueRequest] | None: + """Build the single chat input for the run, scoped to the session, if any.""" + if not parsed.input_value: + return None + return [InputValueRequest(components=[], input_value=parsed.input_value, type="chat", session=parsed.session_id)] + + +def _terminal_node_ids(graph: Graph) -> list[str]: + """The flow's output (sink) vertices, computed the way the converter does.""" + return [vertex.id for vertex in graph.vertices if not graph.successor_map.get(vertex.id, [])] + + +async def run_workflow_sync(graph: Graph, parsed: ParsedWorkflowRun, flow_id: str) -> WorkflowExecutionResponse: + """Run a flow to completion and build a v2 ``WorkflowExecutionResponse``. + + Uses ``run_graph_internal`` (the same primitive the langflow backend sync + path uses) so the result is the aggregated ``RunOutputs`` shape the shared + converter expects. Component-level failures are returned in the body + (HTTP 200) to match the v2 two-tier contract. + """ + job_id = str(uuid4()) + try: + run_outputs, session_id = await run_graph_internal( + graph, + flow_id, + stream=False, + session_id=parsed.session_id, + inputs=_build_inputs(parsed), + outputs=_terminal_node_ids(graph), + ) + except Exception as exc: # noqa: BLE001 + return create_error_response( + flow_id=flow_id, + job_id=job_id, + inputs=parsed.tweaks, + error=exc, + effective_globals={}, + ) + run_response = _RunResponse(outputs=run_outputs, session_id=session_id) + return run_response_to_workflow_response( + run_response=run_response, + flow_id=flow_id, + job_id=job_id, + inputs=parsed.tweaks, + graph=graph, + effective_globals={}, + selected_ids=parsed.output_ids, + ) + + +def _format_sse(data_json: str, seq: int) -> bytes: + """Frame one event as an SSE message with a monotonic id for ``Last-Event-ID``. + + lfx has no ``format_sse_event`` helper, so frame manually to the same wire + shape the backend emits (``id:`` + ``data:`` lines). + """ + return f"id: {seq}\ndata: {data_json}\n\n".encode() + + +async def stream_workflow_frames( + graph: Graph, parsed: ParsedWorkflowRun, adapter: StreamAdapter +) -> AsyncIterator[bytes]: + """Run a flow and stream its events through ``adapter`` as SSE frames. + + The graph runs via ``execute_graph_with_capture`` with a token-stream + ``EventManager`` wired in, so component token/message/error events land on a + queue while this consumer translates them through the adapter. A failure + becomes the adapter's terminal-error event rather than an HTTP error. + """ + queue: asyncio.Queue = asyncio.Queue(maxsize=_STREAM_QUEUE_MAX_SIZE) + event_manager = create_stream_tokens_event_manager(queue=queue) + drive_error: BaseException | None = None + + async def drive() -> None: + nonlocal drive_error + try: + await execute_graph_with_capture( + graph, parsed.input_value or None, session_id=parsed.session_id, event_manager=event_manager + ) + # lfx's async_start does not emit a terminal ``end`` event (the + # langflow build loop does). Emit one so the adapter closes the run + # cleanly: the agui adapter rides RUN_FINISHED on translating ``end``, + # and the langflow adapter emits its final ``end`` frame. + event_manager.on_end(data={}) + except asyncio.CancelledError: + raise + except Exception as exc: # noqa: BLE001 + drive_error = exc + finally: + await queue.put((None, None, time.time())) + + seq = 0 + run_task = asyncio.create_task(drive()) + try: + for event in adapter.initial_events(): + yield _format_sse(event.data_json, seq) + seq += 1 + while True: + _event_id, value, _put_time = await queue.get() + if value is None: + break + payload = json.loads(value.decode("utf-8")) + for event in adapter.translate(payload.get("event", ""), payload.get("data") or {}): + yield _format_sse(event.data_json, seq) + seq += 1 + for event in adapter.final_events(): + yield _format_sse(event.data_json, seq) + seq += 1 + # Guaranteed terminal-error fallback: if the run raised before a + # cooperative error reached the queue, emit the adapter's error event(s). + if drive_error is not None: + for event in adapter.error_events(drive_error): + yield _format_sse(event.data_json, seq) + seq += 1 + finally: + if not run_task.done(): + run_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await run_task + + +def add_v2_workflow_routes(app: FastAPI, registry, *, api_key_dependency) -> None: + """Register ``POST /workflows`` (v2 sync + stream) on the serve app. + + Mirrors the langflow backend ``POST /api/v2/workflows`` contract: same + ``WorkflowRunRequest`` body (``flow_id`` included) and + ``WorkflowExecutionResponse`` / SSE responses, so a client integrates + identically against lfx serve. + """ + + @app.post( + "/workflows", + response_model=None, + tags=["workflow"], + summary="Execute Workflow (v2 sync or stream)", + dependencies=[Depends(api_key_dependency)], + ) + async def execute_workflow(request: WorkflowRunRequest): + result = registry.get(request.flow_id) + if result is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"error": "flow not found", "code": "FLOW_NOT_FOUND", "flow_id": request.flow_id}, + ) + graph, _meta = result + + if request.stream_protocol not in STREAM_ADAPTERS: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail={ + "error": "Unknown stream_protocol", + "code": "UNKNOWN_STREAM_PROTOCOL", + "message": f"Unknown stream_protocol {request.stream_protocol!r}.", + "available": available_protocols(), + }, + ) + + parsed = parse_workflow_run_request(request) + _reject_unsupported_fields(parsed) + + if parsed.mode == "background": + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail={ + "error": "Unsupported mode", + "code": "LFX_SERVE_UNSUPPORTED_MODE", + "message": "lfx serve supports mode 'sync' and 'stream'. Background runs need the " + "langflow backend (durable jobs + queue).", + }, + ) + + # Per-request isolation: never mutate the shared cached graph. Mirrors the + # run/stream endpoints. deepcopy drops graph.context, so re-stamp the + # registry's env policy. + validate_flow_for_current_settings(graph) + graph_copy = deepcopy(graph) + registry.stamp(graph_copy) + + if parsed.mode == "stream": + adapter = get_stream_adapter( + request.stream_protocol, + StreamAdapterContext(run_id=str(uuid4()), thread_id=parsed.session_id or request.flow_id), + ) + return StreamingResponse( + stream_workflow_frames(graph_copy, parsed, adapter), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + + return await run_workflow_sync(graph_copy, parsed, request.flow_id) diff --git a/src/lfx/tests/unit/cli/test_serve_workflow.py b/src/lfx/tests/unit/cli/test_serve_workflow.py new file mode 100644 index 0000000000..77e564d690 --- /dev/null +++ b/src/lfx/tests/unit/cli/test_serve_workflow.py @@ -0,0 +1,130 @@ +"""Tests for the v2 workflow endpoints on ``lfx serve`` (POST /workflows). + +Real graph, no mocks: a ChatInput -> ChatOutput echo flow is registered and +exercised through the FastAPI app, asserting the v2 ``WorkflowExecutionResponse`` +(sync) and the AG-UI / langflow SSE streams, plus the guard responses. +""" + +import pytest +from fastapi.testclient import TestClient +from lfx.cli.serve_app import FlowMeta, FlowRegistry, create_multi_serve_app +from lfx.components.input_output import ChatInput, ChatOutput +from lfx.graph import Graph + +# WorkflowRunRequest requires flow_id to be a UUID (the v2 contract). +_FLOW_ID = "67ccd2be-17f0-4190-81ff-3bb2cf6508e6" +_MISSING_FLOW_ID = "00000000-0000-4000-8000-000000000000" +_API_KEY = "test-key" # pragma: allowlist secret +_HEADERS = {"x-api-key": _API_KEY} + + +def _echo_graph() -> Graph: + """A minimal real flow: ChatInput feeds its message straight to ChatOutput.""" + chat_input = ChatInput(_id="chat_input") + chat_output = ChatOutput(_id="chat_output") + chat_output.set(input_value=chat_input.message_response) + graph = Graph(chat_input, chat_output) + graph.prepare() + return graph + + +@pytest.fixture +def client(monkeypatch): + monkeypatch.setenv("LANGFLOW_API_KEY", _API_KEY) + registry = FlowRegistry() + registry.add( + _echo_graph(), + FlowMeta(id=_FLOW_ID, relative_path=f"{_FLOW_ID}.json", title="echo", description=None), + ) + app = create_multi_serve_app(registry=registry) + with TestClient(app) as test_client: + yield test_client + + +def test_sync_returns_workflow_execution_response(client): + resp = client.post( + "/workflows", + json={"flow_id": _FLOW_ID, "input_value": "hello", "mode": "sync"}, + headers=_HEADERS, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["flow_id"] == _FLOW_ID + assert body["output"]["reason"] == "single" + assert "hello" in (body["output"]["text"] or "") + # The full per-component map is present alongside the primary answer. + assert body["outputs"] + + +def test_stream_agui_emits_run_lifecycle_and_content(client): + resp = client.post( + "/workflows", + json={"flow_id": _FLOW_ID, "input_value": "hello", "mode": "stream", "stream_protocol": "agui"}, + headers=_HEADERS, + ) + assert resp.status_code == 200, resp.text + assert resp.headers["content-type"].startswith("text/event-stream") + body = resp.text + assert "RUN_STARTED" in body + assert "RUN_FINISHED" in body + assert "hello" in body + + +def test_stream_langflow_protocol_passes_through_frames(client): + resp = client.post( + "/workflows", + json={"flow_id": _FLOW_ID, "input_value": "hello", "mode": "stream", "stream_protocol": "langflow"}, + headers=_HEADERS, + ) + assert resp.status_code == 200, resp.text + assert resp.headers["content-type"].startswith("text/event-stream") + body = resp.text + assert "data:" in body + assert "hello" in body + + +def test_unknown_stream_protocol_422(client): + resp = client.post( + "/workflows", + json={"flow_id": _FLOW_ID, "mode": "stream", "stream_protocol": "bogus"}, + headers=_HEADERS, + ) + assert resp.status_code == 422 + assert resp.json()["detail"]["code"] == "UNKNOWN_STREAM_PROTOCOL" + + +def test_unsupported_fields_rejected_422(client): + resp = client.post( + "/workflows", + json={"flow_id": _FLOW_ID, "input_value": "x", "tweaks": {"ChatOutput-x": {"foo": 1}}}, + headers=_HEADERS, + ) + assert resp.status_code == 422 + detail = resp.json()["detail"] + assert detail["code"] == "LFX_SERVE_UNSUPPORTED_FIELDS" + assert "tweaks" in detail["fields"] + + +def test_background_mode_rejected_422(client): + resp = client.post( + "/workflows", + json={"flow_id": _FLOW_ID, "input_value": "x", "mode": "background"}, + headers=_HEADERS, + ) + assert resp.status_code == 422 + assert resp.json()["detail"]["code"] == "LFX_SERVE_UNSUPPORTED_MODE" + + +def test_unknown_flow_404(client): + resp = client.post( + "/workflows", + json={"flow_id": _MISSING_FLOW_ID, "input_value": "x"}, + headers=_HEADERS, + ) + assert resp.status_code == 404 + assert resp.json()["detail"]["code"] == "FLOW_NOT_FOUND" + + +def test_missing_api_key_401(client): + resp = client.post("/workflows", json={"flow_id": _FLOW_ID, "input_value": "x"}) + assert resp.status_code == 401