feat: add DestinationHttpClient to simplify calls to target systems

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.20.2"
+version = "0.21.0"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"

src/sap_cloud_sdk/destination/__init__.py

@@ -48,6 +48,7 @@
 )
 from sap_cloud_sdk.destination.config import load_from_env_or_mount, DestinationConfig
 from sap_cloud_sdk.destination._http import TokenProvider, DestinationHttp
+from sap_cloud_sdk.destination._destination_http_client import DestinationHttpClient
 from sap_cloud_sdk.destination.client import DestinationClient
 from sap_cloud_sdk.destination.fragment_client import FragmentClient
 from sap_cloud_sdk.destination.certificate_client import CertificateClient
@@ -235,6 +236,7 @@ def create_certificate_client(
     "LocalDevDestinationClient",
     "LocalDevFragmentClient",
     "LocalDevCertificateClient",
+    "DestinationHttpClient",
     # Exceptions
     "DestinationError",
     "ClientCreationError",

src/sap_cloud_sdk/destination/_destination_http_client.py

@@ -0,0 +1,67 @@
+"""HTTP client for calling the target system described by a Destination."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+import requests
+from requests import Response
+
+from sap_cloud_sdk.destination._models import Destination, DestinationType
+
+
+class DestinationHttpClient:
+    """Wraps requests.Session to call the target system described by a Destination.
+
+    Pre-bakes headers derived from the destination — ERP headers (sap-client,
+    sap-language), URL.headers.* properties, and auth tokens.
+
+    Usage:
+
+        dest = client.get_destination("my-erp")
+        http = DestinationHttpClient(dest)
+        response = http.request("GET", "/api/resource")
+    """
+
+    def __init__(self, destination: Destination) -> None:
+        if destination.type != DestinationType.HTTP:
+            raise ValueError(
+                f"DestinationHttpClient only supports HTTP destinations, got: {destination.type}"
+            )
+
+        self._session = requests.Session()
+        self._session.headers.update(destination.get_headers())
+        self._base_url = destination.url.rstrip("/") if destination.url else ""
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+        json: Optional[Any] = None,
+        headers: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> Response:
+        """Send an HTTP request to the target system.
+
+        Args:
+            method: HTTP verb (GET, POST, PUT, PATCH, DELETE).
+            path: Path relative to the destination URL.
+            params: Optional query parameters.
+            json: Optional JSON body.
+            headers: Optional additional headers merged on top of pre-baked ones.
+            **kwargs: Passed through to requests.Session.request.
+
+        Returns:
+            requests.Response from the target system.
+        """
+        url = f"{self._base_url}/{path.lstrip('/')}" if path else self._base_url
+        return self._session.request(
+            method=method.upper(),
+            url=url,
+            params=params,
+            json=json,
+            headers=headers,
+            **kwargs,
+        )

src/sap_cloud_sdk/destination/_models.py

@@ -366,6 +366,41 @@ def to_dict(self) -> Dict[str, Any]:
                     payload[k] = v
         return payload
 
+    def get_erp_headers(self) -> Dict[str, str]:
+        """Return SAP ERP-specific headers derived from destination properties (sap-client, sap-language).
+
+        Returns:
+            Headers to inject into requests to the target system.
+        """
+        headers: Dict[str, str] = {}
+        if "sap-client" in self.properties:
+            headers["sap-client"] = self.properties["sap-client"]
+        if "sap-language" in self.properties:
+            headers["sap-language"] = self.properties["sap-language"]
+        return headers
+
+    def get_headers(self) -> Dict[str, str]:
+        """Return HTTP headers derived from this destination (ERP headers, URL.headers.* properties, and auth tokens), each overriding the previous on conflicting keys.
+
+        Returns:
+            Headers ready to inject into requests to the target system.
+        """
+        headers: Dict[str, str] = {}
+        headers.update(self.get_erp_headers())
+
+        _PREFIX = "URL.headers."
+        for key, value in self.properties.items():
+            if key.startswith(_PREFIX):
+                headers[key[len(_PREFIX) :]] = value
+
+        for token in self.auth_tokens:
+            key = token.http_header.get("key")
+            value = token.http_header.get("value")
+            if key and value:
+                headers[key] = value
+
+        return headers
+
 
 @dataclass
 class AuthToken:

src/sap_cloud_sdk/destination/client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import warnings
 from typing import List, Optional, Callable, TypeVar
 
 from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
@@ -196,12 +197,19 @@ def list_subaccount_destinations(
                 f"failed to list subaccount destinations: {e}"
             )
 
-    @record_metrics(Module.DESTINATION, Operation.DESTINATION_GET_INSTANCE_DESTINATION)
+    @record_metrics(
+        Module.DESTINATION,
+        Operation.DESTINATION_GET_INSTANCE_DESTINATION,
+        deprecated=True,
+    )
     def get_instance_destination(
         self, name: str, proxy_enabled: Optional[bool] = None
     ) -> Optional[Destination | TransparentProxyDestination]:
         """Get a destination from the service instance scope.
 
+        .. deprecated::
+            Use ``get_destination()`` instead, which automatically retrieves auth tokens via the v2 API.
+
         Args:
             name: Destination name.
             proxy_enabled: Whether to route the request through a transparent proxy (if configured).
@@ -213,6 +221,12 @@ def get_instance_destination(
         Raises:
             DestinationOperationError: If an HTTP error occurs or response parsing fails.
         """
+        warnings.warn(
+            "get_instance_destination() is deprecated. "
+            "Use get_destination() instead, which also includes automatic token retrieval.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         try:
             if self._should_use_proxy(proxy_enabled):
                 return TransparentProxyDestination.from_proxy(
@@ -226,7 +240,9 @@ def get_instance_destination(
             raise DestinationOperationError(f"failed to get destination '{name}': {e}")
 
     @record_metrics(
-        Module.DESTINATION, Operation.DESTINATION_GET_SUBACCOUNT_DESTINATION
+        Module.DESTINATION,
+        Operation.DESTINATION_GET_SUBACCOUNT_DESTINATION,
+        deprecated=True,
     )
     def get_subaccount_destination(
         self,
@@ -237,6 +253,9 @@ def get_subaccount_destination(
     ) -> Optional[Destination | TransparentProxyDestination]:
         """Get a destination from the subaccount scope with an access strategy.
 
+        .. deprecated::
+            Use ``get_destination()`` instead, which automatically retrieves auth tokens via the v2 API.
+
         Access strategies:
             - SUBSCRIBER_ONLY: Fetch only from subscriber context (tenant required)
             - PROVIDER_ONLY: Fetch only from provider context (no tenant required)
@@ -257,6 +276,12 @@ def get_subaccount_destination(
             DestinationOperationError: If tenant is missing for subscriber access strategies,
                                        on HTTP errors, or response parsing failures.
         """
+        warnings.warn(
+            "get_subaccount_destination() is deprecated. "
+            "Use get_destination() instead, which also includes automatic token retrieval.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         try:
             if self._should_use_proxy(proxy_enabled) and self._transparent_proxy:
                 return TransparentProxyDestination.from_proxy(

src/sap_cloud_sdk/destination/user-guide.md

@@ -24,7 +24,7 @@ fragment_client = create_fragment_client(instance="default")
 certificate_client = create_certificate_client(instance="default")
 
 # Instance-level read
-dest = client.get_instance_destination("my-destination")
+dest = client.get_instance_destination("my-destination")  # deprecated: use get_destination()
 fragment = fragment_client.get_instance_fragment("my-fragment")
 cert = certificate_client.get_instance_certificate("my-cert")
 
@@ -39,12 +39,12 @@ fragments = fragment_client.list_instance_fragments(tenant="tenant-subdomain")
 certificates = certificate_client.list_instance_certificates(tenant="tenant-subdomain")
 
 # Subaccount-level read: provider only (no tenant required)
-dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.PROVIDER_ONLY)
+dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.PROVIDER_ONLY)  # deprecated: use get_destination()
 fragment = fragment_client.get_subaccount_fragment("my-fragment", access_strategy=AccessStrategy.PROVIDER_ONLY)
 cert = certificate_client.get_subaccount_certificate("my-cert", access_strategy=AccessStrategy.PROVIDER_ONLY)
 
 # Subaccount-level read: subscriber-first (tenant required), fallback to provider
-dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")
+dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")  # deprecated: use get_destination()
 fragment = fragment_client.get_subaccount_fragment("my-fragment", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")
 cert = certificate_client.get_subaccount_certificate("my-cert", access_strategy=AccessStrategy.SUBSCRIBER_FIRST, tenant="tenant-subdomain")
 
@@ -126,8 +126,8 @@ The client produced by `create_client()` exposes the following operations:
 ```python
 class DestinationClient:
     # V1 Admin API - Read operations for destinations
-    def get_instance_destination(self, name: str, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...
-    def get_subaccount_destination(self, name: str, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...
+    def get_instance_destination(self, name: str, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...  # deprecated: use get_destination()
+    def get_subaccount_destination(self, name: str, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...  # deprecated: use get_destination()
     def list_instance_destinations(self, tenant: Optional[str] = None, filter: Optional[ListOptions] = None) -> PagedResult[Destination]: ...
     def list_subaccount_destinations(self, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None, filter: Optional[ListOptions] = None) -> PagedResult[Destination]: ...
 
@@ -218,6 +218,55 @@ class CertificateClient:
 - Certificate `content` should be base64-encoded. Supported certificate types include PEM, JKS, P12, etc.
 - The v2 consumption API returns tokens in the `auth_tokens` field with ready-to-use HTTP headers in `http_header` dict.
 
+## Calling Target Systems
+
+`DestinationHttpClient` wraps `requests.Session` to call the target system described by a destination. It injects headers automatically so you don't have to handle auth tokens, ERP headers, or custom destination properties manually.
+
+> **Note:** `DestinationHttpClient` requires a destination fetched via the v2 API (`get_destination()`), which returns pre-fetched auth tokens. It does not support destinations fetched with the deprecated v1 methods.
+
+### Basic Usage
+
+```python
+from sap_cloud_sdk.destination import create_client, DestinationHttpClient
+
+client = create_client(instance="default")
+dest = client.get_destination("my-erp")
+
+http = DestinationHttpClient(dest)
+response = http.request("GET", "/api/resource")
+```
+
+### What headers are pre-baked
+
+When `DestinationHttpClient` is constructed, it reads the destination and pre-bakes the following headers into every request:
+
+1. **ERP headers** — `sap-client` and `sap-language` from destination properties (if present)
+2. **`URL.headers.*` properties** — any destination property prefixed with `URL.headers.` becomes a header (e.g. `URL.headers.apiKey = secret` → `apiKey: secret`)
+3. **Auth tokens** — pre-fetched by BTP and returned in `dest.auth_tokens`; each token's `http_header` is injected directly (e.g. `Authorization: Bearer eyJ...`)
+
+Auth tokens take precedence over `URL.headers.*` properties if both set the same header key.
+
+### Per-request headers
+
+Pass `headers=` to add or override headers for a single request:
+
+```python
+response = http.request("GET", "/api/resource", headers={"X-Correlation-ID": "abc123"})
+```
+
+Per-request headers are merged on top of the pre-baked session headers.
+
+### Using `get_headers()` directly
+
+If you manage your own HTTP client, use `dest.get_headers()` to get all derived headers as a plain dict:
+
+```python
+import requests
+
+dest = client.get_destination("my-erp")
+response = requests.get(dest.url + "/api/resource", headers=dest.get_headers())
+```
+
 ## Transparent Proxy Support
 
 The destination client supports routing requests through a transparent proxy. This enables access to on-premise systems and private network resources through a proxy deployed in your Kubernetes cluster.

tests/destination/integration/destination.feature

@@ -227,6 +227,19 @@ Feature: Destination Service Integration
   #   And I clean up the instance destination "test-v2-full-options"
   #   And I clean up the instance fragment "test-v2-full-fragment"
 
+  Scenario: DestinationHttpClient sends an authenticated request using token fetched from BTP
+    Given I have a destination named "sdk-test-http-client" of type "HTTP"
+    And the destination has URL "https://httpbin.org"
+    And the destination has authentication "OAuth2ClientCredentials"
+    And the destination has OAuth2 credentials from environment
+    When I create the destination at instance level
+    Then the destination creation should be successful
+    When I fetch the destination using the v2 API
+    And I create a DestinationHttpClient from the destination
+    And I send a GET request to "/headers"
+    Then the response contains an Authorization header
+    And I clean up the instance destination "sdk-test-http-client"
+
   Scenario: Manage labels for subaccount destination
     Given I have a destination named "test-dest-labels" of type "HTTP"
     And the destination has URL "https://labels.example.com"
@@ -303,3 +316,4 @@ Feature: Destination Service Integration
   #   Then the destination creation should be successful
   #   When I get subaccount destination "test-dest-sub-isolation" with "PROVIDER_ONLY" access strategy
   #   Then the destination should not be found
+

tests/destination/integration/test_destination_bdd.py

@@ -1,6 +1,7 @@
 """BDD step definitions for Destination integration tests."""
 
 import concurrent.futures
+import os
 from typing import List, Optional
 
 import pytest
@@ -17,6 +18,7 @@
     ConsumptionOptions,
     PatchLabels,
 )
+from sap_cloud_sdk.destination._destination_http_client import DestinationHttpClient
 from sap_cloud_sdk.destination.exceptions import (
     HttpError,
     DestinationOperationError,
@@ -57,6 +59,8 @@ def __init__(self):
         self.updated_certificate_content: Optional[str] = None
         self.tenant: Optional[str] = None
         self.retrieved_labels: List[Label] = []
+        self.http_client: Optional[DestinationHttpClient] = None
+        self.http_response = None
 
 
 @pytest.fixture
@@ -1572,3 +1576,38 @@ def certificate_should_have_label(context, key, value):
         lbl.key == key and value in lbl.values
         for lbl in context.retrieved_labels
     ), f"Expected label key='{key}' value='{value}' in {context.retrieved_labels}"
+
+
+# ==================== DESTINATION HTTP CLIENT STEPS ====================
+
+@given("the destination has OAuth2 credentials from environment")
+def destination_has_oauth2_credentials(context):
+    context.destination.properties.update({
+        "clientId":        os.environ["CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTID"],
+        "clientSecret":    os.environ["CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTSECRET"],
+        "tokenServiceURL": os.environ["CLOUD_SDK_CFG_DESTINATION_DEFAULT_URL"] + "/oauth/token",
+    })
+
+
+@when("I fetch the destination using the v2 API")
+def fetch_destination_v2(context, destination_client):
+    context.retrieved_destination = destination_client.get_destination(context.destination.name)
+
+
+@when("I create a DestinationHttpClient from the destination")
+def create_http_client(context):
+    context.http_client = DestinationHttpClient(context.retrieved_destination)
+
+
+@when(parsers.parse('I send a GET request to "{path}"'))
+def send_get_request(context, path):
+    context.http_response = context.http_client.request("GET", path)
+
+
+@then("the response contains an Authorization header")
+def assert_authorization_header_present(context):
+    echoed = context.http_response.json().get("headers", {})
+    assert "Authorization" in echoed, (
+        f"Expected Authorization header in response, got: {list(echoed.keys())}. "
+        "Check that BTP returned an auth token for the destination."
+    )