Batch XCom lookups when resolving a mapped task's expand() arguments

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/xcom.py

@@ -17,10 +17,15 @@
 
 from __future__ import annotations
 
-from pydantic import JsonValue, RootModel
+from pydantic import Field, JsonValue, RootModel
 
 from airflow.api_fastapi.core_api.base import BaseModel
 
+# Each item expands to 3 SQL bind parameters (task_id, key, map_index); this bounds a
+# single batch request to at most 3000 bind parameters, safely under Postgres/MySQL/
+# modern-SQLite limits regardless of how many kwargs a caller's .expand() call has.
+MAX_XCOM_BATCH_ITEMS = 1000
+
 
 class XComResponse(BaseModel):
     """XCom schema for responses with fields that are needed for Runtime."""
@@ -40,3 +45,34 @@ class XComSequenceSliceResponse(RootModel):
     """XCom schema with minimal structure for slice-based access."""
 
     root: list[JsonValue]
+
+
+class XComBatchItemRequest(BaseModel):
+    """One XCom lookup within a batch request."""
+
+    task_id: str
+    key: str
+    map_index: int = -1
+
+
+class XComBatchRequestBody(BaseModel):
+    """Body for a batch XCom lookup, scoped to a single dag_id/run_id."""
+
+    items: list[XComBatchItemRequest] = Field(max_length=MAX_XCOM_BATCH_ITEMS)
+
+
+class XComBatchItemResponse(BaseModel):
+    """One XCom lookup result within a batch response."""
+
+    task_id: str
+    key: str
+    map_index: int
+    found: bool
+    value: JsonValue = None
+    """The returned XCom value in a JSON-compatible format. Meaningless when ``found`` is False."""
+
+
+class XComBatchResponse(BaseModel):
+    """Batch XCom lookup response, ordered the same as the request's items."""
+
+    items: list[XComBatchItemResponse]

airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py

@@ -61,6 +61,7 @@
 )
 authenticated_router.include_router(variables.router, prefix="/variables", tags=["Variables"])
 authenticated_router.include_router(xcoms.router, prefix="/xcoms", tags=["XComs"])
