Workflow Execution API

Relevant source files

• api/controllers/common/controller_schemas.py
• api/controllers/console/explore/audio.py
• api/controllers/console/explore/completion.py
• api/controllers/console/explore/conversation.py
• api/controllers/console/explore/message.py
• api/controllers/console/explore/saved_message.py
• api/controllers/console/explore/workflow.py
• api/controllers/service_api/app/audio.py
• api/controllers/service_api/app/completion.py
• api/controllers/service_api/app/conversation.py
• api/controllers/service_api/app/message.py
• api/controllers/service_api/app/workflow.py
• api/controllers/web/audio.py
• api/controllers/web/completion.py
• api/controllers/web/conversation.py
• api/controllers/web/message.py
• api/controllers/web/saved_message.py
• api/controllers/web/workflow.py
• api/core/app/apps/advanced_chat/app_runner.py
• api/core/app/apps/advanced_chat/generate_task_pipeline.py
• api/core/app/apps/pipeline/pipeline_runner.py
• api/core/app/apps/workflow/app_runner.py
• api/core/app/apps/workflow/generate_task_pipeline.py
• api/core/app/apps/workflow_app_runner.py
• api/core/app/entities/queue_entities.py
• api/core/app/entities/task_entities.py
• api/core/app/task_pipeline/based_generate_task_pipeline.py
• api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py
• api/services/audio_service.py
• api/services/conversation_service.py
• api/services/web_conversation_service.py
• api/tests/unit_tests/controllers/console/explore/test_completion.py
• api/tests/unit_tests/controllers/service_api/app/test_completion.py
• api/tests/unit_tests/controllers/web/test_pydantic_models.py
• api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_conversation_variables.py
• api/tests/unit_tests/core/app/apps/advanced_chat/test_app_runner_input_moderation.py
• api/tests/unit_tests/core/app/apps/agent_chat/test_agent_chat_app_runner.py
• api/tests/unit_tests/core/app/apps/chat/test_app_generator_and_runner.py
• api/tests/unit_tests/core/app/apps/completion/test_app_runner.py
• api/tests/unit_tests/core/app/apps/pipeline/test_pipeline_runner.py
• web/app/components/develop/__tests__/workflow-version-api-upgrade-notice.spec.tsx
• web/app/components/develop/md.tsx
• web/app/components/develop/template/__tests__/template_advanced_chat.spec.ts
• web/app/components/develop/template/__tests__/template_workflow.spec.ts
• web/app/components/develop/workflow-version-api-upgrade-notice.tsx

Purpose and Scope

This page documents the Workflow Execution API accessible via the POST /workflows/run endpoint. This API enables programmatic invocation of Workflow applications in Dify, designed as stateless, single-run graphs typically used for processing, content generation, or transformation tasks.

We detail request formats, input variable handling including multi-modal files, streaming event generation, and execution tracing for observability and debugging.

---

Workflow Execution Endpoint

Endpoint Overview

• Method: POST
• Path: /workflows/run
• Controller: Implemented in WorkflowApi class in api/controllers/service_api/app/workflow.py157-180

Characteristics

• Stateless and transient runs, no conversation IDs involved.
• Requires input variables matching workflow DSL definitions.
• Supports synchronous (blocking) and asynchronous (streaming) response modes.
• Integrates distributed tracing via trace_session_id for observability.

Sources: api/controllers/service_api/app/workflow.py70-82 api/controllers/service_api/app/workflow.py157-180

---

Request Structure

API Input Parameters

The request accepts a JSON payload modeled by WorkflowRunPayload comprising:

Parameter	Type	Description
inputs	dict	Key-value pairs defining variable inputs to the workflow
response_mode	string enum	"blocking" or "streaming", controls response streaming mode
user	string	User identifier (e.g., end user ID)
files	list of dicts	Optional global file input array, for multi-modal support
trace_session_id	string or null	Optional external trace session id for observability

These inputs flow into the AppGenerateService which launches the workflow execution using WorkflowAppGenerateTaskPipeline.

Request Input Validation and Handling

• The API validates inputs against the workflow's variable schema.
• Workflow's graph runtime state initializes variable pools accordingly.
• The files argument is used for multi-modal inputs such as documents or images, each file object requires:
  • type: one of document, image, audio, video, or custom.
  • transfer_method: either remote_url or local_file.
  • For remote_url: url field is mandatory.
  • For local_file: upload_file_id referencing a previously uploaded file.

This file input validation is implemented in service API completion request validation api/controllers/service_api/app/completion.py75-82 and workflow runtime initialization api/core/workflow/workflow_entry.py163-178

---

Workflow Execution Request Flow

Sources: api/controllers/service_api/app/workflow.py70-82 api/core/app/apps/workflow/generate_task_pipeline.py77-131 api/services/app_generate_service.py48-60

---

Response Modes

1. Blocking Mode

• The API call waits for full workflow completion.
• Final workflow run outputs are returned in JSON payload.
• Includes workflow_run_id, execution status, and output variables as a dictionary.
• Internally, the workflow engine executes synchronously and the result is constructed from WorkflowRun model data.

2. Streaming Mode

• The API returns an SSE (Server-Sent Events) stream.
• Events are emitted in real-time to the client, describing progressive workflow phases and node executions.
• Events include workflow_started, node_started, node_finished, and workflow_finished.
• This streaming mechanism leverages AppQueueManager for event dispatching internally.

