When a flow's folder requires apikey auth, the JSON-RPC route now requires a valid
langflow API key in the x-api-key header whose owner is the flow owner, before any
dispatch (covers message/send, message/stream, tasks/get, tasks/resubscribe). Flows
in none folders stay public (the A2A public-agent model); the discovery card stays
public by spec.
Uses check_key directly rather than api_key_security, which under AUTO_LOGIN returns
the superuser for a missing key and would bypass the gate. The key is scoped to
flow.user_id since the flow runs as its owner. The auth_type read is extracted into a
shared folder_auth_type helper so card-advertise and route-enforce can't drift.
Advertises the streaming capability on both the handler card (the SDK's @validate
gate) and the discovery card, which mounts message/stream and tasks/resubscribe
through the existing dispatcher. message/stream replays the run lifecycle
(submitted -> working -> artifact -> completed) as SSE frames by re-emitting the
same executor events; tasks/resubscribe re-attaches to a live run and returns a
spec error for terminal/cross-worker tasks (tasks/get covers terminal reads).
Per-token deltas and durable cross-worker re-attach are follow-ups. Tests cover
the advertised capability, the SSE lifecycle, and that resubscribe to a finished
task ends with an error frame rather than hanging.
Replaces the in-memory module-singleton task store with DurableTaskStore, a
DB-backed a2a-sdk TaskStore that persists each protobuf Task as a JSON blob in a
new a2a_tasks table (composite PK (id, owner), JSONB on postgres). Tasks now
survive a restart and are visible across workers; save() uses a writing session,
get() a read-only one.
Adds the A2ATask model + alembic migration, and tests covering save/get
round-trip, owner scoping, and a real_services test that runs the migration and
the store on both sqlite and postgres.
Adds POST /api/v1/a2a/{flow_id}/jsonrpc serving the A2A JSON-RPC surface:
message/send runs the flow through the v2 sync surface and returns a
terminal Task, and tasks/get reads it back from an in-memory store. Gated
behind LANGFLOW_A2A_ENABLED and the per-flow agent type + a2a_enabled flag,
like the card route.
The protocol-pure executor (a2a_executor.py) holds no langflow imports and
reaches the flow via an injected run_flow callable, so the protocol layer
stays extractable to lfx. Unexpected build/run failures are logged
server-side and surface a generic message to the unauthenticated caller;
multi-output flows emit each text channel instead of dropping them.
Public GET /api/v1/a2a/{flow_id}/.well-known/agent-card.json serving a
spec-valid A2A agent card for agent-typed, a2a_enabled flows. Gated by
LANGFLOW_A2A_ENABLED (default off) via a per-request guard; the router is
mounted unconditionally so the flag can flip at runtime. Reflects folder
apikey auth onto securitySchemes (reflect-only; F6 enforces). The card
degrades to an empty input schema on unbuildable flows and ignores
malformed a2a_card_overrides instead of 500ing the public endpoint.
Adds a2a-sdk as a direct langflow-base dependency.
cz/hitl-v2 regressed the release-1.11.0 fix: get_type_dict called the
async get_all_types_dict synchronously, so the sync _build_dict received
an un-awaited coroutine and the first Graph.from_payload raised
"'coroutine' object is not a mapping" — breaking json_schema_from_flow
and MCP tool-schema generation. Restore the sync build_custom_components
path (identical to release-1.11.0) and its regression test.
Add an explicit flow_type enum (workflow|agent), a2a_enabled, and
a2a_card_overrides to the Flow model via an additive migration following
the access_type precedent. Surface the fields in the flow create/read/update
schemas, add a flow_type filter to the list endpoint, and add a
suggest_flow_type auto-detect helper (UI suggestion only, never the stored
source of truth).
This is F1, the foundation of the A2A protocol support epic.
Adopt the lfx.workflow shared contract layer: rewire the background
execution service and the v2 routes to import adapters/converters/
agui_translator from lfx.workflow, drop the duplicated backend copies and
the superseded inline workflow_background/execution/validation split.
Reconcile the adapter/converter divergence: keep the shared layer's
cancel_events (distinct cancelled terminal, not error) and valid-derived
component status, and port the background path's is_durable + durable-event
sets and workflow_response_from_output_events.
Moves the protocol-agnostic pieces of the v2 workflows API out of the langflow
backend into lfx so both the backend and `lfx serve` can share one contract.
First step toward giving lfx (the production runtime) the v2 workflows API.
- Move api/v2/adapters/, agui_translator.py, and converters.py to lfx/workflow/.
They depend only on lfx.schema.workflow and ag_ui (already an lfx dep), so lfx
carries the contract with zero langflow imports.
- Decouple the one langflow reference: converters typed run_response against
langflow.api.v1.schemas.RunResponse (TYPE_CHECKING only). Replaced with a local
RunResponseLike Protocol (outputs + session_id), the only attributes used.
- Repoint the six backend v2 workflow modules to import from lfx.workflow.
- Move the five protocol-agnostic contract tests into src/lfx/tests/unit/workflow/
(run in the lfx-only env). test_output_event_parity and test_workflow_agui stay
in langflow (they need langflow.api.build) with repointed imports.
Coverage unchanged: 201 contract tests pass in the lfx-only env, 191 backend v2
tests pass; 392 total, same as before the move.
Splits the ~1.5k-line workflow.py into focused modules and folds in the
execution-timeout and error-sanitization fixes from Cristhianzl's review of #13307.
- B1: workflow.py now holds only the four route handlers. Validation guards move
to workflow_validation, the sync/stream run loop to workflow_execution, and the
durable background machinery to workflow_background (layered, acyclic).
- I1: add workflow_execution_timeout (default 300) and apply a single wall-clock
ceiling across sync, stream, background, and public via _stream_event_frames. A
timeout becomes a sanitized terminal error and marks a background job failed.
- I3: the route error handlers no longer echo raw exception text. They return a
generic, code-tagged message and log the full exception server-side.
- R1: remove the "commented out / future scope" comments that sat over live
dataframe-extraction code in converters.py.
- R4: drop the worker-routing internals from the reattach 409 message.
Tests cover the timeout terminal-error path and the error-body sanitization, and
the settings field-count guard is updated for the new setting.
- recover session_id for completed background jobs from the persisted
terminal message instead of always returning null, so GET status can
continue the same chat/memory thread
- replay a user-cancel as a CUSTOM cancel marker + RUN_FINISHED (agui) and
a `cancelled` terminal (langflow) instead of RUN_ERROR, so a re-attaching
client no longer reads a deliberate stop as a failure
- cancel the evicted still-running buffer writer when the background-run
registry is full, so it stops appending into a run no reader can find
- derive per-component status from the error artifact / valid flag instead
of hardcoding COMPLETED, and stop the langflow adapter dropping `valid`
- throttle the unauthenticated public endpoint per IP and bound its
input_value/session_id length
- document the sync-only scope of request-body globals
- document that live event re-attach is intentionally owner-only
A completed background run's GET status returned a bare COMPLETED with an
empty `outputs` and a null `output`. The COMPLETED branch reconstructs
from `vertex_builds` keyed by job_id, which the durable path does not
write, so reconstruction raised ValueError and fell through to an empty
response; the result was only retrievable via a /events re-attach.
The runner now captures the terminal `output` events (the langflow
adapter's normalized ComponentOutput payloads) into `Job.result`, and the
status COMPLETED branch rebuilds the `outputs` map and resolved `output`
from them via `workflow_response_from_output_events`, matching the sync
response. agui-protocol runs emit no `output` events, so their status
stays result-less (the result remains on the /events log).
The v2 workflows endpoint applied `tweaks` only on mode=sync. The stream
and background paths build the graph via the v1 build-vertex loop
(`generate_flow_events`), which never received the tweaks, so they were
silently dropped. The confusing symptom: a model passed via tweaks
surfaced as "A model selection is required", and any per-component
override was ignored on non-sync runs.
Thread `parsed.tweaks` into `generate_flow_events` and apply them to the
built graph via `vertex.update_raw_params`. We do not use the lfx
`process_tweaks_on_graph` helper because it only sets `vertex.params`,
which does not persist to runtime (the same bug
`lfx.base.tools.run_flow._process_tweaks_on_graph` works around). No-tweaks
runs are unchanged (guarded by `if tweaks`). Adds a streaming regression
test that overrides ChatInput via tweaks and asserts the value drives the run.
The durable background-execution rewrite of workflow.py reverted
`side_channel_events` to its pre-"end" form, dropping the "end" event
from the AG-UI side-channel. That event carries `build_duration` to the
playground chat-view, and the message metadata badge only renders when
`hasDuration || hasTokens`. With build_duration gone the badge vanished,
failing the token-usage and shareable-playground "Finished In" regression
tests. Re-add "end" so the streaming playground path delivers it again.
The durable background-execution work added six Settings fields
(background_max_concurrency, background_job_timeout, background_lease_ttl_s,
background_heartbeat_interval_s, background_watchdog_interval_s, test_redis_url)
without updating EXPECTED_FIELDS, so test_field_count_unchanged failed 152 != 146.
These are intentional bg-exec config; add them to the gate.
As a module-level runtime value, FrameSourceFactory = Callable[..., Any] is a
GenericAlias that passes isinstance(obj, type) but makes issubclass(obj, Service)
raise on Python 3.10/3.14. The service factory scans this module for Service
subclasses (services/factory.py:90), so the runtime alias crashed service
initialization on those interpreters with 'issubclass() arg 1 must be a class'
-> 'Could not initialize services', erroring out dozens of unrelated tests at
setup (3.13 was unaffected, which is why local runs passed).
The alias is only referenced in a lazy annotation (from __future__ import
annotations), so moving it + the Callable import under TYPE_CHECKING removes it
from the runtime namespace with no behavior change.
Verified: alias absent from runtime module namespace, factory scan finds
BackgroundExecutionService cleanly, durable service tests 26/26 pass.
