diff --git a/airflow/providers/databricks/hooks/databricks.py b/airflow/providers/databricks/hooks/databricks.py
index 9ae061b02e55c..2686e7fc72c92 100644
--- a/airflow/providers/databricks/hooks/databricks.py
+++ b/airflow/providers/databricks/hooks/databricks.py
@@ -126,8 +126,9 @@ def __init__(
         retry_limit: int = 3,
         retry_delay: float = 1.0,
         retry_args: Optional[Dict[Any, Any]] = None,
+        caller: str = "DatabricksHook",
     ) -> None:
-        super().__init__(databricks_conn_id, timeout_seconds, retry_limit, retry_delay, retry_args)
+        super().__init__(databricks_conn_id, timeout_seconds, retry_limit, retry_delay, retry_args, caller)
 
     def run_now(self, json: dict) -> int:
         """
diff --git a/airflow/providers/databricks/hooks/databricks_base.py b/airflow/providers/databricks/hooks/databricks_base.py
index 28b7feaf846ca..6c0e5a847ff1f 100644
--- a/airflow/providers/databricks/hooks/databricks_base.py
+++ b/airflow/providers/databricks/hooks/databricks_base.py
@@ -46,8 +46,7 @@
 from airflow.exceptions import AirflowException
 from airflow.hooks.base import BaseHook
 from airflow.models import Connection
-
-USER_AGENT_HEADER = {'user-agent': f'airflow-{__version__}'}
+from airflow.providers_manager import ProvidersManager
 
 # https://docs.microsoft.com/en-us/azure/databricks/dev-tools/api/latest/aad/service-prin-aad-token#--get-an-azure-active-directory-access-token
 # https://docs.microsoft.com/en-us/graph/deployments#app-registration-and-token-service-root-endpoints
@@ -96,6 +95,7 @@ def __init__(
         retry_limit: int = 3,
         retry_delay: float = 1.0,
         retry_args: Optional[Dict[Any, Any]] = None,
+        caller: str = "Unknown",
     ) -> None:
         super().__init__()
         self.databricks_conn_id = databricks_conn_id
@@ -106,6 +106,7 @@ def __init__(
         self.retry_delay = retry_delay
         self.aad_tokens: Dict[str, dict] = {}
         self.aad_timeout_seconds = 10
+        self.caller = caller
 
         def my_after_func(retry_state):
             self._log_request_error(retry_state.attempt_number, retry_state.outcome)
@@ -129,6 +130,21 @@ def databricks_conn(self) -> Connection:
     def get_conn(self) -> Connection:
         return self.databricks_conn
 
+    @cached_property
+    def user_agent_header(self) -> Dict[str, str]:
+        return {'user-agent': self.user_agent_value}
+
+    @cached_property
+    def user_agent_value(self) -> str:
+        manager = ProvidersManager()
+        package_name = manager.hooks[BaseDatabricksHook.conn_type].package_name  # type: ignore[union-attr]
+        provider = manager.providers[package_name]
+        version = provider.version
+        if provider.is_source:
+            version += "-source"
+
+        return f'Airflow/{__version__} Databricks/{version} ({self.caller})'
+
     @cached_property
     def host(self) -> str:
         if 'host' in self.databricks_conn.extra_dejson:
@@ -209,7 +225,7 @@ def _get_aad_token(self, resource: str) -> str:
                         resp = requests.get(
                             AZURE_METADATA_SERVICE_TOKEN_URL,
                             params=params,
-                            headers={**USER_AGENT_HEADER, "Metadata": "true"},
+                            headers={**self.user_agent_header, "Metadata": "true"},
                             timeout=self.aad_timeout_seconds,
                         )
                     else:
@@ -227,7 +243,7 @@ def _get_aad_token(self, resource: str) -> str:
                             AZURE_TOKEN_SERVICE_URL.format(azure_ad_endpoint, tenant_id),
                             data=data,
                             headers={
-                                **USER_AGENT_HEADER,
+                                **self.user_agent_header,
                                 'Content-Type': 'application/x-www-form-urlencoded',
                             },
                             timeout=self.aad_timeout_seconds,
@@ -274,7 +290,7 @@ async def _a_get_aad_token(self, resource: str) -> str:
                         async with self._session.get(
                             url=AZURE_METADATA_SERVICE_TOKEN_URL,
                             params=params,
-                            headers={**USER_AGENT_HEADER, "Metadata": "true"},
+                            headers={**self.user_agent_header, "Metadata": "true"},
                             timeout=self.aad_timeout_seconds,
                         ) as resp:
                             resp.raise_for_status()
@@ -294,7 +310,7 @@ async def _a_get_aad_token(self, resource: str) -> str:
                             url=AZURE_TOKEN_SERVICE_URL.format(azure_ad_endpoint, tenant_id),
                             data=data,
                             headers={
-                                **USER_AGENT_HEADER,
+                                **self.user_agent_header,
                                 'Content-Type': 'application/x-www-form-urlencoded',
                             },
                             timeout=self.aad_timeout_seconds,
@@ -467,7 +483,7 @@ def _do_api_call(
         url = f'https://{self.host}/{endpoint}'
 
         aad_headers = self._get_aad_headers()
-        headers = {**USER_AGENT_HEADER.copy(), **aad_headers}
+        headers = {**self.user_agent_header, **aad_headers}
 
         auth: AuthBase
         token = self._get_token()
@@ -525,7 +541,7 @@ async def _a_do_api_call(self, endpoint_info: Tuple[str, str], json: Optional[Di
         url = f'https://{self.host}/{endpoint}'
 
         aad_headers = await self._a_get_aad_headers()
-        headers = {**USER_AGENT_HEADER.copy(), **aad_headers}
+        headers = {**self.user_agent_header, **aad_headers}
 
         auth: aiohttp.BasicAuth
         token = await self._a_get_token()
@@ -551,7 +567,7 @@ async def _a_do_api_call(self, endpoint_info: Tuple[str, str], json: Optional[Di
                         url,
                         json=json,
                         auth=auth,
-                        headers={**headers, **USER_AGENT_HEADER},
+                        headers={**headers, **self.user_agent_header},
                         timeout=self.timeout_seconds,
                     ) as response:
                         response.raise_for_status()
diff --git a/airflow/providers/databricks/hooks/databricks_sql.py b/airflow/providers/databricks/hooks/databricks_sql.py
index 7a888438e9dc4..79bd69f395557 100644
--- a/airflow/providers/databricks/hooks/databricks_sql.py
+++ b/airflow/providers/databricks/hooks/databricks_sql.py
@@ -22,13 +22,11 @@
 from databricks import sql  # type: ignore[attr-defined]
 from databricks.sql.client import Connection  # type: ignore[attr-defined]
 
-from airflow import __version__
 from airflow.exceptions import AirflowException
 from airflow.providers.common.sql.hooks.sql import DbApiHook
 from airflow.providers.databricks.hooks.databricks_base import BaseDatabricksHook
 
 LIST_SQL_ENDPOINTS_ENDPOINT = ('GET', 'api/2.0/sql/endpoints')
-USER_AGENT_STRING = f'airflow-{__version__}'
 
 
 class DatabricksSqlHook(BaseDatabricksHook, DbApiHook):
@@ -62,9 +60,10 @@ def __init__(
         http_headers: Optional[List[Tuple[str, str]]] = None,
         catalog: Optional[str] = None,
         schema: Optional[str] = None,
+        caller: str = "DatabricksSqlHook",
         **kwargs,
     ) -> None:
-        super().__init__(databricks_conn_id)
+        super().__init__(databricks_conn_id, caller=caller)
         self._sql_conn = None
         self._token: Optional[str] = None
         self._http_path = http_path
@@ -132,7 +131,7 @@ def get_conn(self) -> Connection:
                 catalog=self.catalog,
                 session_configuration=self.session_config,
                 http_headers=self.http_headers,
-                _user_agent_entry=USER_AGENT_STRING,
+                _user_agent_entry=self.user_agent_value,
                 **self._get_extra_config(),
                 **self.additional_params,
             )
diff --git a/airflow/providers/databricks/operators/databricks.py b/airflow/providers/databricks/operators/databricks.py
index a63eaa06c1550..1ecf23639decd 100644
--- a/airflow/providers/databricks/operators/databricks.py
+++ b/airflow/providers/databricks/operators/databricks.py
@@ -22,6 +22,7 @@
 from logging import Logger
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
 
+from airflow.compat.functools import cached_property
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator, BaseOperatorLink, XCom
 from airflow.providers.databricks.hooks.databricks import DatabricksHook, RunState
@@ -383,23 +384,26 @@ def __init__(
         self.run_id: Optional[int] = None
         self.do_xcom_push = do_xcom_push
 
-    def _get_hook(self) -> DatabricksHook:
+    @cached_property
+    def _hook(self):
+        return self._get_hook(caller="DatabricksSubmitRunOperator")
+
+    def _get_hook(self, caller: str) -> DatabricksHook:
         return DatabricksHook(
             self.databricks_conn_id,
             retry_limit=self.databricks_retry_limit,
             retry_delay=self.databricks_retry_delay,
             retry_args=self.databricks_retry_args,
+            caller=caller,
         )
 
     def execute(self, context: 'Context'):
-        hook = self._get_hook()
-        self.run_id = hook.submit_run(self.json)
-        _handle_databricks_operator_execution(self, hook, self.log, context)
+        self.run_id = self._hook.submit_run(self.json)
+        _handle_databricks_operator_execution(self, self._hook, self.log, context)
 
     def on_kill(self):
         if self.run_id:
-            hook = self._get_hook()
-            hook.cancel_run(self.run_id)
+            self._hook.cancel_run(self.run_id)
             self.log.info(
                 'Task: %s with run_id: %s was requested to be cancelled.', self.task_id, self.run_id
             )
@@ -411,7 +415,7 @@ class DatabricksSubmitRunDeferrableOperator(DatabricksSubmitRunOperator):
     """Deferrable version of ``DatabricksSubmitRunOperator``"""
 
     def execute(self, context):
-        hook = self._get_hook()
+        hook = self._get_hook(caller="DatabricksSubmitRunDeferrableOperator")
         self.run_id = hook.submit_run(self.json)
         _handle_deferrable_databricks_operator_execution(self, hook, self.log, context)
 
@@ -635,16 +639,21 @@ def __init__(
         self.run_id: Optional[int] = None
         self.do_xcom_push = do_xcom_push
 
-    def _get_hook(self) -> DatabricksHook:
+    @cached_property
+    def _hook(self):
+        return self._get_hook(caller="DatabricksRunNowOperator")
+
+    def _get_hook(self, caller: str) -> DatabricksHook:
         return DatabricksHook(
             self.databricks_conn_id,
             retry_limit=self.databricks_retry_limit,
             retry_delay=self.databricks_retry_delay,
             retry_args=self.databricks_retry_args,
+            caller=caller,
         )
 
     def execute(self, context: 'Context'):
-        hook = self._get_hook()
+        hook = self._hook
         if 'job_name' in self.json:
             job_id = hook.find_job_id_by_name(self.json['job_name'])
             if job_id is None:
@@ -656,8 +665,7 @@ def execute(self, context: 'Context'):
 
     def on_kill(self):
         if self.run_id:
-            hook = self._get_hook()
-            hook.cancel_run(self.run_id)
+            self._hook.cancel_run(self.run_id)
             self.log.info(
                 'Task: %s with run_id: %s was requested to be cancelled.', self.task_id, self.run_id
             )
@@ -669,7 +677,7 @@ class DatabricksRunNowDeferrableOperator(DatabricksRunNowOperator):
     """Deferrable version of ``DatabricksRunNowOperator``"""
 
     def execute(self, context):
-        hook = self._get_hook()
+        hook = self._get_hook(caller="DatabricksRunNowDeferrableOperator")
         self.run_id = hook.run_now(self.json)
         _handle_deferrable_databricks_operator_execution(self, hook, self.log, context)
 
diff --git a/airflow/providers/databricks/operators/databricks_repos.py b/airflow/providers/databricks/operators/databricks_repos.py
index 97b46b8e244d6..32100bcb34250 100644
--- a/airflow/providers/databricks/operators/databricks_repos.py
+++ b/airflow/providers/databricks/operators/databricks_repos.py
@@ -21,6 +21,7 @@
 from typing import TYPE_CHECKING, Optional, Sequence
 from urllib.parse import urlparse
 
+from airflow.compat.functools import cached_property
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator
 from airflow.providers.databricks.hooks.databricks import DatabricksHook
@@ -116,11 +117,13 @@ def __detect_repo_provider__(url):
             pass
         return provider
 
-    def _get_hook(self) -> DatabricksHook:
+    @cached_property
+    def _hook(self) -> DatabricksHook:
         return DatabricksHook(
             self.databricks_conn_id,
             retry_limit=self.databricks_retry_limit,
             retry_delay=self.databricks_retry_delay,
+            caller="DatabricksReposCreateOperator",
         )
 
     def execute(self, context: 'Context'):
@@ -140,22 +143,21 @@ def execute(self, context: 'Context'):
                     f"repo_path should have form of /Repos/{{folder}}/{{repo-name}}, got '{self.repo_path}'"
                 )
             payload["path"] = self.repo_path
-        hook = self._get_hook()
         existing_repo_id = None
         if self.repo_path is not None:
-            existing_repo_id = hook.get_repo_by_path(self.repo_path)
+            existing_repo_id = self._hook.get_repo_by_path(self.repo_path)
             if existing_repo_id is not None and not self.ignore_existing_repo:
                 raise AirflowException(f"Repo with path '{self.repo_path}' already exists")
         if existing_repo_id is None:
-            result = hook.create_repo(payload)
+            result = self._hook.create_repo(payload)
             repo_id = result["id"]
         else:
             repo_id = existing_repo_id
         # update repo if necessary
         if self.branch is not None:
-            hook.update_repo(str(repo_id), {'branch': str(self.branch)})
+            self._hook.update_repo(str(repo_id), {'branch': str(self.branch)})
         elif self.tag is not None:
-            hook.update_repo(str(repo_id), {'tag': str(self.tag)})
+            self._hook.update_repo(str(repo_id), {'tag': str(self.tag)})
 
         return repo_id
 
@@ -213,17 +215,18 @@ def __init__(
         self.branch = branch
         self.tag = tag
 
-    def _get_hook(self) -> DatabricksHook:
+    @cached_property
+    def _hook(self) -> DatabricksHook:
         return DatabricksHook(
             self.databricks_conn_id,
             retry_limit=self.databricks_retry_limit,
             retry_delay=self.databricks_retry_delay,
+            caller="DatabricksReposUpdateOperator",
         )
 
     def execute(self, context: 'Context'):
-        hook = self._get_hook()
         if self.repo_path is not None:
-            self.repo_id = hook.get_repo_by_path(self.repo_path)
+            self.repo_id = self._hook.get_repo_by_path(self.repo_path)
             if self.repo_id is None:
                 raise AirflowException(f"Can't find Repo ID for path '{self.repo_path}'")
         if self.branch is not None:
@@ -231,7 +234,7 @@ def execute(self, context: 'Context'):
         else:
             payload = {'tag': str(self.tag)}
 
-        result = hook.update_repo(str(self.repo_id), payload)
+        result = self._hook.update_repo(str(self.repo_id), payload)
         return result['head_commit_id']
 
 
@@ -278,18 +281,19 @@ def __init__(
         self.repo_path = repo_path
         self.repo_id = repo_id
 
-    def _get_hook(self) -> DatabricksHook:
+    @cached_property
+    def _hook(self) -> DatabricksHook:
         return DatabricksHook(
             self.databricks_conn_id,
             retry_limit=self.databricks_retry_limit,
             retry_delay=self.databricks_retry_delay,
+            caller="DatabricksReposDeleteOperator",
         )
 
     def execute(self, context: 'Context'):
-        hook = self._get_hook()
         if self.repo_path is not None:
-            self.repo_id = hook.get_repo_by_path(self.repo_path)
+            self.repo_id = self._hook.get_repo_by_path(self.repo_path)
             if self.repo_id is None:
                 raise AirflowException(f"Can't find Repo ID for path '{self.repo_path}'")
 
-        hook.delete_repo(str(self.repo_id))
+        self._hook.delete_repo(str(self.repo_id))
diff --git a/airflow/providers/databricks/operators/databricks_sql.py b/airflow/providers/databricks/operators/databricks_sql.py
index ad4add30fe8e6..6ee125a35e7af 100644
--- a/airflow/providers/databricks/operators/databricks_sql.py
+++ b/airflow/providers/databricks/operators/databricks_sql.py
@@ -114,6 +114,7 @@ def _get_hook(self) -> DatabricksSqlHook:
             http_headers=self.http_headers,
             catalog=self.catalog,
             schema=self.schema,
+            caller="DatabricksSqlOperator",
             **self.client_parameters,
         )
 
@@ -279,6 +280,7 @@ def _get_hook(self) -> DatabricksSqlHook:
             http_headers=self._http_headers,
             catalog=self._catalog,
             schema=self._schema,
+            caller="DatabricksCopyIntoOperator",
             **self._client_parameters,
         )
 
diff --git a/tests/providers/databricks/hooks/test_databricks.py b/tests/providers/databricks/hooks/test_databricks.py
index fa2c416a104de..fe4d67d695843 100644
--- a/tests/providers/databricks/hooks/test_databricks.py
+++ b/tests/providers/databricks/hooks/test_databricks.py
@@ -29,7 +29,6 @@
 from requests import exceptions as requests_exceptions
 from requests.auth import HTTPBasicAuth
 
-from airflow import __version__
 from airflow.exceptions import AirflowException
 from airflow.models import Connection
 from airflow.providers.databricks.hooks.databricks import (
@@ -75,7 +74,6 @@
 LOGIN = 'login'
 PASSWORD = 'password'
 TOKEN = 'token'
-USER_AGENT_HEADER = {'user-agent': f'airflow-{__version__}'}
 RUN_PAGE_URL = 'https://XX.cloud.databricks.com/#jobs/1/runs/1'
 LIFE_CYCLE_STATE = 'PENDING'
 STATE_MESSAGE = 'Waiting for cluster'
@@ -353,7 +351,7 @@ def test_do_api_call_patch(self, mock_requests):
             json={'cluster_name': 'new_name'},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -372,7 +370,7 @@ def test_submit_run(self, mock_requests):
             },
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -391,7 +389,7 @@ def test_spark_python_submit_run(self, mock_requests):
             },
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -411,7 +409,7 @@ def test_run_now(self, mock_requests):
             json={'notebook_params': NOTEBOOK_PARAMS, 'jar_params': JAR_PARAMS, 'job_id': JOB_ID},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -427,7 +425,7 @@ def test_get_run_page_url(self, mock_requests):
             json=None,
             params={'run_id': RUN_ID},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -443,7 +441,7 @@ def test_get_job_id(self, mock_requests):
             json=None,
             params={'run_id': RUN_ID},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -459,7 +457,7 @@ def test_get_run_output(self, mock_requests):
             json=None,
             params={'run_id': RUN_ID},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -475,7 +473,7 @@ def test_get_run_state(self, mock_requests):
             json=None,
             params={'run_id': RUN_ID},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -514,7 +512,7 @@ def test_cancel_run(self, mock_requests):
             json={'run_id': RUN_ID},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -532,7 +530,7 @@ def test_start_cluster(self, mock_requests):
             json={'cluster_id': CLUSTER_ID},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -550,7 +548,7 @@ def test_restart_cluster(self, mock_requests):
             json={'cluster_id': CLUSTER_ID},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -568,7 +566,7 @@ def test_terminate_cluster(self, mock_requests):
             json={'cluster_id': CLUSTER_ID},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -587,7 +585,7 @@ def test_install_libs_on_cluster(self, mock_requests):
             json={'cluster_id': CLUSTER_ID, 'libraries': LIBRARIES},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -606,7 +604,7 @@ def test_uninstall_libs_on_cluster(self, mock_requests):
             json={'cluster_id': CLUSTER_ID, 'libraries': LIBRARIES},
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -630,7 +628,7 @@ def test_list_jobs_success_single_page(self, mock_requests):
             json=None,
             params={'limit': 25, 'offset': 0, 'expand_tasks': False},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -671,7 +669,7 @@ def test_get_job_id_by_name_success(self, mock_requests):
             json=None,
             params={'limit': 25, 'offset': 0, 'expand_tasks': False},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -689,7 +687,7 @@ def test_get_job_id_by_name_not_found(self, mock_requests):
             json=None,
             params={'limit': 25, 'offset': 0, 'expand_tasks': False},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -712,7 +710,7 @@ def test_get_job_id_by_name_raise_exception_with_duplicates(self, mock_requests)
             json=None,
             params={'limit': 25, 'offset': 0, 'expand_tasks': False},
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -729,7 +727,7 @@ def test_connection_success(self, mock_requests):
             json=None,
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -746,7 +744,7 @@ def test_connection_failure(self, mock_requests):
             json=None,
             params=None,
             auth=HTTPBasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -1131,7 +1129,7 @@ async def test_do_api_call_patch(self, mock_patch):
             submit_run_endpoint(HOST),
             json={'cluster_name': 'new_name'},
             auth=aiohttp.BasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -1147,7 +1145,7 @@ async def test_get_run_page_url(self, mock_get):
             get_run_endpoint(HOST),
             json={'run_id': RUN_ID},
             auth=aiohttp.BasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -1164,7 +1162,7 @@ async def test_get_run_state(self, mock_get):
             get_run_endpoint(HOST),
             json={'run_id': RUN_ID},
             auth=aiohttp.BasicAuth(LOGIN, PASSWORD),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -1206,7 +1204,7 @@ async def test_get_run_state(self, mock_post, mock_get):
             get_run_endpoint(HOST),
             json={'run_id': RUN_ID},
             auth=BearerAuth(TOKEN),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -1258,7 +1256,7 @@ async def test_get_run_state(self, mock_post, mock_get):
             get_run_endpoint(HOST),
             json={'run_id': RUN_ID},
             auth=BearerAuth(TOKEN),
-            headers=USER_AGENT_HEADER,
+            headers=self.hook.user_agent_header,
             timeout=self.hook.timeout_seconds,
         )
 
@@ -1321,7 +1319,7 @@ async def test_get_run_state(self, mock_post, mock_get):
             json={'run_id': RUN_ID},
             auth=BearerAuth(TOKEN),
             headers={
-                **USER_AGENT_HEADER,
+                **self.hook.user_agent_header,
                 'X-Databricks-Azure-Workspace-Resource-Id': '/Some/resource',
                 'X-Databricks-Azure-SP-Management-Token': TOKEN,
             },
diff --git a/tests/providers/databricks/operators/test_databricks.py b/tests/providers/databricks/operators/test_databricks.py
index b6cd10e073612..e94de7290163c 100644
--- a/tests/providers/databricks/operators/test_databricks.py
+++ b/tests/providers/databricks/operators/test_databricks.py
@@ -230,6 +230,7 @@ def test_exec_success(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksSubmitRunOperator',
         )
 
         db_mock.submit_run.assert_called_once_with(expected)
@@ -266,6 +267,7 @@ def test_exec_failure(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksSubmitRunOperator',
         )
         db_mock.submit_run.assert_called_once_with(expected)
         db_mock.get_run_page_url.assert_called_once_with(RUN_ID)
@@ -309,6 +311,7 @@ def test_wait_for_termination(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksSubmitRunOperator',
         )
 
         db_mock.submit_run.assert_called_once_with(expected)
@@ -337,6 +340,7 @@ def test_no_wait_for_termination(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksSubmitRunOperator',
         )
 
         db_mock.submit_run.assert_called_once_with(expected)
@@ -372,6 +376,7 @@ def test_execute_task_deferred(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksSubmitRunDeferrableOperator',
         )
 
         db_mock.submit_run.assert_called_once_with(expected)
@@ -548,6 +553,7 @@ def test_exec_success(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksRunNowOperator',
         )
         db_mock.run_now.assert_called_once_with(expected)
         db_mock.get_run_page_url.assert_called_once_with(RUN_ID)
@@ -581,6 +587,7 @@ def test_exec_failure(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksRunNowOperator',
         )
         db_mock.run_now.assert_called_once_with(expected)
         db_mock.get_run_page_url.assert_called_once_with(RUN_ID)
@@ -622,6 +629,7 @@ def test_wait_for_termination(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksRunNowOperator',
         )
 
         db_mock.run_now.assert_called_once_with(expected)
@@ -652,6 +660,7 @@ def test_no_wait_for_termination(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksRunNowOperator',
         )
 
         db_mock.run_now.assert_called_once_with(expected)
@@ -697,6 +706,7 @@ def test_exec_with_job_name(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksRunNowOperator',
         )
         db_mock.find_job_id_by_name.assert_called_once_with(JOB_NAME)
         db_mock.run_now.assert_called_once_with(expected)
@@ -749,6 +759,7 @@ def test_execute_task_deferred(self, db_mock_class):
             retry_limit=op.databricks_retry_limit,
             retry_delay=op.databricks_retry_delay,
             retry_args=None,
+            caller='DatabricksRunNowDeferrableOperator',
         )
 
         db_mock.run_now.assert_called_once_with(expected)
diff --git a/tests/providers/databricks/operators/test_databricks_repos.py b/tests/providers/databricks/operators/test_databricks_repos.py
index aaf03261ba7fa..03124fd6464ed 100644
--- a/tests/providers/databricks/operators/test_databricks_repos.py
+++ b/tests/providers/databricks/operators/test_databricks_repos.py
@@ -45,7 +45,10 @@ def test_update_with_id(self, db_mock_class):
         op.execute(None)
 
         db_mock_class.assert_called_once_with(
-            DEFAULT_CONN_ID, retry_limit=op.databricks_retry_limit, retry_delay=op.databricks_retry_delay
+            DEFAULT_CONN_ID,
+            retry_limit=op.databricks_retry_limit,
+            retry_delay=op.databricks_retry_delay,
+            caller='DatabricksReposUpdateOperator',
         )
 
         db_mock.update_repo.assert_called_once_with('123', {'branch': 'releases'})
@@ -65,7 +68,10 @@ def test_update_with_path(self, db_mock_class):
         op.execute(None)
 
         db_mock_class.assert_called_once_with(
-            DEFAULT_CONN_ID, retry_limit=op.databricks_retry_limit, retry_delay=op.databricks_retry_delay
+            DEFAULT_CONN_ID,
+            retry_limit=op.databricks_retry_limit,
+            retry_delay=op.databricks_retry_delay,
+            caller='DatabricksReposUpdateOperator',
         )
 
         db_mock.update_repo.assert_called_once_with('123', {'tag': 'v1.0.0'})
@@ -104,7 +110,10 @@ def test_delete_with_id(self, db_mock_class):
         op.execute(None)
 
         db_mock_class.assert_called_once_with(
-            DEFAULT_CONN_ID, retry_limit=op.databricks_retry_limit, retry_delay=op.databricks_retry_delay
+            DEFAULT_CONN_ID,
+            retry_limit=op.databricks_retry_limit,
+            retry_delay=op.databricks_retry_delay,
+            caller='DatabricksReposDeleteOperator',
         )
 
         db_mock.delete_repo.assert_called_once_with('123')
@@ -122,7 +131,10 @@ def test_delete_with_path(self, db_mock_class):
         op.execute(None)
 
         db_mock_class.assert_called_once_with(
-            DEFAULT_CONN_ID, retry_limit=op.databricks_retry_limit, retry_delay=op.databricks_retry_delay
+            DEFAULT_CONN_ID,
+            retry_limit=op.databricks_retry_limit,
+            retry_delay=op.databricks_retry_delay,
+            caller='DatabricksReposDeleteOperator',
         )
 
         db_mock.delete_repo.assert_called_once_with('123')
@@ -159,7 +171,10 @@ def test_create_plus_checkout(self, db_mock_class):
         op.execute(None)
 
         db_mock_class.assert_called_once_with(
-            DEFAULT_CONN_ID, retry_limit=op.databricks_retry_limit, retry_delay=op.databricks_retry_delay
+            DEFAULT_CONN_ID,
+            retry_limit=op.databricks_retry_limit,
+            retry_delay=op.databricks_retry_delay,
+            caller='DatabricksReposCreateOperator',
         )
 
         db_mock.create_repo.assert_called_once_with({'url': git_url, 'provider': 'gitHub', 'path': repo_path})
@@ -186,7 +201,10 @@ def test_create_ignore_existing_plus_checkout(self, db_mock_class):
         op.execute(None)
 
         db_mock_class.assert_called_once_with(
-            DEFAULT_CONN_ID, retry_limit=op.databricks_retry_limit, retry_delay=op.databricks_retry_delay
+            DEFAULT_CONN_ID,
+            retry_limit=op.databricks_retry_limit,
+            retry_delay=op.databricks_retry_delay,
+            caller='DatabricksReposCreateOperator',
         )
 
         db_mock.get_repo_by_path.assert_called_once_with(repo_path)
diff --git a/tests/providers/databricks/operators/test_databricks_sql.py b/tests/providers/databricks/operators/test_databricks_sql.py
index 6775ff83988ce..6315e7e699f6f 100644
--- a/tests/providers/databricks/operators/test_databricks_sql.py
+++ b/tests/providers/databricks/operators/test_databricks_sql.py
@@ -61,6 +61,7 @@ def test_exec_success(self, db_mock_class):
             http_headers=None,
             catalog=None,
             schema=None,
+            caller='DatabricksSqlOperator',
         )
         db_mock.run.assert_called_once_with(sql, parameters=None, handler=fetch_all_handler)
 
@@ -92,6 +93,7 @@ def test_exec_write_file(self, db_mock_class):
             http_headers=None,
             catalog=None,
             schema=None,
+            caller='DatabricksSqlOperator',
         )
         db_mock.run.assert_called_once_with(sql, parameters=None, handler=fetch_all_handler)
 
