pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-24 08:23:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: Implement synchronous workflow execution API (#11255)

Browse Source 
* feat: Create controller shell and schema model for new workflow API

- Add workflow API endpoints (POST /workflow, GET /workflow, POST /workflow/stop)
- Implement developer API protection with settings check
- Add comprehensive workflow schema models with proper validation
- Create extensive unit test suite covering all scenarios
- Apply Ruff linting standards and fix all code quality issues
- Support API key authentication for all workflow endpoints

* fix: Move developer API check to router-level dependency

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* fix: Remove response model as its automatically configured.

Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>

* [autofix.ci] apply automated fixes

* feat: Implement synchronous workflow execution API

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* Apply suggestion from @codeflash-ai[bot]

Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com>

* Apply suggestion from @codeflash-ai[bot]

Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com>

* Apply suggestion from @codeflash-ai[bot]

Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com>

* [autofix.ci] apply automated fixes

* Clean up V2 Workflow API code and add unit tests

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* fix: CodeRabbit review fixes

* fix: improve V2 workflow API test quality and code structure

* Fix ruff whitespace issues in v2 API test files

* [autofix.ci] apply automated fixes

* Fix mypy type errors in v2 API files

* [autofix.ci] apply automated fixes

* refactor: improve data extraction with flexible path combinations and prevent flow data mutation

* fix: resolve component index merge conflict after sync

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* refactor: move ComponentOutput construction to avoid dict change after construction

* [autofix.ci] apply automated fixes

---------

Co-authored-by: Janardan S Kavia <janardanskavia@Janardans-MacBook-Pro.local>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Gabriel Luiz Freitas Almeida <gabriel@langflow.org>
Co-authored-by: Janardan S Kavia <janardanskavia@mac.war.can.ibm.com>
Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com>
This commit is contained in:
Janardan Singh Kavia
2026-01-21 16:34:10 -05:00
committed by GitHub
parent 60e3ab4970
commit fa59cc50e4
8 changed files with 3076 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/backend/base/langflow/api/v1/endpoints.py
View File

@ -488,7 +488,7 @@ async def _run_flow_internal(
async def simplified_run_flow(
*,
background_tasks: BackgroundTasks,
flow: Annotated[FlowRead | None, Depends(get_flow_by_id_or_endpoint_name)],
flow: Annotated[FlowRead, Depends(get_flow_by_id_or_endpoint_name)],
input_request: SimplifiedAPIRequest | None = None,
stream: bool = False,
api_key_user: Annotated[UserRead, Depends(api_key_security)],
@ -546,7 +546,7 @@ async def simplified_run_flow(
async def simplified_run_flow_session(
*,
background_tasks: BackgroundTasks,
flow: Annotated[FlowRead | None, Depends(get_flow_by_id_or_endpoint_name)],
flow: Annotated[FlowRead, Depends(get_flow_by_id_or_endpoint_name)],
input_request: SimplifiedAPIRequest | None = None,
stream: bool = False,
api_key_user: CurrentActiveUser,

538
src/backend/base/langflow/api/v2/converters.py Normal file
View File

@ -0,0 +1,538 @@
"""Schema converters for V2 Workflow API.
This module provides conversion functions between the new V2 workflow schemas
and the existing V1 schemas, enabling reuse of existing execution logic while
presenting a new API interface.
Key Functions:
- parse_flat_inputs: Converts flat input format to tweaks structure
- run_response_to_workflow_response: Converts V1 RunResponse to V2 WorkflowExecutionResponse
- create_error_response: Creates standardized error responses
- create_job_response: Creates background job responses
Internal Helpers:
- _extract_nested_value: Safely extracts nested values from dict/object structures
- _extract_text_from_message: Extracts plain text from various message formats
- _simplify_output_content: Simplifies output content based on type
- _get_raw_content: Extracts raw content from vertex output data
- _extract_model_source: Extracts model information from LLM outputs
- _extract_file_path: Extracts file path from SaveToFile outputs
- _build_metadata_for_non_output: Builds metadata for non-output terminal nodes
"""
from __future__ import annotations
import time
from typing import TYPE_CHECKING, Any
from lfx.schema.workflow import (
ComponentOutput,
ErrorDetail,
JobStatus,
WorkflowExecutionRequest,
WorkflowExecutionResponse,
WorkflowJobResponse,
)
if TYPE_CHECKING:
from lfx.graph.graph.base import Graph
from langflow.api.v1.schemas import RunResponse
def parse_flat_inputs(inputs: dict[str, Any]) -> tuple[dict[str, dict[str, Any]], str | None]:
"""Parse flat inputs structure into tweaks and session_id.
Format: {"component_id.param": value}
Example: {"ChatInput-abc.input_value": "hi", "LLM-xyz.temperature": 0.7}
All parameters (including input_value) are treated as tweaks.
The graph's topological sort handles execution order automatically.
Args:
inputs: The inputs dictionary from WorkflowExecutionRequest
Returns:
Tuple of (tweaks_dict, session_id)
- tweaks_dict: {component_id: {param: value}}
- session_id: Session ID if provided
Example:
>>> inputs = {
... "ChatInput-abc.input_value": "hello",
... "ChatInput-abc.session_id": "session-123",
... "LLM-xyz.temperature": 0.7
... }
>>> tweaks, session_id = parse_flat_inputs(inputs)
>>> tweaks
{'ChatInput-abc': {'input_value': 'hello'}, 'LLM-xyz': {'temperature': 0.7}}
>>> session_id
'session-123'
"""
tweaks: dict[str, dict[str, Any]] = {}
session_id: str | None = None
# Group inputs by component_id
component_inputs: dict[str, dict[str, Any]] = {}
for key, value in inputs.items():
if "." in key:
# Split component_id.param
component_id, param_name = key.split(".", 1)
if component_id not in component_inputs:
component_inputs[component_id] = {}
component_inputs[component_id][param_name] = value
# No dot - treat as component-level dict (for backward compatibility)
elif isinstance(value, dict):
tweaks[key] = value
# Process component inputs
for component_id, params in component_inputs.items():
# Extract session_id if present (use first one found)
if "session_id" in params and session_id is None:
session_id = params["session_id"]
# Build tweaks for all parameters except session_id
tweak_params = {k: v for k, v in params.items() if k != "session_id"}
if tweak_params:
tweaks[component_id] = tweak_params
return tweaks, session_id
def _extract_nested_value(data: Any, *keys: str) -> Any:
"""Safely extract nested value from dict-like structure.
Args:
data: The data structure to extract from
*keys: Sequence of keys to traverse
Returns:
The extracted value or None if not found
Example:
>>> _extract_nested_value({'a': {'b': 'value'}}, 'a', 'b')
'value'
"""
current = data
for key in keys:
if isinstance(current, dict):
current = current.get(key)
elif hasattr(current, key):
current = getattr(current, key)
else:
return None
if current is None:
return None
return current
def _extract_text_from_message(content: dict) -> str | None:
"""Extract plain text from nested message structures.
Handles various message formats by trying common paths in order:
- {'message': {'message': 'text', 'type': 'text'}}
- {'text': {'message': 'text'}}
- {'message': {'text': 'text'}}
- {'message': 'text'}
- {'text': {'text': 'text'}}
- {'text': 'text'}
Args:
content: The message content dict
Returns:
Extracted text string or None
"""
paths = [
("message", "message"),
("text", "message"),
("message", "text"),
("message",),
("text", "text"),
("text",),
]
for path in paths:
text = _extract_nested_value(content, *path)
if isinstance(text, str):
return text
return None
def _extract_model_source(raw_content: dict, vertex_id: str, vertex_display_name: str) -> dict | None:
"""Extract model source information from LLM component output.
Args:
raw_content: The raw output data
vertex_id: Vertex ID
vertex_display_name: Vertex display name
Returns:
Source info dict or None
"""
model_name = _extract_nested_value(raw_content, "model_output", "message", "model_name")
if model_name:
return {"id": vertex_id, "display_name": vertex_display_name, "source": model_name}
return None
def _extract_file_path(raw_content: dict, vertex_type: str) -> str | None:
"""Extract file path from SaveToFile component output.
Args:
raw_content: The raw output data
vertex_type: The vertex type
Returns:
File path string or None
"""
if vertex_type != "SaveToFile":
return None
# Extract the message from SaveToFile component
# Return the whole message instead of filtering by specific wording
file_msg = _extract_nested_value(raw_content, "message", "message")
if isinstance(file_msg, str):
return file_msg
return None
def _get_raw_content(vertex_output_data: Any) -> Any:
"""Extract raw content from vertex output data.
Tries multiple fields in order: outputs, results, messages.
Note: Uses 'is not None' checks to avoid treating empty collections as missing.
Args:
vertex_output_data: The output data from RunResponse
Returns:
Raw content or None
"""
if hasattr(vertex_output_data, "outputs") and vertex_output_data.outputs is not None:
return vertex_output_data.outputs
if hasattr(vertex_output_data, "results") and vertex_output_data.results is not None:
return vertex_output_data.results
if hasattr(vertex_output_data, "messages") and vertex_output_data.messages is not None:
return vertex_output_data.messages
if isinstance(vertex_output_data, dict):
# Check for 'results' first, then 'content' if results is None
if "results" in vertex_output_data:
return vertex_output_data["results"]
if "content" in vertex_output_data:
return vertex_output_data["content"]
return vertex_output_data
def _simplify_output_content(content: Any, output_type: str) -> Any:
"""Simplify output content for output nodes.
For message types, extracts plain text from nested structures.
For data/dataframe types, extracts the actual data value.
For other types, returns content as-is.
Args:
content: The raw content
output_type: The output type
Returns:
Simplified content
"""
if not isinstance(content, dict):
return content
if output_type in {"message", "text"}:
text = _extract_text_from_message(content)
return text if text is not None else content
if output_type == "data":
# For data types, try multiple path combinations in order
# This allows flexibility for different component output structures
data_paths = [
("result", "message"), # Standard: {'result': {'message': {...}}}
("results", "message"), # Plural variant: {'results': {'message': {...}}}
]
for path in data_paths:
result_data = _extract_nested_value(content, *path)
if result_data is not None:
return result_data
# TODO: Future scope - Add dataframe-specific extraction logic
# The following code is commented out pending further requirements analysis:
if output_type == "dataframe":
# For dataframe types, try multiple path combinations in order
dataframe_paths = [
("results", "message"), # Plural: {'results': {'message': {...}}}
("result", "message"), # Singular fallback: {'result': {'message': {...}}}
("run_sql_query", "message"), # SQL component specific
]
for path in dataframe_paths:
dataframe_data = _extract_nested_value(content, *path)
if dataframe_data is not None:
return dataframe_data
return content
def _build_metadata_for_non_output(
raw_content: Any, vertex_id: str, vertex_display_name: str, vertex_type: str, output_type: str
) -> dict[str, Any]:
"""Build metadata for non-output terminal nodes.
Extracts:
- source: Model information for LLM components
- file_path: File path for SaveToFile components
Args:
raw_content: The raw output data
vertex_id: Vertex ID
vertex_display_name: Vertex display name
vertex_type: Vertex type
output_type: Output type
Returns:
Metadata dict
"""
metadata: dict[str, Any] = {}
if output_type != "message" or not isinstance(raw_content, dict):
return metadata
# Extract model source for LLM components
source_info = _extract_model_source(raw_content, vertex_id, vertex_display_name)
if source_info:
metadata["source"] = source_info
# Extract file path for SaveToFile components
file_path = _extract_file_path(raw_content, vertex_type)
if file_path:
metadata["file_path"] = file_path
return metadata
def _process_terminal_vertex(
vertex: Any,
output_data_map: dict[str, Any],
display_name_counts: dict[str, int],
) -> tuple[str, ComponentOutput]:
"""Process a single terminal vertex and return (output_key, component_output).
Args:
vertex: The vertex to process
output_data_map: Map of component_id to output data
display_name_counts: Count of each display_name for duplicate detection
Returns:
Tuple of (output_key, ComponentOutput)
"""
# Get output data by vertex.id (component_id)
vertex_output_data = output_data_map.get(vertex.id)
# Determine output type from vertex
output_type = "unknown"
if vertex.outputs and len(vertex.outputs) > 0:
types = vertex.outputs[0].get("types", [])
if types:
output_type = types[0].lower()
if output_type == "unknown" and vertex.vertex_type:
output_type = vertex.vertex_type.lower()
# Initialize metadata with component_type
metadata: dict[str, Any] = {"component_type": vertex.vertex_type}
# Extract content
content = None
if vertex_output_data:
raw_content = _get_raw_content(vertex_output_data)
if vertex.is_output and raw_content is not None:
# Output nodes: simplify content
content = _simplify_output_content(raw_content, output_type)
elif not vertex.is_output and raw_content is not None:
# Non-output nodes:
# - For data types: extract and show content
# - For message types: extract metadata only (source, file_path)
# TODO: Future scope - Add support for "dataframe" output type
if output_type in ["data", "dataframe"]:
# Show data content for non-output data nodes
content = _simplify_output_content(raw_content, output_type)
else:
# For message types, extract metadata only
extra_metadata = _build_metadata_for_non_output(
raw_content,
vertex.id,
vertex.display_name or vertex.vertex_type,
vertex.vertex_type,
output_type,
)
metadata.update(extra_metadata)
# Add any additional metadata from result data
if hasattr(vertex_output_data, "metadata") and vertex_output_data.metadata:
metadata.update(vertex_output_data.metadata)
elif isinstance(vertex_output_data, dict) and "metadata" in vertex_output_data:
result_metadata = vertex_output_data.get("metadata")
if isinstance(result_metadata, dict):
metadata.update(result_metadata)
# Determine output key: use display_name if unique, otherwise use id
display_name = vertex.display_name or vertex.id
if display_name_counts.get(display_name, 0) > 1:
# Duplicate display_name detected, use id instead
output_key = vertex.id
# Store the display_name in metadata for reference
if vertex.display_name and vertex.display_name != vertex.id:
metadata["display_name"] = vertex.display_name
else:
# Unique display_name, use it as key
output_key = display_name
# Build ComponentOutput
component_output = ComponentOutput(
type=output_type,
component_id=vertex.id,
status=JobStatus.COMPLETED,
content=content,
metadata=metadata,
)
return output_key, component_output
def run_response_to_workflow_response(
run_response: RunResponse,
flow_id: str,
job_id: str,
workflow_request: WorkflowExecutionRequest,
graph: Graph,
) -> WorkflowExecutionResponse:
"""Convert V1 RunResponse to V2 WorkflowExecutionResponse.
This function transforms the V1 execution response to the new V2 schema format.
It intelligently handles different node types and determines what content to expose.
Terminal Node Processing Logic:
1. Identifies all terminal nodes (vertices with no successors)
2. For each terminal node:
- Output nodes (is_output=True): Full content is exposed
- Data/DataFrame nodes: Content is exposed regardless of is_output flag
- Message nodes (non-output): Only metadata is exposed (source, file_path)
Output Key Selection:
- Uses vertex.display_name as the primary key for outputs
- Falls back to vertex.id if duplicate display_names are detected
- Stores original display_name in metadata when using id as key
Args:
run_response: The V1 response from simple_run_flow containing execution results
flow_id: The flow identifier
job_id: The generated job ID for tracking this execution
workflow_request: Original workflow request (inputs are echoed back in response)
graph: The Graph instance used for terminal node detection and vertex metadata
Returns:
WorkflowExecutionResponse: V2 schema response with structured outputs
Example:
Terminal nodes: ["ChatOutput-abc", "LLM-xyz", "DataNode-123"]
- ChatOutput-abc (is_output=True, type=message): Full content exposed
- LLM-xyz (is_output=False, type=message): Only metadata (model source)
- DataNode-123 (is_output=False, type=data): Full content exposed
"""
# Get terminal nodes (vertices with no successors)
try:
terminal_node_ids = graph.get_terminal_nodes()
except AttributeError:
# Fallback: manually check successor_map
terminal_node_ids = [vertex.id for vertex in graph.vertices if not graph.successor_map.get(vertex.id, [])]
# Build output data map from run_response using component_id as key
# This ensures unique keys even when components have duplicate display_names
output_data_map: dict[str, Any] = {}
if run_response.outputs:
for run_output in run_response.outputs:
if hasattr(run_output, "outputs") and run_output.outputs:
for result_data in run_output.outputs:
if not result_data:
continue
# Use component_id as key to ensure uniqueness
component_id = result_data.component_id if hasattr(result_data, "component_id") else None
if component_id:
output_data_map[component_id] = result_data
# First pass: collect all terminal vertices and check for duplicate display_names
terminal_vertices = [graph.get_vertex(vertex_id) for vertex_id in terminal_node_ids]
display_name_counts: dict[str, int] = {}
for vertex in terminal_vertices:
display_name = vertex.display_name or vertex.id
display_name_counts[display_name] = display_name_counts.get(display_name, 0) + 1
# Process each terminal vertex
outputs: dict[str, ComponentOutput] = {}
for vertex in terminal_vertices:
output_key, component_output = _process_terminal_vertex(vertex, output_data_map, display_name_counts)
outputs[output_key] = component_output
return WorkflowExecutionResponse(
flow_id=flow_id,
job_id=job_id,
object="response",
created_timestamp=str(int(time.time())),
status=JobStatus.COMPLETED,
errors=[],
inputs=workflow_request.inputs or {},
outputs=outputs,
metadata={},
)
def create_job_response(job_id: str) -> WorkflowJobResponse:
"""Create a background job response.
Args:
job_id: The generated job ID
Returns:
WorkflowJobResponse for background execution
"""
return WorkflowJobResponse(
job_id=job_id,
created_timestamp=str(int(time.time())),
status=JobStatus.QUEUED,
errors=[],
)
def create_error_response(
flow_id: str,
job_id: str,
workflow_request: WorkflowExecutionRequest,
error: Exception,
) -> WorkflowExecutionResponse:
"""Create an error response in workflow format.
Args:
flow_id: The flow ID
job_id: The job ID
workflow_request: Original request
error: The exception that occurred
Returns:
WorkflowExecutionResponse with error details
"""
error_detail = ErrorDetail(
error=str(error), code="EXECUTION_ERROR", details={"flow_id": flow_id, "error_type": type(error).__name__}
)
return WorkflowExecutionResponse(
flow_id=flow_id,
job_id=job_id,
object="response",
created_timestamp=str(int(time.time())),
status=JobStatus.FAILED,
errors=[error_detail],
inputs=workflow_request.inputs or {},
outputs={},
metadata={},
)

423
src/backend/base/langflow/api/v2/workflow.py
View File

@ -1,11 +1,34 @@
"""V2 Workflow execution endpoints."""
"""V2 Workflow execution endpoints.
This module implements the V2 Workflow API endpoints for executing flows with
enhanced error handling, timeout protection, and structured responses.
Endpoints:
POST /workflow: Execute a workflow (sync, stream, or background modes)
GET /workflow: Get workflow job status by job_id
POST /workflow/stop: Stop a running workflow execution
Features:
- Developer API protection (requires developer_api_enabled setting)
- Comprehensive error handling with structured error responses
- Timeout protection for long-running executions
- Support for multiple execution modes (sync, stream, background)
- API key authentication required for all endpoints
Configuration:
EXECUTION_TIMEOUT: Maximum execution time for synchronous workflows (300 seconds)
"""
from __future__ import annotations
import asyncio
from copy import deepcopy
from typing import Annotated
from uuid import uuid4
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, status
from fastapi.responses import StreamingResponse
from lfx.graph.graph.base import Graph
from lfx.schema.workflow import (
WORKFLOW_EXECUTION_RESPONSES,
WORKFLOW_STATUS_RESPONSES,
@ -16,19 +39,48 @@ from lfx.schema.workflow import (
WorkflowStopResponse,
)
from lfx.services.deps import get_settings_service
from pydantic_core import ValidationError as PydanticValidationError
from sqlalchemy.exc import OperationalError
from langflow.api.utils import extract_global_variables_from_headers
from langflow.api.v1.schemas import RunResponse
from langflow.api.v2.converters import (
create_error_response,
parse_flat_inputs,
run_response_to_workflow_response,
)
from langflow.exceptions.api import WorkflowTimeoutError, WorkflowValidationError
from langflow.helpers.flow import get_flow_by_id_or_endpoint_name
from langflow.processing.process import process_tweaks, run_graph_internal
from langflow.services.auth.utils import api_key_security
from langflow.services.database.models.flow.model import FlowRead
from langflow.services.database.models.user.model import UserRead
# Configuration constants
EXECUTION_TIMEOUT = 300 # 5 minutes default timeout for sync execution
def check_developer_api_enabled() -> None:
"""Check if developer API is enabled, raise HTTPException if not."""
"""Check if developer API is enabled.
This dependency function protects all workflow endpoints by verifying that
the developer API feature is enabled in the application settings.
Raises:
HTTPException: 403 Forbidden if developer_api_enabled setting is False
Note:
This is used as a router-level dependency to protect all workflow endpoints.
"""
settings = get_settings_service().settings
if not settings.developer_api_enabled:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="This endpoint is not available",
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": "Developer API disabled",
"code": "DEVELOPER_API_DISABLED",
"message": "Developer API is not enabled. Contact administrator to enable this feature.",
},
)
@ -45,21 +97,301 @@ router = APIRouter(prefix="/workflow", tags=["Workflow"], dependencies=[Depends(
)
async def execute_workflow(
workflow_request: WorkflowExecutionRequest,
background_tasks: BackgroundTasks, # noqa: ARG001
background_tasks: BackgroundTasks,
http_request: Request,
api_key_user: Annotated[UserRead, Depends(api_key_security)],
) -> WorkflowExecutionResponse | WorkflowJobResponse | StreamingResponse:
"""Execute a workflow with multiple execution modes.
"""Execute a workflow with support for multiple execution modes.
- **sync**: Returns complete results immediately (background=False, stream=False)
- **stream**: Returns server-sent events in real-time (stream=True)
- **background**: Starts job and returns job ID immediately (background=True)
This endpoint supports three execution modes:
- **Synchronous** (background=False, stream=False): Returns complete results immediately
- **Streaming** (stream=True): Returns server-sent events in real-time (not yet implemented)
- **Background** (background=True): Starts job and returns job ID (not yet implemented)
Error Handling Strategy:
- System errors (404, 500, 503, 504): Returned as HTTP error responses
- Component execution errors: Returned as HTTP 200 with errors in response body
Args:
workflow_request: The workflow execution request containing flow_id, inputs, and mode flags
background_tasks: FastAPI background tasks for async operations
http_request: The HTTP request object for extracting headers
api_key_user: Authenticated user from API key
Returns:
- WorkflowExecutionResponse: For synchronous execution (HTTP 200)
- WorkflowJobResponse: For background execution (HTTP 202, not yet implemented)
- StreamingResponse: For streaming execution (not yet implemented)
Raises:
HTTPException:
- 403: Developer API disabled
- 404: Flow not found or user lacks access
- 500: Invalid flow data or validation error
- 501: Streaming or background mode not yet implemented
- 503: Database unavailable
- 504: Execution timeout exceeded
"""
flow = await get_flow_by_id_or_endpoint_name(workflow_request.flow_id, api_key_user.id)
if not flow:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Flow {workflow_request.flow_id} not found")
# Validate flow exists and user has permission
try:
flow = await get_flow_by_id_or_endpoint_name(workflow_request.flow_id, api_key_user.id)
except HTTPException as e:
# get_flow_by_id_or_endpoint_name raises HTTPException with string detail
# Convert to structured format
if e.status_code == status.HTTP_404_NOT_FOUND:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"error": "Flow not found",
"code": "FLOW_NOT_FOUND",
"message": f"Flow '{workflow_request.flow_id}' does not exist or you don't have access to it",
"flow_id": workflow_request.flow_id,
},
) from e
# Re-raise other HTTPExceptions as-is
raise
except PydanticValidationError as e:
# Flow data validation errors (invalid flow structure)
error_msg = f"Flow has invalid data structure: {e!s}"
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={
"error": "Invalid flow data",
"code": "INVALID_FLOW_DATA",
"message": error_msg,
"flow_id": workflow_request.flow_id,
},
) from e
except OperationalError as e:
# Database errors specifically
error_msg = f"Failed to fetch flow: {e!s}"
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail={
"error": "Service unavailable",
"code": "DATABASE_ERROR",
"message": error_msg,
"flow_id": workflow_request.flow_id,
},
) from e
# TODO: Implementation
raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /workflow execution yet")
# Generate job_id for tracking
job_id = str(uuid4())
# Phase 1: Background mode (to be implemented)
if workflow_request.background:
raise HTTPException(
status_code=status.HTTP_501_NOT_IMPLEMENTED,
detail={
"error": "Not implemented",
"code": "NOT_IMPLEMENTED",
"message": "Background execution not yet implemented",
},
)
# Phase 2: Streaming mode (to be implemented)
if workflow_request.stream:
raise HTTPException(
status_code=status.HTTP_501_NOT_IMPLEMENTED,
detail={
"error": "Not implemented",
"code": "NOT_IMPLEMENTED",
"message": "Streaming execution not yet implemented",
},
)
# Phase 3: Synchronous execution (default)
# Note: flow is guaranteed to be non-None here because get_flow_by_id_or_endpoint_name
# raises HTTPException if flow is not found
try:
return await execute_sync_workflow_with_timeout(
workflow_request=workflow_request,
flow=flow,
job_id=job_id,
api_key_user=api_key_user,
background_tasks=background_tasks,
http_request=http_request,
)
except WorkflowTimeoutError:
raise HTTPException(
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
detail={
"error": "Execution timeout",
"code": "EXECUTION_TIMEOUT",
"message": f"Workflow execution exceeded {EXECUTION_TIMEOUT} seconds",
"job_id": job_id,
"flow_id": workflow_request.flow_id,
"timeout_seconds": EXECUTION_TIMEOUT,
},
) from None
except WorkflowValidationError as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={
"error": "Workflow validation error",
"code": "INVALID_FLOW_DATA",
"message": str(e),
"job_id": job_id,
"flow_id": workflow_request.flow_id,
},
) from e
async def execute_sync_workflow_with_timeout(
workflow_request: WorkflowExecutionRequest,
flow: FlowRead,
job_id: str,
api_key_user: UserRead,
background_tasks: BackgroundTasks,
http_request: Request,
) -> WorkflowExecutionResponse:
"""Execute workflow with timeout protection.
Args:
workflow_request: The workflow execution request
flow: The flow to execute
job_id: Generated job ID for tracking
api_key_user: Authenticated user
background_tasks: FastAPI background tasks
http_request: The HTTP request object for extracting headers
Returns:
WorkflowExecutionResponse with complete results
Raises:
WorkflowTimeoutError: If execution exceeds timeout
WorkflowValidationError: If flow validation fails
"""
try:
return await asyncio.wait_for(
execute_sync_workflow(
workflow_request=workflow_request,
flow=flow,
job_id=job_id,
api_key_user=api_key_user,
background_tasks=background_tasks,
http_request=http_request,
),
timeout=EXECUTION_TIMEOUT,
)
except asyncio.TimeoutError as e:
msg = f"Execution exceeded {EXECUTION_TIMEOUT} seconds"
raise WorkflowTimeoutError(msg) from e
async def execute_sync_workflow(
workflow_request: WorkflowExecutionRequest,
flow: FlowRead,
job_id: str,
api_key_user: UserRead,
background_tasks: BackgroundTasks, # noqa: ARG001
http_request: Request,
) -> WorkflowExecutionResponse:
"""Execute workflow synchronously and return complete results.
This function implements a two-tier error handling strategy:
1. System-level errors (validation, graph build): Raised as exceptions
2. Component execution errors: Returned in response body with HTTP 200
This approach allows clients to receive partial results even when some
components fail, which is useful for debugging and incremental processing.
Execution Flow:
1. Parse flat inputs into tweaks and session_id
2. Validate flow data exists
3. Extract context from HTTP headers
4. Build graph from flow data with tweaks applied
5. Identify terminal nodes for execution
6. Execute graph and collect results
7. Convert V1 RunResponse to V2 WorkflowExecutionResponse
Args:
workflow_request: The workflow execution request with inputs and configuration
flow: The flow model from database
job_id: Generated job ID for tracking this execution
api_key_user: Authenticated user for permission checks
background_tasks: FastAPI background tasks (unused in sync mode)
http_request: The HTTP request object for extracting headers
Returns:
WorkflowExecutionResponse: Complete execution results with outputs and metadata
Raises:
WorkflowValidationError: If flow data is None or graph build fails
"""
# Parse flat inputs structure
tweaks, session_id = parse_flat_inputs(workflow_request.inputs or {})
# Validate flow data - this is a system error, not execution error
if flow.data is None:
msg = f"Flow {flow.id} has no data. The flow may be corrupted."
raise WorkflowValidationError(msg)
# Extract request-level variables from headers (similar to V1)
# Headers with prefix X-LANGFLOW-GLOBAL-VAR-* are extracted and made available to components
request_variables = extract_global_variables_from_headers(http_request.headers)
# Build context from request variables (similar to V1's _run_flow_internal)
context = {"request_variables": request_variables} if request_variables else None
# Build graph - system error if this fails
try:
flow_id_str = str(flow.id)
user_id = str(api_key_user.id)
# Use deepcopy to prevent mutation of the original flow.data
# process_tweaks modifies nested dictionaries in-place
graph_data = deepcopy(flow.data)
graph_data = process_tweaks(graph_data, tweaks, stream=False)
# Pass context to graph (similar to V1's simple_run_flow)
# This allows components to access request metadata via graph.context
graph = Graph.from_payload(
graph_data, flow_id=flow_id_str, user_id=user_id, flow_name=flow.name, context=context
)
# Set run_id for tracing/logging (similar to V1's simple_run_flow)
graph.set_run_id(job_id)
except Exception as e:
msg = f"Failed to build graph from flow data: {e!s}"
raise WorkflowValidationError(msg) from e
# Get terminal nodes - these are the outputs we want
terminal_node_ids = graph.get_terminal_nodes()
# Execute graph - component errors are caught and returned in response body
try:
task_result, execution_session_id = await run_graph_internal(
graph=graph,
flow_id=flow_id_str,
session_id=session_id,
inputs=None,
outputs=terminal_node_ids,
stream=False,
)
# Build RunResponse
run_response = RunResponse(outputs=task_result, session_id=execution_session_id)
# Convert to WorkflowExecutionResponse
return run_response_to_workflow_response(
run_response=run_response,
flow_id=workflow_request.flow_id,
job_id=job_id,
workflow_request=workflow_request,
graph=graph,
)
except asyncio.CancelledError:
# Re-raise CancelledError to allow timeout mechanism to work properly
# This ensures asyncio.wait_for() can properly cancel and raise TimeoutError
raise
except Exception as exc: # noqa: BLE001
# Component execution errors - return in response body with HTTP 200
# This allows partial results and detailed error information per component
return create_error_response(
flow_id=workflow_request.flow_id,
job_id=job_id,
workflow_request=workflow_request,
error=exc,
)
@router.get(
@ -74,13 +406,38 @@ async def get_workflow_status(
api_key_user: Annotated[UserRead, Depends(api_key_security)], # noqa: ARG001
job_id: Annotated[str, Query(description="Job ID to query")], # noqa: ARG001
) -> WorkflowExecutionResponse | StreamingResponse:
"""Get workflow job status and results by job ID."""
# TODO: Implementation
"""Get workflow job status and results by job ID.
This endpoint allows clients to poll for the status of background workflow executions.
Args:
api_key_user: Authenticated user from API key
job_id: The job ID returned from a background execution request
Returns:
- WorkflowExecutionResponse: If job is complete or failed
- StreamingResponse: If job is still running (for streaming mode)
Raises:
HTTPException:
- 403: Developer API disabled or unauthorized
- 404: Job ID not found
- 501: Not yet implemented
Note:
This endpoint is not yet implemented. It will be added in a future release
to support background and streaming execution modes.
"""
# TODO: Implement job status tracking and retrieval
# - Store job metadata in database or cache
# - Track execution progress and status
# - Return appropriate response based on job state
raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /status yet")
@router.post(
"/stop",
response_model=WorkflowStopResponse,
summary="Stop Workflow",
description="Stop a running workflow execution",
)
@ -90,9 +447,33 @@ async def stop_workflow(
) -> WorkflowStopResponse:
"""Stop a running workflow execution by job_id.
Parameters:
- job_id: The specific job ID to stop
- force: Whether to force stop the workflow (optional)
This endpoint allows clients to gracefully or forcefully stop a running workflow.
Args:
request: Stop request containing job_id and optional force flag
api_key_user: Authenticated user from API key
Returns:
WorkflowStopResponse: Confirmation of stop request with final job status
Raises:
HTTPException:
- 403: Developer API disabled or unauthorized
- 404: Job ID not found
- 409: Job already completed or cannot be stopped
- 501: Not yet implemented
Note:
This endpoint is not yet implemented. It will be added in a future release
to support graceful cancellation of background and streaming executions.
Planned behavior:
- force=False: Graceful shutdown (complete current component)
- force=True: Immediate termination
"""
# TODO: Implementation
# TODO: Implement workflow cancellation
# - Locate running job by job_id
# - Send cancellation signal to execution engine
# - Handle graceful vs forced termination
# - Update job status and return response
raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /stop yet")

12
src/backend/base/langflow/exceptions/api.py
View File

@ -10,6 +10,18 @@ class InvalidChatInputError(Exception):
pass
class WorkflowExecutionError(Exception):
"""Base exception for workflow execution errors."""
class WorkflowTimeoutError(WorkflowExecutionError):
"""Workflow execution timeout."""
class WorkflowValidationError(WorkflowExecutionError):
"""Workflow validation error (e.g., invalid flow data, graph build failure)."""
# create a pidantic documentation for this class
class ExceptionBody(BaseModel):
message: str | list[str]

2
src/backend/base/langflow/helpers/flow.py
View File

@ -396,7 +396,7 @@ def get_arg_names(inputs: list[Vertex]) -> list[dict[str, str]]:
]
async def get_flow_by_id_or_endpoint_name(flow_id_or_name: str, user_id: str | UUID | None = None) -> FlowRead | None:
async def get_flow_by_id_or_endpoint_name(flow_id_or_name: str, user_id: str | UUID | None = None) -> FlowRead:
async with session_scope() as session:
endpoint_name = None
try:

1237
src/backend/tests/unit/api/v2/test_converters.py Normal file
View File

File diff suppressed because it is too large Load Diff

876
src/backend/tests/unit/api/v2/test_workflow.py
View File

@ -1,10 +1,40 @@
"""Comprehensive unit tests for V2 Workflow API endpoints.
This test module provides extensive coverage of the workflow execution endpoints,
including authentication, authorization, error handling, and execution modes.
Test Coverage:
- Developer API protection (enabled/disabled scenarios)
- API key authentication requirements
- Flow validation and error handling
- Database error handling
- Execution timeout protection
- Synchronous execution with various component types
- Error response structure validation
- Multiple execution modes (sync, stream, background)
Test Organization:
- TestWorkflowDeveloperAPIProtection: Tests developer API feature flag
- TestWorkflowErrorHandling: Tests comprehensive error scenarios
- TestWorkflowSyncExecution: Tests successful execution flows
Test Strategy:
- Uses real database with proper cleanup
- Mocks external dependencies (LLM APIs, file operations)
- Tests both success and failure paths
- Validates response structure and status codes
"""
import asyncio
from unittest.mock import MagicMock, patch
from uuid import uuid4
import pytest
from httpx import AsyncClient
from langflow.exceptions.api import WorkflowValidationError
from langflow.services.database.models.flow.model import Flow
from lfx.services.deps import session_scope
from sqlalchemy.exc import OperationalError
class TestWorkflowDeveloperAPIProtection:
@ -42,9 +72,10 @@ class TestWorkflowDeveloperAPIProtection:
headers=headers,
)
assert response.status_code == 404
assert response.status_code == 403
result = response.json()
assert "This endpoint is not available" in result["detail"]
assert result["detail"]["code"] == "DEVELOPER_API_DISABLED"
assert "Developer API" in result["detail"]["message"]
async def test_stop_workflow_blocked_when_dev_api_disabled(
self,
@ -62,9 +93,9 @@ class TestWorkflowDeveloperAPIProtection:
headers=headers,
)
assert response.status_code == 404
assert response.status_code == 403
result = response.json()
assert "This endpoint is not available" in result["detail"]
assert result["detail"]["code"] == "DEVELOPER_API_DISABLED"
@pytest.fixture
def mock_settings_dev_api_enabled(self):
@ -100,9 +131,9 @@ class TestWorkflowDeveloperAPIProtection:
# Should return 404 because flow doesn't exist, NOT because endpoint is disabled
assert response.status_code == 404
assert "Flow identifier" in response.text
assert "not found" in response.text
assert "This endpoint is not available" not in response.text
result = response.json()
assert result["detail"]["code"] == "FLOW_NOT_FOUND"
assert "550e8400-e29b-41d4-a716-446655440000" in result["detail"]["flow_id"]
async def test_get_workflow_allowed_when_dev_api_enabled_job_not_found(
self,
@ -158,9 +189,9 @@ class TestWorkflowDeveloperAPIProtection:
headers=headers,
)
assert response.status_code == 404
assert response.status_code == 403
result = response.json()
assert "This endpoint is not available" in result["detail"]
assert result["detail"]["code"] == "DEVELOPER_API_DISABLED"
async def test_execute_workflow_allowed_when_dev_api_enabled_flow_exists(
self,
@ -168,7 +199,7 @@ class TestWorkflowDeveloperAPIProtection:
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test POST /workflow allowed when dev API enabled - flow exists (501 not implemented)."""
"""Test POST /workflow allowed when dev API enabled - flow exists and executes."""
flow_id = uuid4()
# Create a flow in the database using the established pattern
@ -194,10 +225,17 @@ class TestWorkflowDeveloperAPIProtection:
headers=headers,
)
# Should return 501 because endpoint is not implemented yet
assert response.status_code == 501
assert "Not implemented" in response.text
assert "This endpoint is not available" not in response.text
# Should return 200 because flow is valid (empty nodes/edges is valid)
# The execution will complete successfully with no outputs
assert response.status_code == 200
result = response.json()
# Verify response contains expected fields with proper structure
assert "outputs" in result or "errors" in result
if "outputs" in result:
assert isinstance(result["outputs"], dict)
if "errors" in result:
assert isinstance(result["errors"], list)
finally:
# Clean up the flow following established pattern
@ -270,11 +308,821 @@ class TestWorkflowDeveloperAPIProtection:
assert response.status_code == 403
assert "API key must be passed" in response.json()["detail"]
class TestWorkflowErrorHandling:
"""Test comprehensive error handling for workflow endpoints."""
@pytest.fixture
def mock_settings_dev_api_enabled(self):
"""Mock settings with developer API enabled."""
with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service:
mock_service = MagicMock()
mock_settings = MagicMock()
mock_settings.developer_api_enabled = True
mock_service.settings = mock_settings
mock_get_settings_service.return_value = mock_service
yield mock_settings
async def test_flow_not_found_returns_404_with_error_code(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that non-existent flow returns 404 with FLOW_NOT_FOUND error code."""
flow_id = str(uuid4())
request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 404
result = response.json()
assert result["detail"]["code"] == "FLOW_NOT_FOUND"
assert result["detail"]["flow_id"] == flow_id
assert "does not exist" in result["detail"]["message"]
async def test_database_error_returns_503(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that database errors return 503 with DATABASE_ERROR code."""
flow_id = str(uuid4())
request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None}
# Mock get_flow_by_id_or_endpoint_name to raise OperationalError
with patch("langflow.api.v2.workflow.get_flow_by_id_or_endpoint_name") as mock_get_flow:
mock_get_flow.side_effect = OperationalError("statement", "params", "orig")
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 503
result = response.json()
assert result["detail"]["code"] == "DATABASE_ERROR"
assert "Failed to fetch flow" in result["detail"]["message"]
assert result["detail"]["flow_id"] == flow_id
async def test_flow_with_no_data_returns_500(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that flow with no data returns 500 with INVALID_FLOW_DATA code."""
flow_id = uuid4()
# Create a flow with no data
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Flow with no data",
description="Test flow with no data",
data=None, # No data
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 500
result = response.json()
assert result["detail"]["code"] == "INVALID_FLOW_DATA"
assert "has no data" in result["detail"]["message"]
assert result["detail"]["flow_id"] == str(flow_id)
assert "job_id" in result["detail"]
finally:
# Clean up
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_graph_build_failure_returns_500(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that graph build failure returns 500 with INVALID_FLOW_DATA code."""
flow_id = uuid4()
# Create a flow with invalid data that will fail validation/graph building
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Flow with invalid data",
description="Test flow with invalid graph data",
data={"invalid": "data"}, # Invalid graph data (missing 'nodes' field)
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 500
result = response.json()
assert result["detail"]["code"] == "INVALID_FLOW_DATA"
# The error message should indicate invalid flow data structure
error_msg = result["detail"]["message"].lower()
assert "invalid data structure" in error_msg or "must have nodes" in error_msg
assert result["detail"]["flow_id"] == str(flow_id)
finally:
# Clean up
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_execution_timeout_with_real_delay(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that execution timeout works with real async delay."""
flow_id = uuid4()
# Create a valid flow
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Test Flow",
description="Test flow for timeout",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None}
# Mock execute_sync_workflow to sleep longer than timeout
async def slow_execution(*args, **kwargs): # noqa: ARG001
await asyncio.sleep(2) # Sleep for 2 seconds
return MagicMock()
# Temporarily reduce timeout for testing
with (
patch("langflow.api.v2.workflow.execute_sync_workflow", side_effect=slow_execution),
patch("langflow.api.v2.workflow.EXECUTION_TIMEOUT", 0.5), # 0.5 second timeout
):
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 504
result = response.json()
assert result["detail"]["code"] == "EXECUTION_TIMEOUT"
assert "exceeded" in result["detail"]["message"]
assert result["detail"]["flow_id"] == str(flow_id)
assert "job_id" in result["detail"]
finally:
# Clean up
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_background_mode_returns_501(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that background mode returns 501 with NOT_IMPLEMENTED code."""
flow_id = uuid4()
# Create a valid flow
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Test Flow",
description="Test flow",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {
"flow_id": str(flow_id),
"background": True, # Background mode
"stream": False,
"inputs": None,
}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 501
result = response.json()
assert result["detail"]["code"] == "NOT_IMPLEMENTED"
assert "Background execution not yet implemented" in result["detail"]["message"]
finally:
# Clean up
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_streaming_mode_returns_501(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that streaming mode returns 501 with NOT_IMPLEMENTED code."""
flow_id = uuid4()
# Create a valid flow
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Test Flow",
description="Test flow",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {
"flow_id": str(flow_id),
"background": False,
"stream": True, # Streaming mode
"inputs": None,
}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 501
result = response.json()
assert result["detail"]["code"] == "NOT_IMPLEMENTED"
assert "Streaming execution not yet implemented" in result["detail"]["message"]
finally:
# Clean up
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_error_response_structure(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that all error responses have consistent structure."""
flow_id = str(uuid4())
request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 404
result = response.json()
# Verify error structure
assert "detail" in result
assert "error" in result["detail"]
assert "code" in result["detail"]
assert "message" in result["detail"]
assert "flow_id" in result["detail"]
# Verify types
assert isinstance(result["detail"]["error"], str)
assert isinstance(result["detail"]["code"], str)
assert isinstance(result["detail"]["message"], str)
assert isinstance(result["detail"]["flow_id"], str)
async def test_workflow_validation_error_propagation(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that WorkflowValidationError is properly caught and converted to 500."""
flow_id = uuid4()
# Create a flow
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Test Flow",
description="Test flow",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None}
# Mock execute_sync_workflow to raise WorkflowValidationError
with patch("langflow.api.v2.workflow.execute_sync_workflow") as mock_execute:
mock_execute.side_effect = WorkflowValidationError("Test validation error")
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 500
result = response.json()
assert result["detail"]["code"] == "INVALID_FLOW_DATA"
assert "Test validation error" in result["detail"]["message"]
finally:
# Clean up
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
# Test GET /workflow without API key
response = await client.get("api/v2/workflow?job_id=550e8400-e29b-41d4-a716-446655440001")
assert response.status_code == 403
assert "API key must be passed" in response.json()["detail"]
class TestWorkflowSyncExecution:
"""Test synchronous workflow execution with realistic component mocking."""
@pytest.fixture
def mock_settings_dev_api_enabled(self):
"""Mock settings with developer API enabled."""
with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service:
mock_service = MagicMock()
mock_settings = MagicMock()
mock_settings.developer_api_enabled = True
mock_service.settings = mock_settings
mock_get_settings_service.return_value = mock_service
yield mock_settings
async def test_sync_execution_with_empty_flow_returns_200(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test sync execution with empty flow returns 200 with empty outputs."""
flow_id = uuid4()
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Empty Flow",
description="Flow with no nodes",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 200
result = response.json()
# Verify response structure
assert "flow_id" in result
assert result["flow_id"] == str(flow_id)
assert "job_id" in result
# Verify outputs or errors are present with actual content
assert "outputs" in result or "errors" in result
if "outputs" in result:
assert isinstance(result["outputs"], dict)
if "errors" in result:
assert isinstance(result["errors"], list)
# session_id is only present if provided in inputs
finally:
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_sync_execution_component_error_returns_200_with_error_in_body(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that component execution errors return 200 with error in response body."""
flow_id = uuid4()
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Test Flow",
description="Flow for testing component errors",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None}
# Mock run_graph_internal to raise a component execution error
with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run:
mock_run.side_effect = Exception("Component execution failed: LLM API key not configured")
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
# Component errors should return 200 with error in body
assert response.status_code == 200
result = response.json()
# Verify error is in response body (via create_error_response)
assert "errors" in result
assert len(result["errors"]) > 0
assert "Component execution failed" in str(result["errors"][0])
assert result["status"] == "failed"
assert result["flow_id"] == str(flow_id)
assert "job_id" in result
finally:
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_sync_execution_with_chat_input_output(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test sync execution with ChatInput and ChatOutput components."""
flow_id = uuid4()
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Chat Flow",
description="Flow with chat input/output",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
# Input format: component_id.param = value
request_data = {
"flow_id": str(flow_id),
"background": False,
"stream": False,
"inputs": {
"ChatInput-abc123.input_value": "Hello, how are you?",
"ChatInput-abc123.session_id": "session-456",
},
}
# Mock successful execution with ChatOutput
mock_result_data = MagicMock()
mock_result_data.component_id = "ChatOutput-xyz789"
mock_result_data.outputs = {"message": {"message": "I'm doing well, thank you for asking!", "type": "text"}}
mock_result_data.metadata = {}
# Wrap ResultData in RunOutputs
mock_run_output = MagicMock()
mock_run_output.outputs = [mock_result_data]
with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run:
# run_graph_internal returns tuple[list[RunOutputs], str]
mock_run.return_value = ([mock_run_output], "session-456")
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 200
result = response.json()
# Verify response structure
assert result["flow_id"] == str(flow_id)
assert "job_id" in result
assert "outputs" in result
# Note: Detailed content validation requires proper graph/vertex mocking
# which is beyond the scope of unit tests. Integration tests should validate content.
# Verify inputs were echoed back
assert "inputs" in result
assert result["inputs"] == request_data["inputs"]
# Verify session_id is present when provided in inputs
if "session_id" in result:
assert result["session_id"] == "session-456"
finally:
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_sync_execution_with_llm_output(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test sync execution with LLM component output including model metadata."""
flow_id = uuid4()
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="LLM Flow",
description="Flow with LLM component",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {
"flow_id": str(flow_id),
"background": False,
"stream": False,
"inputs": {
"ChatInput-abc.input_value": "Explain quantum computing",
"OpenAIModel-def.temperature": 0.7,
},
}
# Mock LLM execution with model metadata
mock_result_data = MagicMock()
mock_result_data.component_id = "OpenAIModel-def"
mock_result_data.outputs = {
"model_output": {
"message": {
"message": "Quantum computing uses quantum mechanics...",
"model_name": "gpt-4",
"type": "text",
}
}
}
mock_result_data.metadata = {"tokens_used": 150}
# Wrap ResultData in RunOutputs
mock_run_output = MagicMock()
mock_run_output.outputs = [mock_result_data]
with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run:
# run_graph_internal returns tuple[list[RunOutputs], str]
mock_run.return_value = ([mock_run_output], "session-789")
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 200
result = response.json()
assert result["flow_id"] == str(flow_id)
assert "job_id" in result
assert "outputs" in result
assert "metadata" in result
# Note: Detailed content validation requires proper graph/vertex mocking
finally:
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_sync_execution_with_file_save_output(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test sync execution with SaveToFile component."""
flow_id = uuid4()
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="File Save Flow",
description="Flow with file save component",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {
"flow_id": str(flow_id),
"background": False,
"stream": False,
"inputs": {
"TextInput-abc.text": "Content to save",
"SaveToFile-xyz.file_path": "/tmp/output.txt", # noqa: S108
},
}
# Mock SaveToFile execution
mock_result_data = MagicMock()
mock_result_data.component_id = "SaveToFile-xyz"
mock_result_data.outputs = {
"message": {"message": "File saved successfully to /tmp/output.txt", "type": "text"}
}
mock_result_data.metadata = {"bytes_written": 1024}
# Wrap ResultData in RunOutputs
mock_run_output = MagicMock()
mock_run_output.outputs = [mock_result_data]
with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run:
# run_graph_internal returns tuple[list[RunOutputs], str]
mock_run.return_value = ([mock_run_output], "session-101")
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 200
result = response.json()
assert result["flow_id"] == str(flow_id)
assert "job_id" in result
assert "outputs" in result
assert "metadata" in result
# Note: Detailed content validation requires proper graph/vertex mocking
finally:
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_sync_execution_with_multiple_terminal_nodes(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test sync execution with multiple terminal nodes (outputs)."""
flow_id = uuid4()
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Multi-Output Flow",
description="Flow with multiple terminal nodes",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {
"flow_id": str(flow_id),
"background": False,
"stream": False,
"inputs": {"ChatInput-abc.input_value": "Process this"},
}
# Mock execution with multiple outputs
mock_chat_output = MagicMock()
mock_chat_output.component_id = "ChatOutput-aaa"
mock_chat_output.outputs = {"message": {"message": "Chat response", "type": "text"}}
mock_file_output = MagicMock()
mock_file_output.component_id = "SaveToFile-bbb"
mock_file_output.outputs = {"message": {"message": "File saved successfully", "type": "text"}}
with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run:
# run_graph_internal returns tuple[list[RunOutputs], str]
mock_run.return_value = ([mock_chat_output, mock_file_output], "session-202")
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 200
result = response.json()
assert result["flow_id"] == str(flow_id)
assert "job_id" in result
assert "outputs" in result
# Note: Detailed content validation requires proper graph/vertex mocking
finally:
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
async def test_sync_execution_response_structure_validation(
self,
client: AsyncClient,
created_api_key,
mock_settings_dev_api_enabled, # noqa: ARG002
):
"""Test that sync execution response has correct WorkflowExecutionResponse structure."""
flow_id = uuid4()
async with session_scope() as session:
flow = Flow(
id=flow_id,
name="Test Flow",
description="Flow for response validation",
data={"nodes": [], "edges": []},
user_id=created_api_key.user_id,
)
session.add(flow)
await session.flush()
await session.refresh(flow)
try:
request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None}
headers = {"x-api-key": created_api_key.api_key}
response = await client.post("api/v2/workflow", json=request_data, headers=headers)
assert response.status_code == 200
result = response.json()
# Verify WorkflowExecutionResponse structure
assert "flow_id" in result
assert isinstance(result["flow_id"], str)
assert result["flow_id"] == str(flow_id)
assert "job_id" in result
assert isinstance(result["job_id"], str)
# session_id is optional - only present if provided in inputs
if "session_id" in result:
assert isinstance(result["session_id"], str)
assert "object" in result
assert result["object"] == "response"
assert "created_timestamp" in result
assert isinstance(result["created_timestamp"], str)
assert "status" in result
assert result["status"] in ["completed", "failed", "running", "queued"]
assert "errors" in result
assert isinstance(result["errors"], list)
assert "inputs" in result
assert isinstance(result["inputs"], dict)
assert "outputs" in result
assert isinstance(result["outputs"], dict)
assert "metadata" in result
assert isinstance(result["metadata"], dict)
finally:
async with session_scope() as session:
flow = await session.get(Flow, flow_id)
if flow:
await session.delete(flow)
# Test POST /workflow/stop without API key
response = await client.post(
"api/v2/workflow/stop",

42
src/lfx/src/lfx/schema/workflow.py
View File

@ -36,14 +36,6 @@ class ComponentOutput(BaseModel):
metadata: dict[str, Any] | None = None
class GlobalInputs(BaseModel):
"""Global inputs that apply to all input components in the workflow."""
input_value: str | None = Field(None, description="The input value to send to input components")
input_type: str = Field("chat", description="The type of input (chat, text, etc.)")
session_id: str | None = Field(None, description="Session ID for conversation continuity")
class WorkflowExecutionRequest(BaseModel):
"""Request schema for workflow execution."""
@ -51,7 +43,7 @@ class WorkflowExecutionRequest(BaseModel):
stream: bool = False
flow_id: str
inputs: dict[str, Any] | None = Field(
None, description="Inputs with 'global' key for global inputs and component IDs for component-specific tweaks"
None, description="Component-specific inputs in flat format: 'component_id.param_name': value"
)
model_config = ConfigDict(
@ -62,26 +54,28 @@ class WorkflowExecutionRequest(BaseModel):
"stream": False,
"flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b",
"inputs": {
"global": {
"input_value": "Hello, how can you help me today?",
"input_type": "chat",
"session_id": "session-123",
},
"llm_component": {"temperature": 0.7, "max_tokens": 100},
"opensearch_component": {"opensearch_url": "https://opensearch:9200"},
"ChatInput-abc.input_value": "Hello, how can you help me today?",
"ChatInput-abc.session_id": "session-123",
"LLM-xyz.temperature": 0.7,
"LLM-xyz.max_tokens": 100,
"OpenSearch-def.opensearch_url": "https://opensearch:9200",
},
},
{
"background": True,
"stream": False,
"flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b",
"inputs": {"global": {"input_value": "Process this in the background", "input_type": "text"}},
"inputs": {
"ChatInput-abc.input_value": "Process this in the background",
},
},
{
"background": False,
"stream": True,
"flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b",
"inputs": {"chat_component": {"text": "Stream this conversation"}},
"inputs": {
"ChatInput-abc.input_value": "Stream this conversation",
},
},
]
},
@ -94,7 +88,7 @@ class WorkflowExecutionResponse(BaseModel):
flow_id: str
job_id: str
object: Literal["response"] = "response"
object: Literal["response"] = Field(default="response")
created_timestamp: str
status: JobStatus
errors: list[ErrorDetail] = []
@ -107,6 +101,7 @@ class WorkflowJobResponse(BaseModel):
"""Background job response."""
job_id: str
object: Literal["job"] = Field(default="job")
created_timestamp: str
status: JobStatus
errors: list[ErrorDetail] = []
@ -146,7 +141,14 @@ WORKFLOW_EXECUTION_RESPONSES = {
"oneOf": [
WorkflowExecutionResponse.model_json_schema(),
WorkflowJobResponse.model_json_schema(),
]
],
"discriminator": {
"propertyName": "object",
"mapping": {
"response": "#/components/schemas/WorkflowExecutionResponse",
"job": "#/components/schemas/WorkflowJobResponse",
},
},
}
},
"text/event-stream": {