Event Streaming

Relevant source files

• docs/versioned_docs/version-1.8.0/Develop/environment-variables.mdx
• src/backend/base/langflow/api/build.py
• src/backend/base/langflow/api/v1/chat.py
• src/backend/base/langflow/preload.py
• src/backend/base/langflow/services/cache/base.py
• src/backend/base/langflow/services/cache/factory.py
• src/backend/base/langflow/services/cache/service.py
• src/backend/base/langflow/services/chat/service.py
• src/backend/base/langflow/services/flow_events/factory.py
• src/backend/base/langflow/services/flow_events/service.py
• src/backend/base/langflow/services/job_queue/factory.py
• src/backend/base/langflow/services/job_queue/service.py
• src/backend/base/langflow/services/session/service.py
• src/backend/base/langflow/services/telemetry/opentelemetry.py
• src/backend/tests/unit/base/test_preload.py
• src/backend/tests/unit/services/cache/test_redis_cache.py
• src/backend/tests/unit/services/test_cache_factory.py
• src/backend/tests/unit/test_chat_endpoint.py
• src/backend/tests/unit/test_flow_events_service.py
• src/backend/tests/unit/test_redis_job_queue_service.py
• src/backend/tests/unit/test_telemetry.py
• src/lfx/src/lfx/graph/checkpoint/builder.py
• src/lfx/src/lfx/graph/checkpoint/resume.py
• src/lfx/src/lfx/graph/checkpoint/schema.py
• src/lfx/tests/unit/graph/checkpoint/test_resume_round_trip.py

This document describes the event streaming mechanisms that deliver real-time build progress and execution updates from the Langflow backend to clients. The system utilizes Server-Sent Events (SSE) and specialized API endpoints to decouple graph execution from UI updates, ensuring a responsive user experience during complex AI workflow processing.

---

Overview

Event streaming in Langflow allows for the asynchronous delivery of component build statuses, logs, and final results. When a flow is executed, the backend produces a stream of events that the frontend consumes to update the visual graph state in real-time.

• Asynchronous Execution: Flow builds are initiated via POST requests that return a job_id, allowing the client to subscribe to a dedicated event stream src/backend/base/langflow/api/v1/chat.py175-185
• Server-Sent Events (SSE): The primary transport for real-time updates, providing a unidirectional push from server to client via StreamingResponse src/backend/base/langflow/api/v1/chat.py10
• Heartbeat Mechanism: Periodic keep-alive signals ensure connections remain active through proxies and load balancers src/backend/base/langflow/api/build.py55-58
• Token Streaming: Specialized support for streaming LLM tokens directly to the chat interface during execution, integrated via the EventManager src/backend/base/langflow/api/build.py34

---

Event Architecture and Code Entities

The streaming system bridges the gap between the graph execution engine and the HTTP transport layer. The following diagram maps the logical streaming concepts to their implementation in the codebase.

Title: Event Streaming Architecture

Sources: src/backend/base/langflow/api/v1/chat.py24 src/backend/base/langflow/api/build.py18 src/backend/base/langflow/api/build.py34 src/backend/base/langflow/services/job_queue/service.py87

---

Backend Implementation

The backend leverages FastAPI's StreamingResponse (specifically a custom DisconnectHandlerStreamingResponse) to implement the SSE protocol src/backend/base/langflow/api/disconnect.py1-20

Build and Event Flow

When a user triggers a build, the system follows a two-step process:

1. Initiation: The /api/v1/build/{flow_id}/flow endpoint validates the graph and starts a background task via the JobQueueService src/backend/base/langflow/api/v1/chat.py171-185 This is handled by start_flow_build, which generates a unique job_id and registers the build's owner src/backend/base/langflow/api/v1/chat.py95-103
2. Subscription: The client then calls /api/v1/build/{job_id}/events (handled by get_flow_events_response) to receive the SSE stream src/backend/base/langflow/api/build.py19-24

SSE Heartbeat and Timeouts

To prevent connection drops, the backend implements a heartbeat mechanism. The STREAMING_ACTIVITY_REFRESH_S interval (10 seconds) is used to refresh a polling-watchdog activity key, signaling that the client is still alive src/backend/base/langflow/api/build.py55-58

Data Projection

To maintain backward compatibility, the system includes _project_event_to_v1, which renders add_message events into a shape compatible with release 1.11.0 clients, ensuring that new content block schemas do not break legacy consumers src/backend/base/langflow/api/build.py71-96

---

Data Flow: From Vertex to Event

As the Graph executes, each Vertex (component) generates updates. The EventManager captures these updates and formats them for the stream.

Title: Vertex Execution to Client Event Flow

Sources: src/backend/base/langflow/api/build.py11-12 src/backend/base/langflow/api/build.py34 src/backend/base/langflow/api/build.py55-58 src/backend/base/langflow/services/job_queue/service.py1-10

---

Component Execution and Metadata

During the build process, the system generates authoritative metadata for streamed events.

• Output Metadata: The _output_meta_for_vertex function provides per-output metadata (display names, types, terminal status) for the v2 output stream event, ensuring the stream matches sync execution outputs src/backend/base/langflow/api/build.py99-122
• Checkpoint Restoration: When resuming a flow, _rerun_non_input_predecessors identifies producers whose live outputs (like model clients or Tools) were dropped during serialization and marks them to re-run src/backend/base/langflow/api/build.py125-157
• Telemetry: Component input values are logged via _log_component_input_telemetry if the vertex contains a custom_component src/backend/base/langflow/api/build.py158-171

---

Delivery Modes

Langflow supports multiple event delivery strategies, configurable via EventDeliveryType src/backend/base/langflow/api/utils.py21

Mode	Description	Implementation Detail
STREAMING	Real-time push via SSE	Uses AsyncIterator to yield events from the queue src/backend/base/langflow/api/build.py6
POLLING	Client periodically fetches new events	Managed by the JobQueueService for retrieving job state src/backend/base/langflow/services/job_queue/service.py87-128

---

Error Handling and Security

Streaming requests are subject to strict ownership and validation checks.

• Job Ownership: The _verify_job_ownership function raises a 404 if a user attempts to access a job they do not own, protecting execution logs src/backend/base/langflow/api/v1/chat.py76-92
• Job Registration: _register_job_owner_or_cancel ensures that if the Redis-backed queue is unreachable during job registration, the build is cancelled best-effort to avoid orphaned tasks src/backend/base/langflow/api/v1/chat.py95-111
• Preflight Validation: The validate_flow_for_current_settings check ensures the flow is valid for the current environment before the build starts src/backend/base/langflow/api/v1/chat.py19

Sources: src/backend/base/langflow/api/v1/chat.py76-111 src/backend/base/langflow/api/build.py8-9