---
description: "Complete reference for Conductor workflow definitions — properties, task configurations, input expressions, failure workflows, and timeout policies."
---

# Workflow Definition

The Workflow Definition contains all the information necessary to define the behavior of a workflow. The most important part of this definition is the `tasks` property, which is an array of [**Task Configurations**](#task-configurations).

For the formal JSON Schema definitions of workflow and task structures, see the [`schemas/`](https://github.com/conductor-oss/conductor/tree/main/schemas) directory in the repository.


## Workflow Properties
| Field                         | Type                             | Description                                                                                                                                                                  | Notes                                                                                             |
|:------------------------------|:---------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------| :------------------------------------------------------------------------------------------------ |
| name                          | string                           | Name of the workflow                                                                                                                                                         |                                                                                                   |
| description                   | string                           | Description of the workflow                                                                                                                                                  | Optional                                                                                          |
| version                       | number                           | Numeric field used to identify the version of the schema. Use incrementing numbers.                                                                                          | When starting a workflow execution, if not specified, the definition with highest version is used |
| tasks                         | array of object(s)               | An array of task configurations. [Details](#task-configurations)                                                                                                             |                                                                                                   |
| inputParameters               | array of string(s)               | List of input parameters. Used for documenting the required inputs to workflow                                                                                               | Optional.                                                                                         |
| outputParameters              | object                           | JSON template used to generate the output of the workflow                                                                                                                    | If not specified, the output is defined as the output of the _last_ executed task                 |
| inputTemplate                 | object                           | Default input values. See [Using inputTemplate](#default-input-with-inputtemplate)                                                                                           | Optional.                                                                                         |
| failureWorkflow               | string                           | Workflow to be run on current Workflow failure. Useful for cleanup or post actions on failure. [Explanation](#failure-workflow)                                              | Optional.                                                                                         |
| failureWorkflowVersion        | number                           | When `failureWorkflow` parameter is specified, sets the _failure workflow version_ to be run on current Workflow failure. If not specified, the latest version will be used. | Optional.                                                                                         |
| schemaVersion                 | number                           | Current Conductor Schema version. schemaVersion 1 is discontinued.                                                                                                           | Must be 2                                                                                         |
| restartable                   | boolean                          | Flag to allow Workflow restarts                                                                                                                                              | Defaults to true                                                                                  |
| workflowStatusListenerEnabled | boolean                          | Enable status callback. [Explanation](#workflow-status-listener)                                                                                                             | Defaults to false                                                                                 |
| ownerEmail                    | string                           | Email address of the team that owns the workflow                                                                                                                             | Required                                                                                          |
| timeoutSeconds                | number                           | The timeout in seconds after which the workflow will be marked as `TIMED_OUT` if it hasn't been moved to a terminal state                                                    | No timeouts if set to 0                                                                           |
| timeoutPolicy                 | string ([enum](#timeout-policy)) | Workflow's timeout policy                                                                                                                                                    | Defaults to `TIME_OUT_WF`                                                                         |

### Failure Workflow

The failure workflow gets the _original failed workflow’s input_ along with 3 additional items,

* `workflowId` - The id of the failed workflow which triggered the failure workflow.
* `reason` - A string containing the reason for workflow failure.
* `failureStatus` - A string status representation of the failed workflow.
* `failureTaskId` - The id of the failed task of the workflow that triggered the failure workflow.

### Timeout Policy

* TIME_OUT_WF: Workflow is marked as TIMED_OUT and terminated
* ALERT_ONLY: Registers a counter (workflow_failure with status tag set to `TIMED_OUT`)

### Workflow Status Listener
Setting the `workflowStatusListenerEnabled` field in your Workflow Definition to `true` enables notifications.

To add a custom implementation of the Workflow Status Listener. Refer to the [Workflow Status Listener extension guide](../../advanced/extend.md#workflow-status-listener).

The listener can be implemented in such a way as to either send a notification to an external system or to send an event on the conductor queue to complete/fail another task in another workflow as described in the [event handlers guide](../eventhandlers.md).

### Default Input with `inputTemplate`

* `inputTemplate` allows you to define default input values, which can optionally be overridden at runtime (when the workflow is invoked).
* Eg: In your Workflow Definition, you can define your inputTemplate as:

```json
"inputTemplate": {
    "url": "https://some_url:7004"
}
```

And `url` would be `https://some_url:7004` if no `url` was provided as input to your workflow.




## Task Configurations

The `tasks` property in a Workflow Definition defines an array of *Task Configurations*. This is the blueprint for the workflow. Task Configurations can reference different types of Tasks.

* Simple Tasks
* System Tasks
* Operators

Note: Task Configuration should not be confused with **Task Definitions**, which are used to register SIMPLE (worker based) tasks.

| Field             | Type    | Description                                                                                                                                    | Notes                                                                 |
| :---------------- | :------ | :--------------------------------------------------------------------------------------------------------------------------------------------- | :-------------------------------------------------------------------- |
| name              | string  | Name of the task. MUST be registered as a Task Type with Conductor before starting workflow                                                    |                                                                       |
| taskReferenceName | string  | Alias used to refer the task within the workflow.  MUST be unique within workflow.                                                             |                                                                       |
| type              | string  | Type of task. SIMPLE for tasks executed by remote workers, or one of the system task types                                                     |                                                                       |
| description       | string  | Description of the task                                                                                                                        | optional                                                              |
| optional          | boolean | true  or false.  When set to true - workflow continues even if the task fails.  The status of the task is reflected as `COMPLETED_WITH_ERRORS` | Defaults to `false`                                                   |
| inputParameters   | object  | JSON template that defines the input given to the task. Only one of `inputParameters` or `inputExpression` can be used in a task.              | See [Using Expressions](#using-expressions) for details |
| inputExpression   | object  | JSONPath expression that defines the input given to the task. Only one of `inputParameters` or `inputExpression` can be used in a task.        | See [Using Expressions](#using-expressions) for details |
| asyncComplete     | boolean | `false` to mark status COMPLETED upon execution; `true` to keep the task IN_PROGRESS and wait for an external event to complete it.            | Defaults to `false`                                                   |
| startDelay        | number  | Time in seconds to wait before making the task available to be polled by a worker.                                                             | Defaults to 0.                                                        |


In addition to these parameters, System Tasks have their own parameters. Check out [System Tasks](systemtasks/index.md) for more information.

### Using Expressions
Each executed task is given an input based on the `inputParameters` template or the `inputExpression` configured in the task configuration. Only one of `inputParameters` or `inputExpression` can be used in a task.

#### inputParameters
`inputParameters` can use JSONPath **expressions** to extract values out of the workflow input and other tasks in the workflow.

For example, workflows are supplied an `input` by the client/caller when a new execution is triggered. The workflow `input` is available via an *expression* of the form `${workflow.input...}`. Likewise, the `input` and `output` data of a previously executed task can also be extracted using an *expression* for use in the `inputParameters` of a subsequent task.

Generally, `inputParameters` can use *expressions* of the following syntax:

> `${SOURCE.input/output.JSONPath}`

| Field        | Description                                                              |
| ------------ | ------------------------------------------------------------------------ |
| SOURCE       | Can be either `"workflow"` or the reference name of any task             |
| input/output | Refers to either the input or output of the source                       |
| JSONPath     | JSON path expression to extract JSON fragment from source's input/output |


!!! note "JSON Path Support"
    Conductor supports [JSONPath](http://goessner.net/articles/JsonPath/) specification and uses the [jayway/JsonPath](https://github.com/jayway/JsonPath) Java implementation.

!!! note "Escaping expressions"
    To escape an expression, prefix it with an extra _$_ character (ex.: ```$${workflow.input...}```).

#### inputExpression

`inputExpression` can be used to select an entire object from the workflow input, or the output of another task. The field supports all [definite](https://github.com/json-path/JsonPath#what-is-returned-when) JSONPath expressions.

The syntax for mapping values in `inputExpression` follows the pattern,

> `SOURCE.input/output.JSONPath`

**NOTE:** The ```inputExpression``` field does not require the expression to be wrapped in `${}`.

See [example](#example-3-inputexpression) below.

## Examples

### Example 1 - A Basic Workflow Definition 

Assume your business logic is to simply to get some shipping information and then do the shipping. You start by
logically partitioning them into two tasks:

 1. *shipping_info* - The first task takes the provided account number, and outputs an address.  
 2. *shipping_task* - The 2nd task takes the address info and generates a shipping label.

We can configure these two tasks in the `tasks` array of our Workflow Definition. Let's assume that ```shipping info``` takes an account number, and returns a name and address.

```json
{
  "name": "mail_a_box",
  "description": "shipping Workflow",
  "version": 1,
  "tasks": [
    {
      "name": "shipping_info",
      "taskReferenceName": "shipping_info_ref",
      "inputParameters": {
        "account": "${workflow.input.accountNumber}"
      },
      "type": "SIMPLE"
    },
    {
      "name": "shipping_task",
      "taskReferenceName": "shipping_task_ref",
      "inputParameters": {
        "name": "${shipping_info_ref.output.name}",
		"streetAddress": "${shipping_info_ref.output.streetAddress}",
		"city": "${shipping_info_ref.output.city}",
		"state": "${shipping_info_ref.output.state}",
		"zipcode": "${shipping_info_ref.output.zipcode}",
      },
      "type": "SIMPLE"
    }
  ],
  "outputParameters": {
    "trackingNumber": "${shipping_task_ref.output.trackingNumber}"
  },
  "failureWorkflow": "shipping_issues",
  "failureWorkflowVersion": 1,
  "restartable": true,
  "workflowStatusListenerEnabled": true,
  "ownerEmail": "conductor@example.com",
  "timeoutPolicy": "ALERT_ONLY",
  "timeoutSeconds": 0,
  "variables": {},
  "inputTemplate": {}
}
```

Upon completion of the 2 tasks, the workflow outputs the tracking number generated in the 2nd task.  If the workflow fails, a second workflow named ```shipping_issues``` is run.


### Example 2 - Task Configuration
Consider a task `http_task` with input configured to use input/output parameters from workflow and a task named `loc_task`.

```json
{
  "name": "encode_workflow",
  "description": "Encode movie.",
  "version": 1,
  "inputParameters": [
    "movieId", "fileLocation", "recipe"
  ],
  "tasks": [
    {
      "name": "loc_task",
      "taskReferenceName": "loc_task_ref",
      "taskType": "SIMPLE",
      ...      
    },    
    {
      "name": "http_task",
      "taskReferenceName": "http_task_ref",
      "taskType": "HTTP",
      "inputParameters": {
        "movieId": "${workflow.input.movieId}",
        "url": "${workflow.input.fileLocation}",
        "lang": "${loc_task.output.languages[0]}",
        "http_request": {
          "method": "POST",
          "url": "http://example.com/${loc_task.output.fileId}/encode",
          "body": {
            "recipe": "${workflow.input.recipe}",
            "params": {
              "width": 100,
              "height": 100
            }
          },
          "headers": {
            "Accept": "application/json",
            "Content-Type": "application/json"
          }
        }
      }
    }
  ],
  "ownerEmail": "conductor@example.com",
  "variables": {},
  "inputTemplate": {}
}

```

Consider the following as the _workflow input_

```json
{
  "movieId": "movie_123",
  "fileLocation":"s3://moviebucket/file123",
  "recipe":"png"
}
```
And the output of the _loc_task_ as the following;

```json
{
  "fileId": "file_xxx_yyy_zzz",
  "languages": ["en","ja","es"]
}
```

When scheduling the task, Conductor will merge the values from workflow input and `loc_task`'s output and create the input to the `http_task` as follows:

```json
{
  "movieId": "movie_123",
  "url": "s3://moviebucket/file123",
  "lang": "en",
  "http_request": {
    "method": "POST",
    "url": "http://example.com/file_xxx_yyy_zzz/encode",
    "body": {
      "recipe": "png",
      "params": {
        "width": 100,
        "height": 100
      }
    },
    "headers": {
    	"Accept": "application/json",
    	"Content-Type": "application/json"
    }
  }
}
```

### Example 3 - inputExpression
Given the following task configuration:
```json
{
  "name": "loc_task",
  "taskReferenceName": "loc_task_ref",
  "taskType": "SIMPLE",
  "inputExpression": {
    "expression": "workflow.input",
    "type": "JSON_PATH"
  }  
}
```

When the workflow is invoked with the following _workflow input_
```json
{
  "movieId": "movie_123",
  "fileLocation":"s3://moviebucket/file123",
  "recipe":"png"
}
```

When the task `loc_task` is scheduled, the entire workflow input object will be passed in as the task input:
```json
{
  "movieId": "movie_123",
  "fileLocation":"s3://moviebucket/file123",
  "recipe":"png"
}
```
