Correct xcom patch API to be usable with tasks

[amoghrajesh]

Body

Context: #58900 (comment)

If we use the xcom patch API to perform an update on an XCom via the PATCH API, you wouldn't really get a usable value out in the task.

Ways to repro this:

1. Define a dag like this:

from datetime import datetime

from airflow.decorators import task
from airflow.models.dag import DAG

with DAG(dag_id="add_dag", schedule=None, start_date=datetime(2022, 3, 4)) as dag:

    @task
    def add_one(x: int):
        return x + 1

    @task
    def add_two(y: int):
        # y is result of add_one
        return y + 1

    @task
    def add_all_past_values(**context):
        xcoms = context['ti'].xcom_pull(task_ids=["add_one", "add_two"], include_prior_dates=True)
        print(xcoms)


    res1 = add_one(1)
    add_two(res1)

    add_all_past_values()

2. Run the dag a couple times
3. Identify a past run and update its xcom entry:

curl -X 'PATCH' \
  'http://localhost:28080/api/v2/dags/add_dag/dagRuns/manual__2025-12-02T07%3A34%3A26%2B00%3A00/taskInstances/add_one/xcomEntries/return_value' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "value": 7,
  "map_index": -1
}'

4. Observe the output

Note the format of 7 from earlier patch, it is excessively serialised

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[jason810496]

Thanks for picking up the issue.

Could you provide any additional context or point me to relevant code sections?

The #58900 PR is a good point to get started with. Just FYI that the unit test for corresponding change will be located in airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py.

[amoghrajesh]

@sunank200 would you be working on this one?