mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-23 23:13:58 +08:00
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.
This commit is contained in:
ogabrielluiz
@ -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"
|
||||
}
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
|
||||
301
src/lfx/src/lfx/cli/serve_workflow.py
Normal file
301
src/lfx/src/lfx/cli/serve_workflow.py
Normal file
@ -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)
|
||||
130
src/lfx/tests/unit/cli/test_serve_workflow.py
Normal file
130
src/lfx/tests/unit/cli/test_serve_workflow.py
Normal file
@ -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
|
||||
Reference in New Issue
Block a user