# Subflows in Kestra – Modularize and Reuse Flows Subflows let you build **modular** and **reusable** workflow components. They work like function calls: executing a subflow creates a new flow run from within another flow.
## Why use a subflow? Subflows allow you to build modular and reusable components that you can use across multiple flows. For example, you might define a subflow that handles error alerts by posting to Slack and email. By using a Subflow, you can reuse these two tasks together for all flows that you want to send error notifications, instead of having to copy the individual tasks for every flow. :::alert{type="warning"} Recursive flows are not supported. Kestra doesn’t allow a flow to call itself (directly or indirectly). Any cycle **(flowA→flowA)** makes the flow invalid. Recursive execution can create infinite loops and unbounded fan-out. **Do instead:** Use **[ForEach](/plugins/core/flow/io.kestra.plugin.core.flow.foreach)** and **[branching flowable](../01.tasks/00.flowable-tasks/index.md)** tasks to iterate or split work without creating cycles (e.g., [LoopUntil](/plugins/core/flow/io.kestra.plugin.core.flow.loopuntil)). ::: ## How to declare a subflow To call a flow from another flow, use the `io.kestra.plugin.core.flow.Subflow` task, and in that task, specify the `flowId` and `namespace` of the subflow that you want to execute. You can also specify custom `inputs`, similar to passing arguments to a function. The optional properties `wait` and `transmitFailed` control the execution behavior. By default, if `wait` is omitted or set to `false`, the parent flow continues without waiting for the subflow to finish. The `transmitFailed` property determines whether a failure in the subflow execution should cause the parent flow to fail. :::alert{type="info"} A Subflow task acts like a trigger to execute the child flow. While not managed like [Triggers](../07.triggers/index.mdx) in the UI, it is conceptually similar. ::: ## Practical example A subflow can encapsulate critical business logic, making it reusable across flows and easier to test in isolation. Here is a simple example of a subflow: ```yaml id: critical_service namespace: company.team tasks: - id: return_data type: io.kestra.plugin.jdbc.duckdb.Query sql: | INSTALL httpfs; LOAD httpfs; SELECT sum(total) as total, avg(quantity) as avg_quantity FROM read_csv_auto('https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv', header=True); store: true outputs: - id: some_output type: STRING value: "{{ outputs.return_data.uri }}" ``` In this example, `return_data` outputs `uri` of the query output. That URI is a reference to the internal storage location of the stored file. This output can be used in the parent flow to perform further processing. ```yaml id: parent_service namespace: company.team tasks: - id: subflow_call type: io.kestra.plugin.core.flow.Subflow namespace: company.team flowId: critical_service wait: true transmitFailed: true - id: log_subflow_output type: io.kestra.plugin.scripts.shell.Commands taskRunner: type: io.kestra.plugin.core.runner.Process commands: - cat "{{ outputs.subflow_call.outputs.some_output }}" ``` The `outputs` map task IDs to their results. Here, the parent flow accesses the `some_output` value from the `subflow_call` task. ## Subflow properties Below is a full list of all properties of the `io.kestra.plugin.core.flow.Subflow` task. You don’t need to memorize all properties — the task documentation always lists them. | Field | Description | |------------------------|-----------------------------------------------------------------------------| | `flowId` | The subflow's identifier | | `namespace` | The namespace where the subflow is located | | `inheritLabels` | Determines if the subflow inherits labels from the parent (default: false). | | `inputs` | Inputs passed to the subflow | | `labels` | Labels assigned to the subflow | | `outputs` (deprecated) | Allows passing outputs from the subflow execution to the parent flow. | | `revision` | The subflow revision to execute (defaults to the latest) | | `scheduleDate` | Schedule subflow execution on a specific date rather than immediately. | | `transmitFailed` | If true, parent flow fails on subflow failure (requires `wait` to be true). | | `wait` | If true, parent flow waits for subflow completion (default: true). | ## Passing data between parent and child flows Flows can emit outputs that can be accessed by the parent flow. Using the `io.kestra.plugin.core.flow.Subflow` task you can call any flow as a subflow and access its outputs in downstream tasks. For more details and examples, check the [Outputs page](../06.outputs/index.md#pass-data-between-flows-using-flow-outputs). ### Accessing Outputs from a subflow execution Outputs include the execution ID, extracted outputs, and the final state (if `wait` is true). Subflows improve maintainability of complex workflows. Use them to build modular, reusable components that can be shared across namespaces, projects, and teams. Here’s an example of a subflow with explicitly defined outputs. ```yaml id: flow_outputs namespace: company.team tasks: - id: mytask type: io.kestra.plugin.core.debug.Return format: this is a task output used as a final flow output outputs: - id: final type: STRING value: "{{ outputs.mytask.value }}" ``` We can access these outputs from a parent task as seen in the example below: ```yaml id: parent_flow namespace: company.team tasks: - id: subflow type: io.kestra.plugin.core.flow.Subflow flowId: flow_outputs namespace: company.team wait: true - id: log_subflow_output type: io.kestra.plugin.core.log.Log message: "{{ outputs.subflow.outputs.final }}" ``` For more details, see the [sublow outputs documentation](../../11.migration-guide/v0.15.0/subflow-outputs/index.md). ### Passing inputs to a subflow You can pass inputs to a Subflow task. The example below passes two inputs to a subflow. Subflow: ```yaml id: subflow_example namespace: company.team inputs: - id: http_uri type: STRING tasks: - id: download type: io.kestra.plugin.core.http.Request uri: "{{ inputs.http_uri }}" - id: log type: io.kestra.plugin.core.log.Log message: "{{ outputs.download.body }}" outputs: - id: data type: STRING value: "{{ outputs.download.body }}" ``` Parent flow: ```yaml id: inputs_subflow namespace: company.team inputs: - id: url type: STRING tasks: - id: subflow type: io.kestra.plugin.core.flow.Subflow flowId: subflow_example namespace: company.team inputs: http_uri: "{{ inputs.url }}" wait: true - id: hello type: io.kestra.plugin.core.log.Log message: "{{ outputs.subflow.outputs.data }}" ``` In this example, the parent flow successfully passes an input to the subflow. #### Nested inputs In the example below, the flow extracts JSON data from a REST API and passes it to a subflow as a nested input: ```yaml id: extract_json namespace: company.team tasks: - id: api type: io.kestra.plugin.core.http.Request uri: https://dummyjson.com/users - id: read_json type: io.kestra.plugin.core.log.Log message: "{{ outputs.api.body }}" - id: subflow type: io.kestra.plugin.core.flow.Subflow namespace: company.team flowId: subflow inputs: users.firstName: "{{ outputs.api.body | jq('.users') | first | first | jq('.firstName') | first }}" users.lastName: "{{ outputs.api.body | jq('.users') | first | first | jq('.lastName') | first }}" wait: true transmitFailed: true ``` To provide type validation to extracted JSON fields, you can use [nested inputs](../05.inputs/index.md#nested-inputs) in the subflow definition: ```yaml id: subflow namespace: company.team inputs: - id: users.firstName type: STRING defaults: Rick - id: users.lastName type: STRING defaults: Astley tasks: - id: process_user_data type: io.kestra.plugin.core.log.Log message: hello {{ inputs.users }} ``` You can then pass the entire `users` object, including nested fields, to any task in the subflow.