+authenticated_router.include_router(xcoms.batch_router, prefix="/xcoms", tags=["XComs"])
 authenticated_router.include_router(hitl.router, prefix="/hitlDetails", tags=["Human in the Loop"])
 authenticated_router.include_router(task_state_store.router, prefix="/store/ti", tags=["Task State Store"])
 authenticated_router.include_router(

airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py

@@ -22,12 +22,15 @@
 
 from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Request, Response, status
 from pydantic import JsonValue
-from sqlalchemy import delete
+from sqlalchemy import delete, select, tuple_
 from sqlalchemy.sql.selectable import Select
 
 from airflow.api_fastapi.common.db.common import SessionDep
 from airflow.api_fastapi.core_api.base import BaseModel
 from airflow.api_fastapi.execution_api.datamodels.xcom import (
+    XComBatchItemResponse,
+    XComBatchRequestBody,
+    XComBatchResponse,
     XComResponse,
     XComSequenceIndexResponse,
     XComSequenceSliceResponse,
@@ -38,23 +41,14 @@
 from airflow.utils.db import get_query_count
 
 
-def has_xcom_access(
-    dag_id: str,
-    run_id: str,
-    task_id: str,
-    xcom_key: Annotated[str, Path(alias="key", min_length=1)],
-    request: Request,
-    session: SessionDep,
-    token=CurrentTIToken,
-) -> bool:
+def _check_dag_team_access(dag_id: str, *, write: bool, session: SessionDep, token) -> None:
     """
-    Check whether the requesting task may access the XCom for ``dag_id``.
+    Raise 403 unless the requesting task's team may access XComs for ``dag_id``.
 
     In multi-team mode, XCom access is scoped by team ownership (resolved via the
     ``dag -> bundle -> team`` chain). There is no cross-team XCom sharing:
 
-    * reads (``GET``/``HEAD``) are allowed for the requester's own team or for
-      global (teamless) dags;
+    * reads are allowed for the requester's own team or for global (teamless) dags;
     * writes and deletes are allowed only for the requester's own team; a team
       task may not mutate a global dag's XCom, mirroring how team-scoped
       Variables and Connections behave.
@@ -67,18 +61,8 @@ def has_xcom_access(
     """
     from airflow.configuration import conf
 
-    write = request.method not in {"GET", "HEAD", "OPTIONS"}
-
-    log.debug(
-        "Checking %s XCom access for task instance '%s' to XCom '%s' on dag '%s'",
-        "write" if write else "read",
-        token.id,
-        xcom_key,
-        dag_id,
-    )
-
     if not conf.getboolean("core", "multi_team"):
-        return True
+        return
 
     from airflow.api_fastapi.execution_api.security import (
         _team_name_for_dag_stmt,
@@ -90,10 +74,10 @@ def has_xcom_access(
 
     # Same team (including a teamless task accessing a global, teamless dag) is always allowed.
     if target_team == requester_team:
-        return True
+        return
     # Reads may additionally reach global (teamless) dags; writes and deletes may not.
     if not write and target_team is None:
-        return True
+        return
 
     raise HTTPException(
         status_code=status.HTTP_403_FORBIDDEN,
@@ -104,6 +88,40 @@ def has_xcom_access(
     )
 
 
+def has_xcom_access(
+    dag_id: str,
+    run_id: str,
+    task_id: str,
+    xcom_key: Annotated[str, Path(alias="key", min_length=1)],
+    request: Request,
+    session: SessionDep,
+    token=CurrentTIToken,
+) -> bool:
+    """Check whether the requesting task may access the XCom for ``dag_id``. See ``_check_dag_team_access``."""
+    write = request.method not in {"GET", "HEAD", "OPTIONS"}
+    log.debug(
+        "Checking %s XCom access for task instance '%s' to XCom '%s' on dag '%s'",
+        "write" if write else "read",
+        token.id,
+        xcom_key,
+        dag_id,
+    )
+    _check_dag_team_access(dag_id, write=write, session=session, token=token)
+    return True
+
+
+def has_xcom_batch_access(
+    dag_id: str,
+    run_id: str,
+    session: SessionDep,
+    token=CurrentTIToken,
+) -> bool:
+    """Check whether the requesting task may batch-read XComs for ``dag_id``. See ``_check_dag_team_access``."""
+    log.debug("Checking read XCom batch access for task instance '%s' on dag '%s'", token.id, dag_id)
+    _check_dag_team_access(dag_id, write=False, session=session, token=token)
+    return True
+
+
 router = APIRouter(
     responses={
         status.HTTP_401_UNAUTHORIZED: {"description": "Unauthorized"},
@@ -482,3 +500,57 @@ def delete_xcom(
     )
     session.execute(query)
     return {"message": f"XCom with key: {key} successfully deleted."}
+
+
+# A separate router (registered directly by routes/__init__.py, not merged via
+# router.include_router) because ``router`` above carries ``has_xcom_access`` as a
+# constructor-level dependency, which needs ``task_id``/``key`` path params this
+# route doesn't have. Merging would leak that dependency onto this route's OpenAPI
+# operation as an unresolvable path parameter.
+batch_router = APIRouter(dependencies=[Depends(has_xcom_batch_access)])
+
+
+@batch_router.post(
+    "/{dag_id}/{run_id}/batch",
+    description="Look up multiple XCom values in one request, scoped to a single dag run",
+)
+def get_xcom_batch(
+    dag_id: str,
+    run_id: str,
+    body: XComBatchRequestBody,
+    session: SessionDep,
+) -> XComBatchResponse:
+    """
+    Batch-fetch XComs from the database - not other XCom Backends.
+
+    Collapses what would otherwise be one Execution API round trip per requested
+    XCom (e.g. one per XComArg kwarg of a mapped task's ``.expand()``) into a
+    single request. Missing items are reported via ``found=False`` rather than
+    failing the whole batch, since a partial miss (e.g. an upstream that hasn't
+    pushed yet) is an expected outcome for a batch, not an error.
+    """
+    if not body.items:
+        return XComBatchResponse(items=[])
+
+    requested = [(item.task_id, item.key, item.map_index) for item in body.items]
+    query = select(XComModel.task_id, XComModel.key, XComModel.map_index, XComModel.value).where(
+        XComModel.dag_id == dag_id,
+        XComModel.run_id == run_id,
+        tuple_(XComModel.task_id, XComModel.key, XComModel.map_index).in_(requested),
+    )
+    found_values = {
+        (task_id, key, map_index): value for task_id, key, map_index, value in session.execute(query)
+    }
+
+    return XComBatchResponse(
+        items=[
+            XComBatchItemResponse(
+                task_id=item.task_id,
+                key=item.key,
+                map_index=item.map_index,
+                found=(item.task_id, item.key, item.map_index) in found_values,
+                value=found_values.get((item.task_id, item.key, item.map_index)),
+            )
+            for item in body.items
+        ]
+    )

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -51,11 +51,14 @@
     AddTeamNameField,
     AddVariableKeysEndpoint,
 )
-from airflow.api_fastapi.execution_api.versions.v2026_10_30 import AddArgBindingsToTIRunContext
+from airflow.api_fastapi.execution_api.versions.v2026_10_30 import (
+    AddArgBindingsToTIRunContext,
+    AddXComBatchEndpoint,
+)
 
 bundle = VersionBundle(
     HeadVersion(),
-    Version("2026-10-30", AddArgBindingsToTIRunContext),
+    Version("2026-10-30", AddArgBindingsToTIRunContext, AddXComBatchEndpoint),
     Version(
         "2026-06-30",
         AddVariableKeysEndpoint,

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py

@@ -19,14 +19,26 @@
 
 from cadwyn import (
     ResponseInfo,
+    VersionChange,
     VersionChangeWithSideEffects,
     convert_response_to_previous_version_for,
+    endpoint,
     schema,
 )
 
 from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext
 
 
+class AddXComBatchEndpoint(VersionChange):
+    """Add a batch XCom lookup endpoint that resolves multiple XComs in one request, capped at MAX_XCOM_BATCH_ITEMS items per request."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (
+        endpoint("/xcoms/{dag_id}/{run_id}/batch", ["POST"]).didnt_exist,
+    )
+
+
 class AddArgBindingsToTIRunContext(VersionChangeWithSideEffects):
     """Add the ``arg_bindings`` argument-binding spec for stub (foreign-runtime) tasks."""
 

airflow-core/src/airflow/dag_processing/processor.py

@@ -54,6 +54,7 @@
     GetVariable,
     GetVariableKeys,
     GetXCom,
+    GetXComBatch,
     GetXComCount,
     GetXComSequenceItem,
     GetXComSequenceSlice,
@@ -66,6 +67,7 @@
     TaskStatesResult,
     VariableKeysResult,
     VariableResult,
+    XComBatchResult,
     XComCountResponse,
     XComResult,
     XComSequenceIndexResult,
@@ -80,6 +82,7 @@
     handle_get_ti_count,
     handle_get_variable_keys,
     handle_get_xcom,
+    handle_get_xcom_batch,
     handle_get_xcom_count,
     handle_get_xcom_sequence_item,
     handle_get_xcom_sequence_slice,
@@ -158,6 +161,7 @@ class DagFileParsingResult(BaseModel):
     | GetPreviousDagRun
     | GetPreviousTI
     | GetXCom
+    | GetXComBatch
     | GetXComCount
     | GetXComSequenceItem
     | GetXComSequenceSlice
@@ -176,6 +180,7 @@ class DagFileParsingResult(BaseModel):
     | PrevSuccessfulDagRunResult
     | ErrorResponse
     | OKResponse
+    | XComBatchResult
     | XComCountResponse
     | XComResult
     | XComSequenceIndexResult
@@ -718,6 +723,8 @@ def _handle_request(self, msg: ToManager, log: FilteringBoundLogger, req_id: int
             resp, dump_opts = handle_get_prev_successful_dag_run(self.client, self.id)
         elif isinstance(msg, GetXCom):
             resp, dump_opts = handle_get_xcom(self.client, msg)
+        elif isinstance(msg, GetXComBatch):
+            resp, dump_opts = handle_get_xcom_batch(self.client, msg)
         elif isinstance(msg, GetXComCount):
             resp, dump_opts = handle_get_xcom_count(self.client, msg)
         elif isinstance(msg, GetXComSequenceItem):