diff --git a/databusclient/__init__.py b/databusclient/__init__.py
index 37ef670..bfc449b 100644
--- a/databusclient/__init__.py
+++ b/databusclient/__init__.py
@@ -7,9 +7,18 @@
 
 from databusclient import cli
 from databusclient.api.deploy import create_dataset, create_distribution, deploy
+from databusclient.api.download import download
+from databusclient.api.delete import delete
 from databusclient.version import __version__
 
-__all__ = ["__version__", "create_dataset", "deploy", "create_distribution"]
+__all__ = [
+    "__version__",
+    "create_dataset",
+    "deploy",
+    "create_distribution",
+    "download",
+    "delete",
+]
 
 
 def run():
diff --git a/databusclient/api/delete.py b/databusclient/api/delete.py
index 199e5a4..a1e50dd 100644
--- a/databusclient/api/delete.py
+++ b/databusclient/api/delete.py
@@ -14,6 +14,7 @@
 from databusclient.api.utils import (
     fetch_databus_jsonld,
     get_databus_id_parts_from_file_url,
+    get_default_headers,
 )
 
 
@@ -23,16 +24,18 @@ class DeleteQueue:
     Allows adding multiple databus URIs to a queue and executing their deletion in batch.
     """
 
-    def __init__(self, databus_key: str, manifest_context=None):
+    def __init__(self, databus_key: str, manifest_context=None, headers=None):
         """Create a DeleteQueue bound to a given Databus API key.
 
         Args:
             databus_key: API key used to authenticate deletion requests.
             manifest_context: Optional ManifestContext to record deletion
                 outcomes into. Passed through to _delete_list on execute().
+            headers: Optional custom HTTP headers to send with deletion requests.
         """
         self.databus_key = databus_key
         self.manifest_context = manifest_context
+        self.headers = headers
         self.queue: set[str] = set()
 
     def add_uri(self, databusURI: str):
@@ -122,6 +125,7 @@ def _delete_resource(
     force: bool = False,
     queue: DeleteQueue = None,
     manifest_context=None,
+    headers=None,
 ):
     """Delete a single Databus resource (version, artifact, group).
 