---

Workflow Streaming Sequence Diagram

Sources: api/controllers/service_api/app/workflow.py40 api/controllers/service_api/app/workflow.py70-82 api/core/app/apps/workflow/generate_task_pipeline.py77-131

---

Execution Events and Tracing

Key Workflow Execution Event Types

Event Name	Description	Typical Fields / Data	Code Entity References
workflow_started	Workflow execution initiated	workflow_run_id, workflow_id, inputs, created_at	WorkflowRun (models/workflow.py), QueueWorkflowStartedEvent
node_started	A workflow node starts execution	node_id, node_type, inputs	GraphNodeEventBase, QueueNodeStartedEvent
node_finished	Node execution completed	outputs, elapsed_time, status	QueueNodeSucceededEvent
workflow_paused	Workflow pauses (e.g., for user input)	Cleared outputs, paused nodes, pause reasons	WorkflowPauseStreamResponse._clear_paused_outputs
workflow_finished	Workflow run completes	status, outputs, total_tokens, elapsed_time	WorkflowRunResponse
node_failed	Node execution failure event	Exception info	QueueNodeFailedEvent

The events are serialized as specific stream responses derived from StreamResponse, such as WorkflowStartStreamResponse, WorkflowFinishStreamResponse, and others in api/core/app/entities/task_entities.py

Execution Logs and Querying

• Workflow executions are recorded as WorkflowRun database entries.
• Logs can be queried with filters such as status, keywords, creation time, and user session via the WorkflowLogQuery.
• The logs provide insights into step counts, tokens used, errors, and timestamps.

Sources: api/controllers/service_api/app/workflow.py83-106 api/core/workflow/workflow_entry.py38 api/core/app/entities/task_entities.py51-300

---

Distributed Tracing and Observability

Trace Propagation

Dify supports distributed tracing for workflow executions by accepting an external trace session ID to correlate spans across services.

• The API extracts trace_session_id from the run payload (WorkflowRunPayload) api/controllers/service_api/app/workflow.py78-80
• Traces are propagated down through internal execution layers, including AppGenerateService and into GraphEngine components.
• The OpsTraceManager and WorkflowAppRunnerHandler attach tracing spans to workflow execution operations api/core/workflow/workflow_entry.py12

Trace Propagation Diagram

Sources: api/controllers/service_api/app/workflow.py47 api/controllers/service_api/app/workflow.py78-80 api/core/workflow/workflow_entry.py10-30

---

Implementation Highlights

Key Classes and Functions

Component	Role and Responsibility	File Reference
WorkflowApi	REST endpoint handler for /workflows/run, validation, authentication, streaming response setup	api/controllers/service_api/app/workflow.py150-183
AppGenerateService	Business logic to run workflow, load workflow models, manage workflow runs and users	api/services/app_generate_service.py30-80 (referenced)
WorkflowAppGenerateTaskPipeline	Task pipeline coordinating graph execution, state, streaming event generation	api/core/app/apps/workflow/generate_task_pipeline.py77-190
AppQueueManager	Manages event sending, queueing, and streaming to clients	api/core/app/apps/base_app_queue_manager.py (general)
WorkflowBasedAppRunner	Core graph execution runner, linking graph engine to the queue manager	api/core/app/apps/workflow_app_runner.py90-150
GraphEngine	Executes workflow graph nodes, handles node events and lifecycle	(Part of graphon.graph_engine)
OpsTraceManager	Tracing provider integration, manages spans	core/ops/ops_trace_manager.py

---

Error Handling

The workflow execution API can return various HTTP errors mapped from backend exceptions:

Exception	HTTP Status	Description	Source
WorkflowNotFoundError	404	Workflow ID does not exist	api/services/errors/app.py
IsDraftWorkflowError	400	Trying to run an unpublished draft	api/services/errors/app.py
InvokeRateLimitError	429	Too many concurrent requests for the app	api/services/errors/llm.py
ProviderQuotaExceededError	400	Model provider quota exhausted	api/controllers/service_api/app/error.py
AppUnavailableError	400	App is disabled or misconfigured	api/controllers/service_api/app/error.py

Sources: api/controllers/service_api/app/workflow.py25-32 api/services/errors/app.py

---

Summary

The Workflow Execution API enables clients to invoke published workflow graphs as single stateless executions with controlled inputs. Detailed streaming responses provide live insight into execution progress, leveraging event queue management. Built-in distributed tracing supports observability, connecting API requests to backend execution spans. The backend architecture separates REST routing, service logic, execution pipelines, runtime graph engines, and event dispatching layers to deliver robust workflow runs.

This API is essential for users integrating Dify workflows into external systems and for building custom UI or automation layers atop Dify's workflow engine.

---

References and Sources

• api/controllers/service_api/app/workflow.py:1-180
• api/core/app/apps/workflow/generate_task_pipeline.py:77-190
• api/core/app/apps/workflow_app_runner.py:90-150
• api/core/app/apps/base_app_queue_manager.py
• api/core/app/entities/task_entities.py
• api/services/app_generate_service.py:30-80
• api/core/workflow/workflow_entry.py:163-178
• api/controllers/service_api/app/completion.py:75-82
• core/ops/ops_trace_manager.py
• api/controllers/service_api/app/error.py