[justinpakzad]

---

Currently, tasks that use .map() over a dict input fail because _MapResult.__getitem__ uses the index as a dict key (dict[0]), which raises KeyError.MapXComArg.resolve() says "XCom map expects sequence or dict", so dictionaries appear to be a valid type here. This PR updates __getitem__ to select the dict entry by position, so each callable gets a key/value pair, matching how expand() handles a dict. Additional test cases were added to cover the new behavior.

Repro Dag

from airflow.sdk import dag, task
@dag
def xcom_map_dict():
    @task
    def push():
        return {"orders": 120, "customers": 45}

    @task
    def pull(value):
        print(value)

    pull.expand(value=push().map(lambda item: f"{item[0]}: {item[1]} rows"))
xcom_map_dict()

Was generative AI tooling used to co-author this PR?

• Yes (Claude Sonnet 5.5)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.