> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Synchronous Dag execution

Synchronous Dag execution refers to the ability in Airflow 3.1+ to trigger a Dag run using an API call and wait for it to complete before returning [XCom](/docs/learn/airflow-passing-data-between-tasks) values pushed by one or more tasks in the Dag run. This is useful both for single DAG runs and for cases where the same DAG may be triggered multiple times in parallel.

<Note>
  Synchronous Dag execution was added as an [experimental feature](https://airflow.apache.org/docs/apache-airflow/stable/release-process.html#experimental-features) in Airflow 3.1.
</Note>

## Assumed knowledge

* Basic knowledge of Airflow. See [Introduction to Apache Airflow](/docs/learn/intro-to-airflow).
* Knowing how to use the [Airflow REST API](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html).
* Basic understanding of XCom. See [Passing data between tasks](/docs/learn/airflow-passing-data-between-tasks).

## When to use synchronous Dag execution

Synchronous Dag execution is a way to use Airflow as the backend for services processing user requests coming from a frontend application like a website, mobile app, or slack bot. Common use cases include:

* Inference execution: A user provides input to a pipeline that interacts with one or more LLMs and/or AI agents to generate a response. The response is served back to the user as soon as the Dag has completed running.
* Ad-hoc requests: Non-technical stakeholders request data analyses that use a Dag to retrieve the desired result.
* Data submission: Non-technical users can submit their data to a Dag to be processed and get immediate feedback on the status of the request and the result.

## API endpoint

The endpoint to wait for a Dag run to complete is:

```text wrap theme={null}
GET api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/wait
```

It includes the following path parameters:

* `dag_id`: (Mandatory) The id of the DAG to wait for.
* `dag_run_id`: (Mandatory) The id of the DAG run to wait for.

The query parameters are:

* `interval`: (Mandatory) Seconds to wait between Dag run state checks.
* `result`: (Optional) Array of strings or null. A list of task ids from which to pull the XCom value pushed under the `return_value` key. In Airflow 3.3+ you specify which task in a Dag returns the result in the Dag code, see [specify the Dag result](#specify-the-dag-result).

Calling this endpoint on any running Dag will start a waiting process until the Dag run completes. If any task is specified as returning a result or any XCom are requested in the `result` parameter, they are returned in the response upon Dag run completion.

## Example script

The following script creates a Dag run for the `my_dag` Dag and waits for it to complete. It includes XComs pushed under the `return_value` key of the `my_task` task in the response.

```python expandable wrap theme={null}
import requests
from datetime import datetime
import json

_USERNAME = "admin"
_PASSWORD = "admin"
_HOST = "http://localhost:8080"  # To learn how to send API requests to Airflow running on Astro see: https://www.astronomer.io/docs/astro/airflow-api/

_DAG_ID = "my_dag"
_TASK_ID = "my_task"


def _get_jwt_token():
    token_url = f"{_HOST}/auth/token"
    payload = {"username": _USERNAME, "password": _PASSWORD}
    headers = {"Content-Type": "application/json"}
    response = requests.post(token_url, json=payload, headers=headers)

    token = response.json().get("access_token")
    return token


def _trigger_dag_run(dag_id: str):
    url = f"{_HOST}/api/v2/dags/{dag_id}/dagRuns"
    headers = {
        "Authorization": f"Bearer {_get_jwt_token()}",
        "Content-Type": "application/json",
    }
    payload = {
        "logical_date": None,
    }
    response = requests.post(url, headers=headers, json=payload)
    return response.json()["dag_run_id"]


def _wait_for_dag_run_completion(dag_id: str, dag_run_id: str):
    url = f"{_HOST}/api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/wait"
    headers = {
        "Authorization": f"Bearer {_get_jwt_token()}",
    }
    params = {
        "interval": 1,
        "result": [_TASK_ID],  # In Airflow 3.3+ the result task can also be defined in the Dag code
    }
    response = requests.get(url, headers=headers, params=params)
    print(f"Status Code: {response.status_code}")

    lines = response.text.strip().split("\n")
    json_objects = []

    for line in lines:
        if line.strip():
            json_obj = json.loads(line)
            json_objects.append(json_obj)
            print(f"Status: {json_obj.get('state', 'unknown')}")

    if json_objects:
        last_status_update = json_objects[-1]
        xcom_results = last_status_update.get("results", {})
        print("Last status update: ", last_status_update)
        print("XCom results: ", xcom_results)
        return xcom_results


if __name__ == "__main__":
    _dag_run_id = _trigger_dag_run(_DAG_ID)
    _wait_for_dag_run_completion(_DAG_ID, _dag_run_id)
```

Running the script above returns an output similar to the following:

```text wrap theme={null}
Status Code: 200
Status: queued
Status: running
Status: running
Status: success
Last status update:  {'state': 'success', 'results': {'my_task': 'Hello World!'}}
XCom results:  {'my_task': 'Hello World!'}
```

## Specify the Dag result

In Airflow 3.3+ you can define which task in a Dag returns the result the `wait` endpoint should return, using the `@result` decorator on top of a `@task` decorated function.

```python {3} wrap theme={null}
from airflow.sdk import task, result

@result
@task
def my_task():
    return "hello!"

my_task()
```

When using traditional operators, you can add a task's `.output` (the XCom pushed with the key `return_value`) to the Dag object.

```python {12} wrap theme={null}
from airflow.sdk import DAG, chain
from airflow.providers.standard.operators.python import PythonOperator


def _my_task_func():
    return "hello!"


with DAG("my_dag") as dag:
    _my_task = PythonOperator(task_id="my_task", python_callable=_my_task_func)

    dag.add_result(_my_task.output)
```