@@ -159,8 +163,11 @@ def _delete_resource(
         return
 
     print(f"[DELETE] {databusURI}")
-    headers = {"accept": "*/*", "X-API-KEY": databus_key}
-    response = requests.delete(databusURI, headers=headers, timeout=30)
+    req_headers = get_default_headers(headers)
+    if "accept" not in req_headers and "Accept" not in req_headers:
+        req_headers["accept"] = "*/*"
+    req_headers["X-API-KEY"] = databus_key
+    response = requests.delete(databusURI, headers=req_headers, timeout=30)
 
     if response.status_code in (200, 204):
         print(f"Successfully deleted: {databusURI}")
@@ -179,6 +186,7 @@ def _delete_list(
     force: bool = False,
     queue: DeleteQueue = None,
     manifest_context=None,
+    headers=None,
 ):
     """Delete a list of Databus resources.
 
@@ -191,7 +199,7 @@ def _delete_list(
     """
     for databusURI in databusURIs:
         _delete_resource(
-            databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context
+            databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers
         )
 
 
@@ -202,6 +210,7 @@ def _delete_artifact(
     force: bool = False,
     queue: DeleteQueue = None,
     manifest_context=None,
+    headers=None,
 ):
     """Delete an artifact and all its versions.
 
@@ -215,7 +224,7 @@ def _delete_artifact(
         force: If True, skip confirmation prompt and proceed with deletion.
         queue: If queue is provided, add the URI to the queue instead of deleting immediately.
     """
-    artifact_body = fetch_databus_jsonld(databusURI, databus_key)
+    artifact_body = fetch_databus_jsonld(databusURI, databus_key, headers=headers)
 
     json_dict = json.loads(artifact_body)
     versions = json_dict.get("databus:hasVersion")
@@ -235,7 +244,7 @@ def _delete_artifact(
         else:
             # Delete all versions
             _delete_list(
-                version_uris, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context
+                version_uris, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers
             )
 
     # Finally, delete the artifact itself
@@ -249,6 +258,7 @@ def _delete_group(
     force: bool = False,
     queue: DeleteQueue = None,
     manifest_context=None,
+    headers=None,
 ):
     """Delete a group and all its artifacts and versions.
 
@@ -262,7 +272,7 @@ def _delete_group(
         force: If True, skip confirmation prompt and proceed with deletion.
         queue: If queue is provided, add the URI to the queue instead of deleting immediately.
     """
-    group_body = fetch_databus_jsonld(databusURI, databus_key)
+    group_body = fetch_databus_jsonld(databusURI, databus_key, headers=headers)
 
     json_dict = json.loads(group_body)
     artifacts = json_dict.get("databus:hasArtifact", [])
@@ -286,7 +296,14 @@ def _delete_group(
     _delete_resource(databusURI, databus_key, dry_run=dry_run, force=force, queue=queue,manifest_context=manifest_context)
 
 
-def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool, manifest_context=None):
+def delete(
+    databusURIs: List[str],
+    databus_key: str,
+    dry_run: bool = False,
+    force: bool = False,
+    manifest_context=None,
+    headers=None,
+):
     """Delete a dataset from the databus.
 
     Delete a group, artifact, or version identified by the given databus URI.
@@ -299,7 +316,7 @@ def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool,
         force: If True, skip confirmation prompt and proceed with deletion.
     """
 
-    queue = DeleteQueue(databus_key, manifest_context=manifest_context)
+    queue = DeleteQueue(databus_key, manifest_context=manifest_context, headers=headers)
 
     for databusURI in databusURIs:
         _host, _account, group, artifact, version, file = (
@@ -309,14 +326,14 @@ def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool,
         if group == "collections" and artifact is not None:
             print(f"Deleting collection: {databusURI}")
             _delete_resource(
-                databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context
+                databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers
             )
         elif file is not None:
             print(f"Deleting file is not supported via API: {databusURI}")
         elif version is not None:
             print(f"Deleting version: {databusURI}")
             _delete_resource(
-                databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context
+                databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers
             )
         elif artifact is not None:
             print(f"Deleting artifact and all its versions: {databusURI}")
diff --git a/databusclient/api/deploy.py b/databusclient/api/deploy.py
index 2c8cd08..63cc702 100644
--- a/databusclient/api/deploy.py
+++ b/databusclient/api/deploy.py
@@ -8,10 +8,12 @@
 import hashlib
 import json
 from enum import Enum
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
 
 import requests
 
+from databusclient.api.utils import get_default_headers
+
 _debug = False
 
 
@@ -176,14 +178,17 @@ def _get_file_stats(distribution_str: str) -> Tuple[Optional[str], Optional[int]
     return sha256sum, content_length
 
 
-def _load_file_stats(url: str) -> Tuple[str, int]:
+def _load_file_stats(
+    url: str, headers: Union[Dict[str, Any], Iterable[str], None] = None
+) -> Tuple[str, int]:
     """Download the file at ``url`` and compute its SHA-256 and length.
 
     This is used as a fallback when the caller did not supply checksum/size
     information in the CLI or metadata file.
     """
 
-    resp = requests.get(url, timeout=30)
+    req_headers = get_default_headers(headers)
+    resp = requests.get(url, headers=req_headers, timeout=30)
     if resp.status_code >= 400:
         raise requests.exceptions.RequestException(response=resp)
 
@@ -463,6 +468,7 @@ def deploy(
     verify_parts: bool = False,
     log_level: DeployLogLevel = DeployLogLevel.debug,
     debug: bool = False,
+    headers: Union[Dict[str, Any], Iterable[str], None] = None,
 ) -> None:
     """Deploys a dataset to the databus. The endpoint is inferred from the DataID identifier.
     Parameters
@@ -479,7 +485,10 @@ def deploy(
         controls whether output shold be printed to the console (stdout)
     """
 
-    headers = {"X-API-KEY": f"{api_key}", "Content-Type": "application/json"}
+    req_headers = get_default_headers(headers)
+    req_headers["X-API-KEY"] = f"{api_key}"
+    if "Content-Type" not in req_headers:
+        req_headers["Content-Type"] = "application/json"
     data = json.dumps(dataid)
 
     try:
@@ -491,7 +500,7 @@ def deploy(
         base
         + f"/api/publish?verify-parts={str(verify_parts).lower()}&log-level={log_level.name}"
     )
-    resp = requests.post(api_uri, data=data, headers=headers, timeout=30)
+    resp = requests.post(api_uri, data=data, headers=req_headers, timeout=30)
 
     if debug or _debug:
         try:
@@ -517,6 +526,7 @@ def deploy_from_metadata(
     artifact_version_description: str,
     license_url: str,
     apikey: str,
+    headers: Union[Dict[str, Any], Iterable[str], None] = None,
 ) -> None:
     """
     Deploy a dataset from metadata entries.
diff --git a/databusclient/api/download.py b/databusclient/api/download.py
index 356a2dd..9cd371f 100644
--- a/databusclient/api/download.py
+++ b/databusclient/api/download.py
@@ -18,6 +18,8 @@
     fetch_databus_jsonld,
     get_databus_id_parts_from_file_url,
     compute_sha256_and_length,
+    parse_headers,
+    get_default_headers,
 )
 from databusclient.filehandling.format import (
     convert_file,
@@ -259,7 +261,9 @@ def _convert_compression_format(
         raise RuntimeError(f"Compression conversion failed: {e}")
 
 
-def get_expected_checksum(databusURI: str, databus_key: str) -> str | None:
+def get_expected_checksum(
+    databusURI: str, databus_key: str, headers: dict | None = None
+) -> str | None:
     expected = None
 
     host, account, group, artifact, version, file = get_databus_id_parts_from_file_url(
@@ -269,7 +273,9 @@ def get_expected_checksum(databusURI: str, databus_key: str) -> str | None:
     try:
         if version is not None:
             version_uri = f"https://{host}/{account}/{group}/{artifact}/{version}"
-            json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key)
+            json_str = fetch_databus_jsonld(
+                version_uri, databus_key=databus_key, headers=headers
+            )
             checks = _extract_checksums_from_jsonld(json_str)
             expected = checks.get(databusURI) or checks.get(
                 "https://" + databusURI.removeprefix("http://").removeprefix("https://")
@@ -371,7 +377,9 @@ def _extract_checksums_from_jsonld(json_str: str) -> dict:
     return checksums
 
 
-def _resolve_checksums_for_urls(file_urls: List[str], databus_key: str | None) -> dict:
+def _resolve_checksums_for_urls(
+    file_urls: List[str], databus_key: str | None, headers: dict | None = None
+) -> dict:
     """
     Group file URLs by their Version URI, fetch each Version JSON-LD once,
     and return a combined url->checksum mapping for the provided URLs.
@@ -394,7 +402,9 @@ def _resolve_checksums_for_urls(file_urls: List[str], databus_key: str | None) -
     checksums: dict = {}
     for version_uri, urls_in_version in versions_map.items():
         try:
-            json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key)
+            json_str = fetch_databus_jsonld(
+                version_uri, databus_key=databus_key, headers=headers
+            )
             extracted_checksums = _extract_checksums_from_jsonld(json_str)
             for url in urls_in_version:
                 if url in extracted_checksums:
@@ -420,6 +430,7 @@ def _download_file(
     validate_checksum: bool = False,
     expected_checksum: str | None = None,
     manifest_context=None,
+    headers: dict | None = None,
 ) -> None:
     """Download a file from the internet with a progress bar using tqdm.
 
@@ -454,10 +465,12 @@ def _download_file(
         os.makedirs(dirpath, exist_ok=True)  # Create the necessary directories
 
     # --- 1. Get redirect URL by requesting HEAD ---
-    headers = {}
+    req_headers = get_default_headers(headers)
 
     # --- 1a. public databus ---
-    response = requests.head(url, timeout=30, allow_redirects=False)
+    response = requests.head(
+        url, headers=req_headers, timeout=30, allow_redirects=False
+    )
 
     # Check for redirect and update URL if necessary
     if response.headers.get("Location") and response.status_code in [
@@ -470,7 +483,7 @@ def _download_file(
         url = response.headers.get("Location")
         print("Redirects url: ", url)
         # Re-do HEAD request on redirect URL
-        response = requests.head(url, timeout=30)
+        response = requests.head(url, headers=req_headers, timeout=30)
 
     # Extract hostname from final URL (after redirect) to check if vault token needed.
     # This is the actual download location that may require authentication.
@@ -489,19 +502,22 @@ def _download_file(
             # Token provided; will handle in GET request below
         else:
             # Not a vault host; might need databus API key
-            if not databus_key:
+            if not databus_key and "X-API-KEY" not in req_headers:
                 raise DownloadAuthError(
                     "Databus API key not given for protected download"
                 )
-            headers = {"X-API-KEY": databus_key}
-            response = requests.head(url, headers=headers, timeout=30)
+            if databus_key:
+                req_headers["X-API-KEY"] = databus_key
+            response = requests.head(url, headers=req_headers, timeout=30)
 
     # --- 2. Try direct GET to redirected URL ---
-    headers["Accept-Encoding"] = (
-        "identity"  # disable gzip to get correct content-length
-    )
+    get_headers = dict(req_headers)
+    if "Accept-Encoding" not in get_headers:
+        get_headers["Accept-Encoding"] = (
+            "identity"  # disable gzip to get correct content-length
+        )
     response = requests.get(
-        url, headers=headers, stream=True, allow_redirects=True, timeout=30
+        url, headers=get_headers, stream=True, allow_redirects=True, timeout=30
     )
     www = response.headers.get(
         "WWW-Authenticate", ""
@@ -530,11 +546,13 @@ def _download_file(
         # for known hosts. __get_vault_access__ handles reading the refresh
         # token and exchanging it; errors are translated to DownloadAuthError
         # for user-friendly CLI output.
-        vault_token = __get_vault_access__(url, vault_token_file, auth_url, client_id)
-        headers["Authorization"] = f"Bearer {vault_token}"
+        vault_token = __get_vault_access__(
+            url, vault_token_file, auth_url, client_id, headers=req_headers
+        )
+        get_headers["Authorization"] = f"Bearer {vault_token}"
 
         # Retry with token
-        response = requests.get(url, headers=headers, stream=True, timeout=30)
+        response = requests.get(url, headers=get_headers, stream=True, timeout=30)
 
         # Map common auth failures to friendly messages
         if response.status_code == 401:
@@ -866,6 +884,7 @@ def _download_files(
     manifest_context=None,
     validate_checksum: bool = False,
     checksums: dict | None = None,
+    headers: dict | None = None,
 ) -> None:
     """Download multiple files from the databus.
 
@@ -903,35 +922,44 @@ def _download_files(
             validate_checksum=validate_checksum,
             expected_checksum=expected,
             manifest_context=manifest_context,
+            headers=headers,
         )
 
 
-def _get_sparql_query_of_collection(uri: str, databus_key: str | None = None) -> str:
+def _get_sparql_query_of_collection(
+    uri: str, databus_key: str | None = None, headers: dict | None = None
+) -> str:
     """Get SPARQL query of collection members from databus collection URI.
 
     Args:
         uri: The full databus collection URI.
         databus_key: Optional Databus API key for authentication on protected resources.
+        headers: Optional custom HTTP headers to send with the request.
 
     Returns:
         SPARQL query string to get download URLs of all files in the collection.
     """
-    headers = {"Accept": "text/sparql"}
+    req_headers = get_default_headers(headers)
+    if "Accept" not in req_headers:
+        req_headers["Accept"] = "text/sparql"
     if databus_key is not None:
-        headers["X-API-KEY"] = databus_key
+        req_headers["X-API-KEY"] = databus_key
 
-    response = requests.get(uri, headers=headers, timeout=30)
+    response = requests.get(uri, headers=req_headers, timeout=30)
     response.raise_for_status()
     return response.text
 
 
-def _query_sparql_endpoint(endpoint_url, query, databus_key=None) -> dict:
+def _query_sparql_endpoint(
+    endpoint_url, query, databus_key=None, headers=None
+) -> dict:
     """Query a SPARQL endpoint and return results in JSON format.
 
     Args:
         endpoint_url: The URL of the SPARQL endpoint.
         query: The SPARQL query string.
         databus_key: Optional API key for authentication.
+        headers: Optional custom HTTP headers to send with the SPARQL query.
 
     Returns:
         Dictionary containing the query results.
@@ -940,14 +968,30 @@ def _query_sparql_endpoint(endpoint_url, query, databus_key=None) -> dict:
     sparql.method = "POST"
     sparql.setQuery(query)
     sparql.setReturnFormat(JSON)
-    if databus_key is not None:
-        sparql.setCustomHttpHeaders({"X-API-KEY": databus_key})
+
+    req_headers = get_default_headers(headers)
+    if hasattr(sparql, "setCustomHttpHeaders"):
+        custom = dict(req_headers)
+        if databus_key is not None:
+            custom["X-API-KEY"] = databus_key
+        sparql.setCustomHttpHeaders(custom)
+    if hasattr(sparql, "addCustomHttpHeader"):
+        for k, v in req_headers.items():
+            if k.lower() == "user-agent" and hasattr(sparql, "agent"):
+                sparql.agent = v
+            sparql.addCustomHttpHeader(k, v)
+        if databus_key is not None:
+            sparql.addCustomHttpHeader("X-API-KEY", databus_key)
+    elif hasattr(sparql, "agent"):
+        for k, v in req_headers.items():
+            if k.lower() == "user-agent":
+                sparql.agent = v
     results = sparql.query().convert()
     return results
 
 
 def _get_file_download_urls_from_sparql_query(
-    endpoint_url, query, databus_key=None
+    endpoint_url, query, databus_key=None, headers=None
 ) -> List[str]:
     """Execute a SPARQL query to get databus file download URLs.
 
@@ -955,11 +999,14 @@ def _get_file_download_urls_from_sparql_query(
         endpoint_url: The URL of the SPARQL endpoint.
         query: The SPARQL query string.
         databus_key: Optional API key for authentication.
+        headers: Optional custom HTTP headers to send with the SPARQL query.
 
     Returns:
         List of file download URLs.
     """
-    result_dict = _query_sparql_endpoint(endpoint_url, query, databus_key=databus_key)
+    result_dict = _query_sparql_endpoint(
+        endpoint_url, query, databus_key=databus_key, headers=headers
+    )
 
     bindings = result_dict.get("results", {}).get("bindings")
     if not isinstance(bindings, list):
@@ -983,7 +1030,11 @@ def _get_file_download_urls_from_sparql_query(
 
 
 def __get_vault_access__(
-    download_url: str, token_file: str, auth_url: str, client_id: str
+    download_url: str,
+    token_file: str,
+    auth_url: str,
+    client_id: str,
+    headers: dict | None = None,
 ) -> str:
     """
     Get Vault access token for a protected databus download.
@@ -998,6 +1049,8 @@ def __get_vault_access__(
     if len(refresh_token) < 80:
         print(f"Warning: token from {token_file} is short (<80 chars)")
 
+    post_headers = get_default_headers(headers)
+
     # 2. Refresh token -> access token
     resp = requests.post(
         auth_url,
@@ -1006,6 +1059,7 @@ def __get_vault_access__(
             "grant_type": "refresh_token",
             "refresh_token": refresh_token,
         },
+        headers=post_headers,
         timeout=30,
     )
     resp.raise_for_status()
@@ -1030,6 +1084,7 @@ def __get_vault_access__(
             "subject_token": access_token,
             "audience": audience,
         },
+        headers=post_headers,
         timeout=30,
     )
     resp.raise_for_status()
@@ -1054,6 +1109,7 @@ def _download_collection(
     graph_mode: str = None,
     manifest_context=None,
     validate_checksum: bool = False,
+    headers: dict | None = None,
 ) -> None:
     """Download all files in a databus collection.
 
@@ -1072,15 +1128,15 @@ def _download_collection(
         graph_mode: Optional graph sidecar mode. Currently supports 'download-url'.
         validate_checksum: Whether to validate checksums after downloading.
     """
-    query = _get_sparql_query_of_collection(uri, databus_key=databus_key)
+    query = _get_sparql_query_of_collection(uri, databus_key=databus_key, headers=headers)
     file_urls = _get_file_download_urls_from_sparql_query(
-        endpoint, query, databus_key=databus_key
+        endpoint, query, databus_key=databus_key, headers=headers
     )
 
     # If checksum validation requested, attempt to build url->checksum mapping
     checksums: dict = {}
     if validate_checksum:
-        checksums = _resolve_checksums_for_urls(list(file_urls), databus_key)
+        checksums = _resolve_checksums_for_urls(list(file_urls), databus_key, headers=headers)
 
     _download_files(
         list(file_urls),
@@ -1097,6 +1153,7 @@ def _download_collection(
         manifest_context=manifest_context,
         validate_checksum=validate_checksum,
         checksums=checksums if checksums else None,
+        headers=headers,
     )
 
 
@@ -1114,6 +1171,7 @@ def _download_version(
     graph_mode: str = None,
     manifest_context=None,
     validate_checksum: bool = False,
+    headers: dict | None = None,
 ) -> None:
     """Download all files in a databus artifact version.
 
@@ -1131,7 +1189,7 @@ def _download_version(
         graph_mode: Optional graph sidecar mode. Currently supports 'download-url'.
         validate_checksum: Whether to validate checksums after downloading.
     """
-    json_str = fetch_databus_jsonld(uri, databus_key=databus_key)
+    json_str = fetch_databus_jsonld(uri, databus_key=databus_key, headers=headers)
     file_urls = _get_file_download_urls_from_artifact_jsonld(json_str)
     # build url -> checksum mapping from JSON-LD when available
     checksums: dict = {}
@@ -1173,6 +1231,7 @@ def _download_artifact(
     graph_mode: str = None,
     manifest_context=None,
     validate_checksum: bool = False,
+    headers: dict | None = None,
 ) -> None:
     """Download files in a databus artifact.
 
@@ -1191,13 +1250,13 @@ def _download_artifact(
         graph_mode: Optional graph sidecar mode. Currently supports 'download-url'.
         validate_checksum: Whether to validate checksums after downloading.
     """
-    json_str = fetch_databus_jsonld(uri, databus_key=databus_key)
+    json_str = fetch_databus_jsonld(uri, databus_key=databus_key, headers=headers)
     versions = _get_databus_versions_of_artifact(json_str, all_versions=all_versions)
     if isinstance(versions, str):
         versions = [versions]
     for version_uri in versions:
         print(f"Downloading version: {version_uri}")
-        json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key)
+        json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key, headers=headers)
         file_urls = _get_file_download_urls_from_artifact_jsonld(json_str)
         # extract checksums for this version
         checksums: dict = {}
@@ -1390,6 +1449,7 @@ def download(
     graph_mode=None,
     validate_checksum: bool = False,
     manifest_context=None,
+    headers=None,
 ) -> None:
     """Download datasets from databus.
 
@@ -1412,6 +1472,7 @@ def download(
         validate_checksum: Whether to validate checksums after downloading.
     """
     _validate_graph_mode(graph_mode)
+    parsed_headers = parse_headers(headers) if headers else None
     for databusURI in databusURIs:
         host, account, group, artifact, version, file = get_databus_id_parts_from_file_url(databusURI)
 
@@ -1442,12 +1503,13 @@ def download(
                     graph_mode=graph_mode,
                     manifest_context=manifest_context,
                     validate_checksum=validate_checksum,
+                    headers=parsed_headers,
                 )
             elif file is not None:
                 print(f"Downloading file: {databusURI}")
                 # Try to fetch expected checksum from the parent Version metadata
                 expected = (
-                    get_expected_checksum(databusURI, databus_key)
+                    get_expected_checksum(databusURI, databus_key, headers=parsed_headers)
                     if validate_checksum
                     else None
                 )
@@ -1468,6 +1530,7 @@ def download(
                     manifest_context=manifest_context,
                     validate_checksum=validate_checksum,
                     expected_checksum=expected,
+                    headers=parsed_headers,
                 )
             elif version is not None:
                 print(f"Downloading version: {databusURI}")
@@ -1485,6 +1548,7 @@ def download(
                     graph_mode=graph_mode,
                     manifest_context=manifest_context,
                     validate_checksum=validate_checksum,
+                    headers=parsed_headers,
                 )
             elif artifact is not None:
                 print(
@@ -1505,6 +1569,7 @@ def download(
                     graph_mode=graph_mode,
                     manifest_context=manifest_context,
                     validate_checksum=validate_checksum,
+                    headers=parsed_headers,
                 )
             elif group is not None and group != "collections":
                 print(
@@ -1525,6 +1590,7 @@ def download(
                     graph_mode=graph_mode,
                     manifest_context=manifest_context,
                     validate_checksum=validate_checksum,
+                    headers=parsed_headers,
                 )
             elif account is not None:
                 print("accountId not supported yet")  # TODO
@@ -1541,13 +1607,13 @@ def download(
             if uri_endpoint is None:  # endpoint is required for queries (--databus)
                 raise ValueError("No endpoint given for query")
             res = _get_file_download_urls_from_sparql_query(
-                uri_endpoint, databusURI, databus_key=databus_key
+                uri_endpoint, databusURI, databus_key=databus_key, headers=parsed_headers
             )
 
             # If checksum validation requested, try to build url->checksum mapping
             checksums: dict = {}
             if validate_checksum:
-                checksums = _resolve_checksums_for_urls(res, databus_key)
+                checksums = _resolve_checksums_for_urls(res, databus_key, headers=parsed_headers)
                 if not checksums:
                     print(
                         "WARNING: Checksum validation enabled but no checksums found for query results."
@@ -1568,4 +1634,5 @@ def download(
                 manifest_context=manifest_context,
                 validate_checksum=validate_checksum,
                 checksums=checksums if checksums else None,
+                headers=parsed_headers,
             )
diff --git a/databusclient/api/utils.py b/databusclient/api/utils.py
index df4f49d..698b43a 100644
--- a/databusclient/api/utils.py
+++ b/databusclient/api/utils.py
@@ -4,11 +4,78 @@
 `download`, `deploy` and `delete` modules.
 """
 
-from typing import Optional, Tuple
+from typing import Any, Dict, Iterable, Optional, Tuple, Union
 import hashlib
+import os
 import requests
 
 
+def parse_headers(
+    headers: Union[Dict[str, Any], Iterable[str], None],
+) -> Dict[str, str]:
+    """Parse HTTP headers from dict, list, tuple, or None into a clean dict of strings.
+
+    Supports:
+    - Dict of headers: e.g. {"User-Agent": "MyBot/1.0", "Accept": "*/*"}
+    - Iterable of 'Name: Value' strings: e.g. ["User-Agent: MyBot/1.0"]
+
+    Args:
+        headers: Dict or iterable of header strings, or None.
+
+    Returns:
+        Dict[str, str]: Parsed headers dictionary.
+    """
+    if not headers:
+        return {}
+
+    if isinstance(headers, dict):
+        return {str(k).strip(): str(v).strip() for k, v in headers.items()}
+
+    parsed: Dict[str, str] = {}
+    for h in headers:
+        if isinstance(h, str):
+            if ":" not in h:
+                raise ValueError(
+                    f"Invalid header format '{h}'. Expected 'Header-Name: Header-Value'."
+                )
+            k, v = h.split(":", 1)
+            parsed[k.strip()] = v.strip()
+        else:
+            raise ValueError(f"Unexpected header type: {type(h).__name__}")
+    return parsed
+
+
+def get_default_headers(
+    headers: Union[Dict[str, Any], Iterable[str], None] = None,
+) -> Dict[str, str]:
+    """Return headers dictionary with default User-Agent and any overrides applied.
+
+    If User-Agent is not provided in headers, uses the DATABUS_USER_AGENT
+    environment variable if set, otherwise falls back to databusclient/<version>.
+
+    Args:
+        headers: User-supplied headers (dict or iterable of strings).
+
+    Returns:
+        Dict[str, str]: Resolved headers dictionary.
+    """
+    custom = parse_headers(headers)
+    has_user_agent = any(k.lower() == "user-agent" for k in custom.keys())
+
+    res: Dict[str, str] = {}
+    if not has_user_agent:
+        env_ua = os.environ.get("DATABUS_USER_AGENT")
+        if env_ua:
+            res["User-Agent"] = env_ua
+        else:
+            from databusclient.version import __version__
+
+            res["User-Agent"] = f"databusclient/{__version__}"
+
+    res.update(custom)
+    return res
+
+
 def get_databus_id_parts_from_file_url(
     uri: str,
 ) -> Tuple[
@@ -40,21 +107,28 @@ def get_databus_id_parts_from_file_url(
     return tuple(parts[:6])  # return only the first 6 parts
 
 
-def fetch_databus_jsonld(uri: str, databus_key: str | None = None) -> str:
+def fetch_databus_jsonld(
+    uri: str,
+    databus_key: str | None = None,
+    headers: Union[Dict[str, Any], Iterable[str], None] = None,
+) -> str:
     """Fetch the JSON-LD representation of a Databus resource.
 
     Args:
         uri: Full Databus resource URI.
         databus_key: Optional API key for protected resources.
+        headers: Optional custom headers to send with the request.
 
     Returns:
         The response body as a string containing JSON-LD.
     """
 
-    headers = {"Accept": "application/ld+json"}
+    req_headers = get_default_headers(headers)
+    if "Accept" not in req_headers:
+        req_headers["Accept"] = "application/ld+json"
     if databus_key is not None:
-        headers["X-API-KEY"] = databus_key
-    response = requests.get(uri, headers=headers, timeout=30)
+        req_headers["X-API-KEY"] = databus_key
+    response = requests.get(uri, headers=req_headers, timeout=30)
     response.raise_for_status()
 
     return response.text
diff --git a/databusclient/cli.py b/databusclient/cli.py
index 3127186..8ddf2be 100644
--- a/databusclient/cli.py
+++ b/databusclient/cli.py
@@ -8,6 +8,7 @@
 import databusclient.api.deploy as api_deploy
 from databusclient.api.delete import delete as api_delete
 from databusclient.api.download import download as api_download, DownloadAuthError
+from databusclient.api.utils import parse_headers
 from databusclient.manifest.context import ManifestContext
 from databusclient.manifest.writer import ManifestWriter
 from databusclient.manifest.replay import ManifestReplayError, replay_manifest, load_manifest
@@ -66,6 +67,13 @@ def app():
     "webdav_url",
     help="WebDAV URL (e.g., https://cloud.example.com/remote.php/webdav)",
 )
+@click.option(
+    "-H",
+    "--header",
+    "headers",
+    multiple=True,
+    help="Custom HTTP header(s) in 'Name: Value' format (e.g. -H 'User-Agent: MyBot/1.0').",
+)
 @click.option(
     "--manifest",
     "manifest_path",
@@ -88,6 +96,7 @@ def deploy(
     path,
     distributions: List[str],
     manifest_path,
+    headers: List[str] = (),
 ):
     """
     Flexible deploy to Databus command supporting three modes:\n
@@ -106,10 +115,15 @@ def deploy(
             "Invalid combination: when using WebDAV/Nextcloud mode, please provide --webdav-url, --remote, and --path together."
         )
 
+    try:
+        parsed_headers = parse_headers(headers) if headers else None
+    except ValueError as e:
+        raise click.BadParameter(str(e), param_hint="--header")
+
     manifest_context = None
     if manifest_path:
         manifest_context = ManifestContext(command="deploy")
-        manifest_context.record_params({
+        params = {
             "version_id": version_id,
             "title": title,
             "abstract": abstract,
@@ -117,7 +131,13 @@ def deploy(
             "license_url": license_url,
             "distributions": list(distributions) if distributions else [],
             "metadata_file": metadata_file,
-        })
+        }
+        if parsed_headers:
+            params["headers"] = {
+                k: ("***" if k.lower() in ("authorization", "x-api-key", "token") else v)
+                for k, v in parsed_headers.items()
+            }
+        manifest_context.record_params(params)
 
     def _write_manifest():
         if manifest_path and manifest_context is not None:
@@ -146,7 +166,7 @@ def _write_manifest():
             if manifest_context:
                 manifest_context.replay_params["deploy_mode"] = "classic"
                 manifest_context.replay_params["resolved_distributions"] = dataid["@graph"][-1].get("distribution", [])
-            api_deploy.deploy(dataid=dataid, api_key=apikey)
+            api_deploy.deploy(dataid=dataid, api_key=apikey, headers=parsed_headers)
             if manifest_context:
                 for dist in distributions:
                     url = str(dist).split("|")[0]
@@ -169,7 +189,14 @@ def _write_manifest():
                 manifest_context.replay_params["deploy_mode"] = "metadata"
                 manifest_context.replay_params["resolved_metadata"] = metadata
             api_deploy.deploy_from_metadata(
-                metadata, version_id, title, abstract, description, license_url, apikey
+                metadata,
+                version_id,
+                title,
+                abstract,
+                description,
+                license_url,
+                apikey,
+                headers=parsed_headers,
             )
             if manifest_context:
                 for entry in metadata:
@@ -205,7 +232,14 @@ def _write_manifest():
         try:
             metadata = webdav.upload_to_webdav(distributions, remote, path, webdav_url)
             api_deploy.deploy_from_metadata(
-                metadata, version_id, title, abstract, description, license_url, apikey
+                metadata,
+                version_id,
+                title,
+                abstract,
+                description,
+                license_url,
+                apikey,
+                headers=parsed_headers,
             )
             if manifest_context:
                 for entry in metadata:
@@ -318,6 +352,13 @@ def _write_manifest():
     default=None,
     help="Write a JSON-LD manifest of this operation to PATH (e.g. --manifest manifest.jsonld).",
 )
+@click.option(
+    "-H",
+    "--header",
+    "headers",
+    multiple=True,
+    help="Custom HTTP header(s) in 'Name: Value' format (e.g. -H 'User-Agent: MyBot/1.0').",
+)
 @click.option(
     "--validate-checksum", is_flag=True, help="Validate checksums of downloaded files"
 )
@@ -337,6 +378,7 @@ def download(
     graph_mode,
     validate_checksum,
     manifest_path,
+    headers: List[str] = (),
 ):
     """
     Download datasets from databus, optionally using vault access if vault options are provided.
@@ -349,6 +391,11 @@ def download(
     elif databus_key:
         auth_method = "databus_key"
 
+    try:
+        parsed_headers = parse_headers(headers) if headers else None
+    except ValueError as e:
+        raise click.BadParameter(str(e), param_hint="--header")
+
     manifest_context = None
     if manifest_path:
         manifest_context = ManifestContext(
@@ -357,7 +404,7 @@ def download(
             auth_method=auth_method,
         )
         # Record safe replay params — sensitive fields excluded
-        manifest_context.record_params({
+        safe_params = {
             "databusURIs": list(databusuris),
             "compression": compression,
             "convert_format": convert_format,
@@ -368,7 +415,13 @@ def download(
             "validate_checksum": validate_checksum,
             "authurl": authurl,
             "clientid": clientid,
-        })
+        }
+        if parsed_headers:
+            safe_params["headers"] = {
+                k: ("***" if k.lower() in ("authorization", "x-api-key", "token") else v)
+                for k, v in parsed_headers.items()
+            }
+        manifest_context.record_params(safe_params)
     try:
         api_download(
             localDir=localdir,
@@ -386,6 +439,7 @@ def download(
             graph_mode=graph_mode,
             validate_checksum=validate_checksum,
             manifest_context=manifest_context,
+            headers=parsed_headers,
         )
     except DownloadAuthError as e:
         if manifest_context:
@@ -422,13 +476,27 @@ def download(
 @click.option(
     "--force", is_flag=True, help="Force deletion without confirmation prompt"
 )
+@click.option(
+    "-H",
+    "--header",
+    "headers",
+    multiple=True,
+    help="Custom HTTP header(s) in 'Name: Value' format (e.g. -H 'User-Agent: MyBot/1.0').",
+)
 @click.option(
     "--manifest",
     "manifest_path",
     default=None,
     help="Write a JSON-LD manifest of this operation to PATH.",
 )
-def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool, manifest_path: str):
+def delete(
+    databusuris: List[str],
+    databus_key: str,
+    dry_run: bool,
+    force: bool,
+    manifest_path: str,
+    headers: List[str] = (),
+):
     """
     Delete a dataset from the databus.
 
@@ -436,13 +504,24 @@ def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool,
     Will recursively delete all data associated with the dataset.
     """
 
+    try:
+        parsed_headers = parse_headers(headers) if headers else None
+    except ValueError as e:
+        raise click.BadParameter(str(e), param_hint="--header")
+
     manifest_context = None
     if manifest_path:
         manifest_context = ManifestContext(command="delete")
-        manifest_context.record_params({
+        params = {
             "databusURIs": list(databusuris),
             "dry_run": dry_run,
-        })
+        }
+        if parsed_headers:
+            params["headers"] = {
+                k: ("***" if k.lower() in ("authorization", "x-api-key", "token") else v)
+                for k, v in parsed_headers.items()
+            }
+        manifest_context.record_params(params)
 
     try:
         api_delete(
@@ -451,6 +530,7 @@ def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool,
             dry_run=dry_run,
             force=force,
             manifest_context=manifest_context,
+            headers=parsed_headers,
         )
     except Exception as exc:
         if manifest_context:
diff --git a/databusclient/workflow/steps.py b/databusclient/workflow/steps.py
index 49d6061..23bd088 100644
--- a/databusclient/workflow/steps.py
+++ b/databusclient/workflow/steps.py
@@ -69,21 +69,24 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None:
         capture_context = context.manifest_context or ManifestContext(command="download")
         files_before = len(capture_context.files)
 
-        api_download(
-            localDir=local_dir,
-            endpoint=resolved.get("databus"),
-            databusURIs=uris,
-            token=resolved.get("vault_token"),
-            databus_key=resolved.get("databus_key"),
-            all_versions=resolved.get("all_versions", False),
-            compression=resolved.get("convert_to") or resolved.get("compression"),
-            convert_format=resolved.get("format"),
-            graph_name=resolved.get("graph_name"),
-            base_uri=resolved.get("base_uri"),
-            graph_mode=resolved.get("graph_mode"),
-            validate_checksum=resolved.get("validate_checksum", False),
-            manifest_context=capture_context,
-        )
+        dl_kwargs = {
+            "localDir": local_dir,
+            "endpoint": resolved.get("databus"),
+            "databusURIs": uris,
+            "token": resolved.get("vault_token"),
+            "databus_key": resolved.get("databus_key"),
+            "all_versions": resolved.get("all_versions", False),
+            "compression": resolved.get("convert_to") or resolved.get("compression"),
+            "convert_format": resolved.get("format"),
+            "graph_name": resolved.get("graph_name"),
+            "base_uri": resolved.get("base_uri"),
+            "graph_mode": resolved.get("graph_mode"),
+            "validate_checksum": resolved.get("validate_checksum", False),
+            "manifest_context": capture_context,
+        }
+        if resolved.get("headers"):
+            dl_kwargs["headers"] = resolved["headers"]
+        api_download(**dl_kwargs)
 
         new_entries = capture_context.files[files_before:]
         resolved_urls = [e["url"] for e in new_entries if e.get("status") == "success"]
@@ -190,7 +193,10 @@ def _run_classic_mode(self, resolved: Dict[str, Any], name: str) -> list:
             license_url=resolved["license"],
             distributions=files,
         )
-        api_deploy_call(dataid=dataid, api_key=resolved["api_key"])
+        deploy_kwargs = {"dataid": dataid, "api_key": resolved["api_key"]}
+        if resolved.get("headers"):
+            deploy_kwargs["headers"] = resolved["headers"]
+        api_deploy_call(**deploy_kwargs)
         return files
 
     def _run_webdav_mode(self, resolved: Dict[str, Any], name: str) -> list:
@@ -206,15 +212,27 @@ def _run_webdav_mode(self, resolved: Dict[str, Any], name: str) -> list:
         metadata = webdav.upload_to_webdav(
             local_files, resolved["remote"], resolved["path"], resolved["webdav_url"]
         )
-        deploy_from_metadata(
-            metadata,
-            resolved["version_id"],
-            resolved["title"],
-            resolved["abstract"],
-            resolved["description"],
-            resolved["license"],
-            resolved["api_key"],
-        )
+        if resolved.get("headers"):
+            deploy_from_metadata(
+                metadata,
+                resolved["version_id"],
+                resolved["title"],
+                resolved["abstract"],
+                resolved["description"],
+                resolved["license"],
+                resolved["api_key"],
+                headers=resolved["headers"],
+            )
+        else:
+            deploy_from_metadata(
+                metadata,
+                resolved["version_id"],
+                resolved["title"],
+                resolved["abstract"],
+                resolved["description"],
+                resolved["license"],
+                resolved["api_key"],
+            )
         return [entry.get("url", "") for entry in metadata]
 
 
@@ -241,13 +259,16 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None:
         if not api_key:
             raise StepValidationError(f"Step '{name}': delete step requires 'api_key'.")
 
-        api_delete(
-            databusURIs=uris,
-            databus_key=api_key,
-            dry_run=resolved.get("dry_run", False),
-            force=True,
-            manifest_context=context.manifest_context,
-        )
+        delete_kwargs = {
+            "databusURIs": uris,
+            "databus_key": api_key,
+            "dry_run": resolved.get("dry_run", False),
+            "force": True,
+            "manifest_context": context.manifest_context,
+        }
+        if resolved.get("headers"):
+            delete_kwargs["headers"] = resolved["headers"]
+        api_delete(**delete_kwargs)
 
         context.set_output(name, "output_files", [])
 
diff --git a/doc/cli-usage.md b/doc/cli-usage.md
index 1325220..3e4c297 100644
--- a/doc/cli-usage.md
+++ b/doc/cli-usage.md
@@ -40,6 +40,8 @@ databusclient download $DOWNLOADTARGET
   - Required when converting CSV/TSV to RDF triples. Used as the base for constructing subject URIs from CSV row identifiers. Example: `--format ntriples --base-uri https://example.org/data/`.
 - `--validate-checksum`
   - Validates the checksums of downloaded files against the checksums provided by the Databus. If a checksum does not match, an error is raised and the file is deleted.
+- `-H, --header`
+  - Custom HTTP header(s) to send with requests, in `'Name: Value'` format. Can be specified multiple times (e.g. `-H 'User-Agent: MyBot/1.0 (contact@example.org)'`). Required when accessing servers (such as Wikidata/Wikimedia) that enforce specific User-Agent policies. You can also configure the default User-Agent via the `DATABUS_USER_AGENT` environment variable.
 
 **Help and further information on download command:**
 ```bash
@@ -78,6 +80,11 @@ databusclient download https://databus.dbpedia.org/dbpedia/collections/dbpedia-s
 databusclient download 'PREFIX dcat: <http://www.w3.org/ns/dcat#> SELECT ?x WHERE { ?sub dcat:downloadURL ?x . } LIMIT 10' --databus https://databus.dbpedia.org/sparql
 ```
 
+**Download with Custom Headers / User-Agent**: send specific headers required by servers (such as Wikidata)
+```bash
+databusclient download https://databus.dbpedia.org/dbpedia/dbpedia-wikidata-kg-dump -H "User-Agent: MyBot/1.0 (contact@example.org)"
+```
+
 **Download with Compression Conversion**: download files and convert compression format on-the-fly. Source compression is auto-detected from the file extension.
 ```bash
 # Convert all compressed files to gzip format
diff --git a/doc/module-usage.md b/doc/module-usage.md
index 68a8282..491352a 100644
--- a/doc/module-usage.md
+++ b/doc/module-usage.md
@@ -80,3 +80,19 @@ deploy(dataset, "mysterious API key")
 ```
 
 The API key can be found or generated in the Databus account settings.
+
+<a id="module-download"></a>
+### Download with Custom Headers
+
+When querying or downloading datasets from servers that require specific HTTP headers (such as Wikidata requiring a custom User-Agent):
+
+```python
+from databusclient import download
+
+download(
+    localDir="./downloads",
+    endpoint=None,
+    databusURIs=["https://databus.dbpedia.org/dbpedia/dbpedia-wikidata-kg-dump"],
+    headers={"User-Agent": "MyBot/1.0 (https://example.org/bot; contact@example.org)"},
+)
+```
diff --git a/tests/test_headers.py b/tests/test_headers.py
new file mode 100644
index 0000000..8fc6cb2
--- /dev/null
+++ b/tests/test_headers.py
@@ -0,0 +1,307 @@
+"""Tests for overriding HTTP headers and User-Agent configuration."""
+
+import pytest
+from click.testing import CliRunner
+
+from databusclient.api.utils import parse_headers, get_default_headers, fetch_databus_jsonld
+from databusclient.api.download import (
+    _download_file,
+    _query_sparql_endpoint,
+)
+from databusclient.cli import app
+from databusclient.version import __version__
+from databusclient.workflow.context import StepContext
+from databusclient.workflow.steps import DownloadStep
+
+
+def test_parse_headers_none_and_empty():
+    assert parse_headers(None) == {}
+    assert parse_headers([]) == {}
+    assert parse_headers({}) == {}
+
+
+def test_parse_headers_dict():
+    input_dict = {"User-Agent": "MyBot/1.0", "X-Custom": "val123"}
+    parsed = parse_headers(input_dict)
+    assert parsed == {"User-Agent": "MyBot/1.0", "X-Custom": "val123"}
+    # Verify non-string types get stringified
+    assert parse_headers({"X-Num": 42}) == {"X-Num": "42"}
+
+
+def test_parse_headers_iterable_strings():
+    header_list = [
+        "User-Agent: MyBot/1.0",
+        "Authorization: Bearer token123",
+        "Accept: application/json",
+    ]
+    parsed = parse_headers(header_list)
+    assert parsed == {
+        "User-Agent": "MyBot/1.0",
+        "Authorization": "Bearer token123",
+        "Accept": "application/json",
+    }
+
+
+def test_parse_headers_with_colons_in_value():
+    headers = [
+        "User-Agent: MyBot/1.0 (https://example.org/bot; contact: admin@example.org)",
+        "X-Custom-Time: 2026-10-25T10:00:00Z",
+    ]
+    parsed = parse_headers(headers)
+    assert (
+        parsed["User-Agent"]
+        == "MyBot/1.0 (https://example.org/bot; contact: admin@example.org)"
+    )
+    assert parsed["X-Custom-Time"] == "2026-10-25T10:00:00Z"
+
+
+def test_parse_headers_invalid_format():
+    with pytest.raises(ValueError, match="Invalid header format"):
+        parse_headers(["InvalidHeaderWithoutColon"])
+
+    with pytest.raises(ValueError, match="Unexpected header type"):
+        parse_headers([123])
+
+
+def test_get_default_headers_without_overrides(monkeypatch):
+    monkeypatch.delenv("DATABUS_USER_AGENT", raising=False)
+    headers = get_default_headers()
+    assert headers["User-Agent"] == f"databusclient/{__version__}"
+
+
+def test_get_default_headers_env_var_override(monkeypatch):
+    monkeypatch.setenv("DATABUS_USER_AGENT", "EnvBot/2.0")
+    headers = get_default_headers()
+    assert headers["User-Agent"] == "EnvBot/2.0"
+
+
+def test_get_default_headers_explicit_override():
+    headers = get_default_headers({"User-Agent": "WikidataBot/1.0"})
+    assert headers["User-Agent"] == "WikidataBot/1.0"
+
+
+def test_get_default_headers_case_insensitive_user_agent():
+    headers = get_default_headers(["user-agent: CustomCaseBot/1.0"])
+    # Should not duplicate User-Agent
+    assert any(k.lower() == "user-agent" for k in headers)
+    ua_keys = [k for k in headers if k.lower() == "user-agent"]
+    assert len(ua_keys) == 1
+    assert headers[ua_keys[0]] == "CustomCaseBot/1.0"
+
+
+def test_get_default_headers_preserves_custom_headers():
+    headers = get_default_headers({"X-App-ID": "test-app", "Accept-Language": "de"})
+    assert headers["X-App-ID"] == "test-app"
+    assert headers["Accept-Language"] == "de"
+    assert "User-Agent" in headers
+
+
+def test_fetch_databus_jsonld_sends_headers(monkeypatch):
+    captured_headers = {}
+
+    class FakeResponse:
+        status_code = 200
+        text = '{"@id": "https://databus.dbpedia.org/test"}'
+
+        def raise_for_status(self):
+            pass
+
+    def fake_get(url, headers=None, timeout=30):
+        captured_headers.update(headers or {})
+        return FakeResponse()
+
+    monkeypatch.setattr("requests.get", fake_get)
+    fetch_databus_jsonld(
+        "https://databus.dbpedia.org/test",
+        headers={"User-Agent": "JsonLdBot/1.0", "X-Trace": "abc"},
+    )
+
+    assert captured_headers.get("User-Agent") == "JsonLdBot/1.0"
+    assert captured_headers.get("X-Trace") == "abc"
+    assert captured_headers.get("Accept") == "application/ld+json"
+
+
+def test_download_file_sends_custom_headers(tmp_path, monkeypatch):
+    head_headers = {}
+    get_headers = {}
+
+    class FakeHead:
+        status_code = 200
+        headers = {}
+
+    class FakeGet:
+        status_code = 200
+        headers = {"content-length": "13"}
+
+        def iter_content(self, chunk_size):
+            yield b"dummy content"
+
+        def raise_for_status(self):
+            pass
+
+    def fake_head(url, headers=None, timeout=30, allow_redirects=False):
+        head_headers.update(headers or {})
+        return FakeHead()
+
+    def fake_get(url, headers=None, stream=True, allow_redirects=True, timeout=30):
+        get_headers.update(headers or {})
+        return FakeGet()
+
+    monkeypatch.setattr("requests.head", fake_head)
+    monkeypatch.setattr("requests.get", fake_get)
+
+    target_url = "https://databus.dbpedia.org/org/group/art/1.0/file.ttl"
+    _download_file(
+        url=target_url,
+        localDir=str(tmp_path),
+        headers={"User-Agent": "CustomDownloader/1.0", "X-Custom-Auth": "secret123"},
+    )
+
+    assert head_headers.get("User-Agent") == "CustomDownloader/1.0"
+    assert head_headers.get("X-Custom-Auth") == "secret123"
+    assert get_headers.get("User-Agent") == "CustomDownloader/1.0"
+    assert get_headers.get("X-Custom-Auth") == "secret123"
+
+
+def test_query_sparql_endpoint_sets_headers(monkeypatch):
+    class MockSPARQL:
+        def __init__(self, endpoint):
+            self.endpoint = endpoint
+            self.method = None
+            self.customHttpHeaders = {}
+            self.agent = None
+
+        def setQuery(self, q):
+            pass
+
+        def setReturnFormat(self, f):
+            pass
+
+        def addCustomHttpHeader(self, k, v):
+            self.customHttpHeaders[k] = v
+
+        def query(self):
+            class R:
+                def convert(self):
+                    return {"results": {"bindings": []}}
+
+            return R()
+
+    monkeypatch.setattr("databusclient.api.download.SPARQLWrapper", MockSPARQL)
+
+    result = _query_sparql_endpoint(
+        "https://query.wikidata.org/sparql",
+        "SELECT * WHERE { ?s ?p ?o } LIMIT 1",
+        headers={"User-Agent": "WikidataSPARQLBot/1.0", "X-Request-ID": "req-99"},
+    )
+    assert result == {"results": {"bindings": []}}
+
+
+def test_cli_download_with_header_flag(tmp_path, monkeypatch):
+    captured_headers = None
+
+    def fake_download(*args, **kwargs):
+        nonlocal captured_headers
+        captured_headers = kwargs.get("headers")
+
+    monkeypatch.setattr("databusclient.cli.api_download", fake_download)
+
+    runner = CliRunner()
+    result = runner.invoke(
+        app,
+        [
+            "download",
+            "https://databus.dbpedia.org/test/artifact/1.0/file.ttl",
+            "--localdir",
+            str(tmp_path),
+            "-H",
+            "User-Agent: CLIBot/1.0",
+            "-H",
+            "X-Extra: value",
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert captured_headers == {
+        "User-Agent": "CLIBot/1.0",
+        "X-Extra": "value",
+    }
+
+
+def test_cli_download_invalid_header_error():
+    runner = CliRunner()
+    result = runner.invoke(
+        app,
+        [
+            "download",
+            "https://databus.dbpedia.org/test/artifact/1.0/file.ttl",
+            "-H",
+            "MalformedHeaderWithoutColon",
+        ],
+    )
+    assert result.exit_code != 0
+    assert "Invalid header format" in result.output
+
+
+def test_manifest_records_redacted_headers(tmp_path, monkeypatch):
+    def fake_download(*args, **kwargs):
+        pass
+
+    monkeypatch.setattr("databusclient.cli.api_download", fake_download)
+
+    manifest_file = tmp_path / "test-manifest.jsonld"
+    runner = CliRunner()
+    result = runner.invoke(
+        app,
+        [
+            "download",
+            "https://databus.dbpedia.org/test/artifact/1.0/file.ttl",
+            "--manifest",
+            str(manifest_file),
+            "-H",
+            "User-Agent: PublicBot/1.0",
+            "-H",
+            "Authorization: Bearer sensitive_token",
+            "-H",
+            "X-API-KEY: secret_key",
+        ],
+    )
+    assert result.exit_code == 0
+    assert manifest_file.exists()
+
+    content = manifest_file.read_text(encoding="utf-8")
+    assert "PublicBot/1.0" in content
+    assert "sensitive_token" not in content
+    assert "secret_key" not in content
+    assert "***" in content
+
+
+def test_workflow_step_executes_with_headers(monkeypatch, tmp_path):
+    received_kwargs = {}
+
+    def fake_download(*args, **kwargs):
+        received_kwargs.update(kwargs)
+
+    monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download)
+
+    step = DownloadStep()
+    ctx = StepContext()
+
+    step.run(
+        {
+            "name": "download_data",
+            "command": "download",
+            "uri": "https://databus.dbpedia.org/test/artifact/1.0/file.ttl",
+            "localdir": str(tmp_path),
+            "headers": {
+                "User-Agent": "WorkflowBot/1.0",
+                "X-Workflow-Run": "42",
+            },
+        },
+        ctx,
+    )
+
+    assert received_kwargs.get("headers") == {
+        "User-Agent": "WorkflowBot/1.0",
+        "X-Workflow-Run": "42",
+    }