diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py
index 495eb27d6a7f6..36321dbe10a28 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/data_factory.py
@@ -181,7 +181,7 @@ def execute(self, context: Context) -> None:
             start_from_failure=self.start_from_failure,
             parameters=self.parameters,
         )
-        self.run_id = vars(response)["run_id"]
+        self.run_id = response.run_id
         # Push the ``run_id`` value to XCom regardless of what happens during execution. This allows for
         # retrieval the executed pipeline's ``run_id`` for downstream tasks especially if performing an
         # asynchronous wait.
diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py
index b0c171a13bd53..19a082ced77d6 100644
--- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py
+++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/operators/synapse.py
@@ -104,7 +104,7 @@ def execute(self, context: Context) -> None:
         self.log.info("Executing the Synapse spark job.")
         response = self.hook.run_spark_job(payload=self.payload)
         self.log.info(response)
-        self.job_id = vars(response)["id"]
+        self.job_id = response.id
         # Push the ``job_id`` value to XCom regardless of what happens during execution. This allows for
         # retrieval the executed job's ``id`` for downstream tasks especially if performing an
         # asynchronous wait.
@@ -257,7 +257,7 @@ def execute(self, context) -> None:
             start_activity_name=self.start_activity_name,
             parameters=self.parameters,
         )
-        self.run_id = vars(response)["run_id"]
+        self.run_id = response.run_id
         # Push the ``run_id`` value to XCom regardless of what happens during execution. This allows for
         # retrieval the executed pipeline's ``run_id`` for downstream tasks especially if performing an
         # asynchronous wait.
diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py
index c640303048061..7804891fd4b06 100644
--- a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py
+++ b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_data_factory.py
@@ -235,6 +235,32 @@ def test_execute_no_wait_for_termination(self, mock_run_pipeline):
             # Checking the pipeline run status should _not_ be called when ``wait_for_termination`` is False.
             mock_get_pipeline_run.assert_not_called()
 
+    @mock.patch("airflow.providers.microsoft.azure.hooks.data_factory.AzureDataFactoryHook.run_pipeline")
+    def test_run_id_extracted_from_hybrid_model_response(self, mock_run_pipeline):
+        """Regression test: azure-mgmt-datafactory v10 hybrid models don't expose attributes via vars().
+
+        Hybrid models use property descriptors, so vars(response)["run_id"] raises KeyError.
+        The operator must use attribute access (response.run_id) which works with both old and new models.
+        """
+
+        class HybridModelResponse(dict):
+            """Simulates an azure-mgmt-datafactory v10 hybrid model response."""
+
+            def __init__(self):
+                super().__init__({"runId": "hybrid-run-id-123"})
+
+            @property
+            def run_id(self):
+                return self["runId"]
+
+        mock_run_pipeline.return_value = HybridModelResponse()
+
+        operator = AzureDataFactoryRunPipelineOperator(wait_for_termination=False, **self.config)
+        operator.execute(context=self.mock_context)
+
+        assert operator.run_id == "hybrid-run-id-123"
+        self.mock_ti.xcom_push.assert_called_once_with(key="run_id", value="hybrid-run-id-123")
+
     @pytest.mark.db_test
     @pytest.mark.parametrize(
         ("resource_group", "factory"),
diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py
index f7c256f04c35b..419e34c00eb50 100644
--- a/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py
+++ b/providers/microsoft/azure/tests/unit/microsoft/azure/operators/test_synapse.py
@@ -135,6 +135,30 @@ def test_azure_synapse_run_spark_batch_operator_on_kill(
         op.on_kill()
         mock_cancel_job_run.assert_called_once_with(job_id=JOB_RUN_RESPONSE["id"])
 
+    @mock.patch("airflow.providers.microsoft.azure.hooks.synapse.AzureSynapseHook.get_job_run_status")
+    @mock.patch("airflow.providers.microsoft.azure.hooks.synapse.AzureSynapseHook.run_spark_job")
+    def test_job_id_extracted_from_hybrid_model_response(self, mock_run_spark_job, mock_get_job_run_status):
+        """Regression test: azure SDK v10 hybrid models don't expose attributes via vars()."""
+
+        class HybridModelResponse(dict):
+            """Simulates an azure SDK v10 hybrid model response."""
+
+            def __init__(self):
+                super().__init__({"id": 456})
+
+            @property
+            def id(self):
+                return self["id"]
+
+        mock_get_job_run_status.return_value = "success"
+        mock_run_spark_job.return_value = HybridModelResponse()
+
+        op = AzureSynapseRunSparkBatchOperator(
+            task_id="test", azure_synapse_conn_id=AZURE_SYNAPSE_CONN_ID, spark_pool="test_pool", payload={}
+        )
+        op.execute(context=self.mock_context)
+        assert op.job_id == 456
+
 
 class TestAzureSynapseRunPipelineOperator:
     @pytest.fixture(autouse=True)
@@ -288,6 +312,28 @@ def test_execute_no_wait_for_termination(self, mock_run_pipeline):
             # Checking the pipeline run status should _not_ be called when ``wait_for_termination`` is False.
             mock_get_pipeline_run.assert_not_called()
 
+    @mock.patch("airflow.providers.microsoft.azure.hooks.synapse.AzureSynapsePipelineHook.run_pipeline")
+    def test_run_id_extracted_from_hybrid_model_response(self, mock_run_pipeline):
+        """Regression test: azure SDK v10 hybrid models don't expose attributes via vars()."""
+
+        class HybridModelResponse(dict):
+            """Simulates an azure SDK v10 hybrid model response."""
+
+            def __init__(self):
+                super().__init__({"runId": "hybrid-run-id-456"})
+
+            @property
+            def run_id(self):
+                return self["runId"]
+
+        mock_run_pipeline.return_value = HybridModelResponse()
+
+        operator = AzureSynapseRunPipelineOperator(wait_for_termination=False, **self.config)
+        operator.execute(context=self.mock_context)
+
+        assert operator.run_id == "hybrid-run-id-456"
+        self.mock_ti.xcom_push.assert_called_once_with(key="run_id", value="hybrid-run-id-456")
+
     @pytest.mark.db_test
     def test_run_pipeline_operator_link(
         self,