diff --git a/src/backend/base/langflow/api/router.py b/src/backend/base/langflow/api/router.py index e11f89e416..d5d1b4a8b3 100644 --- a/src/backend/base/langflow/api/router.py +++ b/src/backend/base/langflow/api/router.py @@ -27,6 +27,7 @@ from langflow.api.v1.voice_mode import router as voice_mode_router from langflow.api.v2 import files_router as files_router_v2 from langflow.api.v2 import mcp_router as mcp_router_v2 from langflow.api.v2 import registration_router as registration_router_v2 +from langflow.api.v2 import workflow_router as workflow_router_v2 router_v1 = APIRouter( prefix="/v1", @@ -61,6 +62,7 @@ router_v1.include_router(model_options_router) router_v2.include_router(files_router_v2) router_v2.include_router(mcp_router_v2) router_v2.include_router(registration_router_v2) +router_v2.include_router(workflow_router_v2) router = APIRouter( prefix="/api", diff --git a/src/backend/base/langflow/api/v2/__init__.py b/src/backend/base/langflow/api/v2/__init__.py index 7eccfb56a7..b3a6e06f57 100644 --- a/src/backend/base/langflow/api/v2/__init__.py +++ b/src/backend/base/langflow/api/v2/__init__.py @@ -1,9 +1,8 @@ -from langflow.api.v2.files import router as files_router -from langflow.api.v2.mcp import router as mcp_router -from langflow.api.v2.registration import router as registration_router +"""V2 API module.""" -__all__ = [ - "files_router", - "mcp_router", - "registration_router", -] +from .files import router as files_router +from .mcp import router as mcp_router +from .registration import router as registration_router +from .workflow import router as workflow_router + +__all__ = ["files_router", "mcp_router", "registration_router", "workflow_router"] diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py new file mode 100644 index 0000000000..cca3c36e79 --- /dev/null +++ b/src/backend/base/langflow/api/v2/workflow.py @@ -0,0 +1,98 @@ +"""V2 Workflow execution endpoints.""" + +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status +from fastapi.responses import StreamingResponse +from lfx.schema.workflow import ( + WORKFLOW_EXECUTION_RESPONSES, + WORKFLOW_STATUS_RESPONSES, + WorkflowExecutionRequest, + WorkflowExecutionResponse, + WorkflowJobResponse, + WorkflowStopRequest, + WorkflowStopResponse, +) +from lfx.services.deps import get_settings_service + +from langflow.helpers.flow import get_flow_by_id_or_endpoint_name +from langflow.services.auth.utils import api_key_security +from langflow.services.database.models.user.model import UserRead + + +def check_developer_api_enabled() -> None: + """Check if developer API is enabled, raise HTTPException if not.""" + 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", + ) + + +router = APIRouter(prefix="/workflow", tags=["Workflow"], dependencies=[Depends(check_developer_api_enabled)]) + + +@router.post( + "", + response_model=None, + response_model_exclude_none=True, + responses=WORKFLOW_EXECUTION_RESPONSES, + summary="Execute Workflow", + description="Execute a workflow with support for sync, stream, and background modes", +) +async def execute_workflow( + workflow_request: WorkflowExecutionRequest, + background_tasks: BackgroundTasks, # noqa: ARG001 + api_key_user: Annotated[UserRead, Depends(api_key_security)], +) -> WorkflowExecutionResponse | WorkflowJobResponse | StreamingResponse: + """Execute a workflow with 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) + """ + 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") + + # TODO: Implementation + raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /workflow execution yet") + + +@router.get( + "", + response_model=None, + response_model_exclude_none=True, + responses=WORKFLOW_STATUS_RESPONSES, + summary="Get Workflow Status", + description="Get status of workflow job by job ID", +) +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 + raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /status yet") + + +@router.post( + "/stop", + summary="Stop Workflow", + description="Stop a running workflow execution", +) +async def stop_workflow( + request: WorkflowStopRequest, # noqa: ARG001 + api_key_user: Annotated[UserRead, Depends(api_key_security)], # noqa: ARG001 +) -> 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) + """ + # TODO: Implementation + raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /stop yet") diff --git a/src/backend/tests/unit/api/v2/test_workflow.py b/src/backend/tests/unit/api/v2/test_workflow.py new file mode 100644 index 0000000000..ecc7438390 --- /dev/null +++ b/src/backend/tests/unit/api/v2/test_workflow.py @@ -0,0 +1,284 @@ +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +import pytest +from httpx import AsyncClient +from langflow.services.database.models.flow.model import Flow +from lfx.services.deps import session_scope + + +class TestWorkflowDeveloperAPIProtection: + """Test developer API protection for workflow endpoints.""" + + @pytest.fixture + def mock_settings_dev_api_disabled(self): + """Mock settings with developer API disabled.""" + 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 = False + mock_service.settings = mock_settings + mock_get_settings_service.return_value = mock_service + yield mock_settings + + async def test_execute_workflow_blocked_when_dev_api_disabled( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_disabled, # noqa: ARG002 + ): + """Test workflow execution is blocked when developer API is disabled.""" + request_data = { + "flow_id": "550e8400-e29b-41d4-a716-446655440000", + "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 "This endpoint is not available" in result["detail"] + + async def test_stop_workflow_blocked_when_dev_api_disabled( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_disabled, # noqa: ARG002 + ): + """Test POST workflow/stop endpoint is blocked when developer API is disabled.""" + request_data = {"job_id": "550e8400-e29b-41d4-a716-446655440001"} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflow/stop", + json=request_data, + headers=headers, + ) + + assert response.status_code == 404 + result = response.json() + assert "This endpoint is not available" in result["detail"] + + @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_execute_workflow_allowed_when_dev_api_enabled_flow_not_found( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test POST workflow execution is allowed when developer API is enabled - flow not found.""" + request_data = { + "flow_id": "550e8400-e29b-41d4-a716-446655440000", # Non-existent 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, + ) + + # 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 + + async def test_get_workflow_allowed_when_dev_api_enabled_job_not_found( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test GET workflow endpoint is allowed when developer API is enabled - job not found.""" + headers = {"x-api-key": created_api_key.api_key} + response = await client.get( + "api/v2/workflow?job_id=550e8400-e29b-41d4-a716-446655440001", # Non-existent job ID + headers=headers, + ) + + # Should return 501 because endpoint is not implemented yet, NOT 404 because endpoint is disabled + assert response.status_code == 501 + assert "Not implemented" in response.text + assert "This endpoint is not available" not in response.text + + async def test_stop_workflow_allowed_when_dev_api_enabled_job_not_found( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test POST workflow/stop endpoint is allowed when developer API is enabled - job not found.""" + request_data = { + "job_id": "550e8400-e29b-41d4-a716-446655440001" # Non-existent job ID + } + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflow/stop", + json=request_data, + headers=headers, + ) + + # Should return 501 because endpoint is not implemented yet, NOT 404 because endpoint is disabled + assert response.status_code == 501 + assert "Not implemented" in response.text + assert "This endpoint is not available" not in response.text + + async def test_get_workflow_blocked_when_dev_api_disabled( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_disabled, # noqa: ARG002 + ): + """Test GET workflow endpoint is blocked when developer API is disabled.""" + headers = {"x-api-key": created_api_key.api_key} + response = await client.get( + "api/v2/workflow?job_id=550e8400-e29b-41d4-a716-446655440001", + headers=headers, + ) + + assert response.status_code == 404 + result = response.json() + assert "This endpoint is not available" in result["detail"] + + async def test_execute_workflow_allowed_when_dev_api_enabled_flow_exists( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test POST /workflow allowed when dev API enabled - flow exists (501 not implemented).""" + flow_id = uuid4() + + # Create a flow in the database using the established pattern + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Test Flow", + description="Test flow for API testing", + 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, + ) + + # 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 + + finally: + # Clean up the flow following established pattern + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_get_workflow_allowed_when_dev_api_enabled_job_exists( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test GET /workflow allowed when dev API enabled - job exists (501 not implemented).""" + # Since job management isn't implemented, we'll test with any job_id + # The endpoint should return 501 regardless of whether the job exists + headers = {"x-api-key": created_api_key.api_key} + response = await client.get( + "api/v2/workflow?job_id=550e8400-e29b-41d4-a716-446655440002", + headers=headers, + ) + + assert response.status_code == 501 + assert "Not implemented" in response.text + assert "This endpoint is not available" not in response.text + + async def test_stop_workflow_allowed_when_dev_api_enabled_job_exists( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test POST /workflow/stop allowed when dev API enabled - job exists (501 not implemented).""" + # Since job management isn't implemented, we'll test with any job_id + # The endpoint should return 501 regardless of whether the job exists + request_data = {"job_id": "550e8400-e29b-41d4-a716-446655440002"} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post( + "api/v2/workflow/stop", + json=request_data, + headers=headers, + ) + + assert response.status_code == 501 + assert "Not implemented" in response.text + assert "This endpoint is not available" not in response.text + + async def test_all_endpoints_require_api_key_authentication( + self, + client: AsyncClient, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that all workflow endpoints require API key authentication.""" + # Test POST /workflow without API key + request_data = { + "flow_id": "550e8400-e29b-41d4-a716-446655440000", + "background": False, + "stream": False, + "inputs": None, + } + + response = await client.post( + "api/v2/workflow", + json=request_data, + ) + # The API returns 403 Forbidden for missing API keys (not 401 Unauthorized) + # This is the correct behavior according to the api_key_security implementation + assert response.status_code == 403 + assert "API key must be passed" in response.json()["detail"] + + # 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"] + + # Test POST /workflow/stop without API key + response = await client.post( + "api/v2/workflow/stop", + json={"job_id": "550e8400-e29b-41d4-a716-446655440001"}, + ) + assert response.status_code == 403 + assert "API key must be passed" in response.json()["detail"] diff --git a/src/lfx/pyproject.toml b/src/lfx/pyproject.toml index 89cd16cd2d..3023bdf50f 100644 --- a/src/lfx/pyproject.toml +++ b/src/lfx/pyproject.toml @@ -93,6 +93,7 @@ omit = [ "*/tests/*", "*/__init__.py", "*/components/*", # Third-party integrations tracked separately + "*/schema/workflow.py", # Schema definitions don't require test coverage ] [tool.coverage.report] diff --git a/src/lfx/src/lfx/schema/__init__.py b/src/lfx/src/lfx/schema/__init__.py index 9d5b5ceae6..0356e07024 100644 --- a/src/lfx/src/lfx/schema/__init__.py +++ b/src/lfx/src/lfx/schema/__init__.py @@ -1,9 +1,12 @@ """Schema modules for lfx package.""" __all__ = [ + "ComponentOutput", "Data", "DataFrame", + "ErrorDetail", "InputValue", + "JobStatus", "Message", "OpenAIErrorResponse", "OpenAIResponsesRequest", @@ -11,6 +14,13 @@ __all__ = [ "OpenAIResponsesStreamChunk", "Tweaks", "UUIDstr", + "WorkflowExecutionRequest", + "WorkflowExecutionResponse", + "WorkflowJobResponse", + "WorkflowStatusResponse", + "WorkflowStopRequest", + "WorkflowStopResponse", + "WorkflowStreamEvent", "dotdict", ] @@ -61,6 +71,46 @@ def __getattr__(name: str): from .openai_responses_schemas import OpenAIErrorResponse return OpenAIErrorResponse + if name == "WorkflowExecutionRequest": + from .workflow import WorkflowExecutionRequest + + return WorkflowExecutionRequest + if name == "WorkflowExecutionResponse": + from .workflow import WorkflowExecutionResponse + + return WorkflowExecutionResponse + if name == "WorkflowJobResponse": + from .workflow import WorkflowJobResponse + + return WorkflowJobResponse + if name == "WorkflowStreamEvent": + from .workflow import WorkflowStreamEvent + + return WorkflowStreamEvent + if name == "WorkflowStatusResponse": + from .workflow import WorkflowStatusResponse + + return WorkflowStatusResponse + if name == "WorkflowStopRequest": + from .workflow import WorkflowStopRequest + + return WorkflowStopRequest + if name == "WorkflowStopResponse": + from .workflow import WorkflowStopResponse + + return WorkflowStopResponse + if name == "JobStatus": + from .workflow import JobStatus + + return JobStatus + if name == "ErrorDetail": + from .workflow import ErrorDetail + + return ErrorDetail + if name == "ComponentOutput": + from .workflow import ComponentOutput + + return ComponentOutput msg = f"module '{__name__}' has no attribute '{name}'" raise AttributeError(msg) diff --git a/src/lfx/src/lfx/schema/workflow.py b/src/lfx/src/lfx/schema/workflow.py new file mode 100644 index 0000000000..0c766055fd --- /dev/null +++ b/src/lfx/src/lfx/schema/workflow.py @@ -0,0 +1,171 @@ +"""Workflow execution schemas for V2 API.""" + +from __future__ import annotations + +from enum import Enum +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class JobStatus(str, Enum): + """Job execution status.""" + + QUEUED = "queued" + IN_PROGRESS = "in_progress" + COMPLETED = "completed" + FAILED = "failed" + ERROR = "error" + + +class ErrorDetail(BaseModel): + """Error detail schema.""" + + error: str + code: str | None = None + details: dict[str, Any] | None = None + + +class ComponentOutput(BaseModel): + """Component output schema.""" + + type: str = Field(..., description="Type of the component output (e.g., 'message', 'data', 'tool', 'text')") + component_id: str + status: JobStatus + content: Any | None = None + 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.""" + + background: bool = False + 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" + ) + + model_config = ConfigDict( + json_schema_extra={ + "examples": [ + { + "background": False, + "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"}, + }, + }, + { + "background": True, + "stream": False, + "flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b", + "inputs": {"global": {"input_value": "Process this in the background", "input_type": "text"}}, + }, + { + "background": False, + "stream": True, + "flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b", + "inputs": {"chat_component": {"text": "Stream this conversation"}}, + }, + ] + }, + extra="forbid", + ) + + +class WorkflowExecutionResponse(BaseModel): + """Synchronous workflow execution response.""" + + flow_id: str + job_id: str + object: Literal["response"] = "response" + created_timestamp: str + status: JobStatus + errors: list[ErrorDetail] = [] + inputs: dict[str, Any] = {} + outputs: dict[str, ComponentOutput] = {} + metadata: dict[str, Any] = {} + + +class WorkflowJobResponse(BaseModel): + """Background job response.""" + + job_id: str + created_timestamp: str + status: JobStatus + errors: list[ErrorDetail] = [] + + +class WorkflowStreamEvent(BaseModel): + """Streaming event response.""" + + type: str + run_id: str + timestamp: int + raw_event: dict[str, Any] + + +class WorkflowStopRequest(BaseModel): + """Request schema for stopping workflow.""" + + job_id: str + force: bool = Field(default=False, description="Force stop the workflow") + + +class WorkflowStopResponse(BaseModel): + """Response schema for stopping workflow.""" + + job_id: str + status: Literal["stopped", "stopping", "not_found", "error"] + message: str + + +# OpenAPI response definitions +WORKFLOW_EXECUTION_RESPONSES = { + 200: { + "description": "Workflow execution response", + "content": { + "application/json": { + "schema": { + "oneOf": [ + WorkflowExecutionResponse.model_json_schema(), + WorkflowJobResponse.model_json_schema(), + ] + } + }, + "text/event-stream": { + "schema": WorkflowStreamEvent.model_json_schema(), + "description": "Server-sent events for streaming execution", + }, + }, + } +} + +WORKFLOW_STATUS_RESPONSES = { + 200: { + "description": "Workflow status response", + "content": { + "application/json": {"schema": WorkflowExecutionResponse.model_json_schema()}, + "text/event-stream": { + "schema": WorkflowStreamEvent.model_json_schema(), + "description": "Server-sent events for streaming status", + }, + }, + } +} diff --git a/src/lfx/src/lfx/services/settings/base.py b/src/lfx/src/lfx/services/settings/base.py index 51346d3b00..6e0f5f90f8 100644 --- a/src/lfx/src/lfx/services/settings/base.py +++ b/src/lfx/src/lfx/services/settings/base.py @@ -313,6 +313,10 @@ class Settings(BaseSettings): """If set to True, Langflow will start the agentic MCP server that provides tools for flow/component operations, template search, and graph visualization.""" + # Developer API + developer_api_enabled: bool = False + """If set to True, Langflow will enable developer API endpoints for advanced debugging and introspection.""" + # Public Flow Settings public_flow_cleanup_interval: int = Field(default=3600, gt=600) """The interval in seconds at which public temporary flows will be cleaned up.