This module integrates with SAP BTP Destination Service to manage destinations, fragments, and certificates at subaccount and service instance levels. It uses a Pythonic dataclass pattern for type-safe message construction.
This package is part of the SAP Cloud SDK for Python. Import and use it directly in your application.
from sap_cloud_sdk.destination import (
create_client,
create_fragment_client,
create_certificate_client,
Level,
AccessStrategy
)
# Auto-detection based on environment; in cloud mode it will load credentials
client = create_client(instance="default")
fragment_client = create_fragment_client(instance="default")
certificate_client = create_certificate_client(instance="default")
# Instance-level read
dest = client.get_instance_destination("my-destination")
fragment = fragment_client.get_instance_fragment("my-fragment")
cert = certificate_client.get_instance_certificate("my-cert")
# Subaccount-level read: provider only (no tenant required)
dest = client.get_subaccount_destination("my-destination", access_strategy=AccessStrategy.PROVIDER_ONLY)
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")
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")
# Fragment write operations with tenant (subscriber context)
new_fragment = Fragment(name="my-fragment", properties={"URL": "https://api.example.com"})
fragment_client.create_fragment(new_fragment, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
fragment_client.update_fragment(new_fragment, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
fragment_client.delete_fragment("my-fragment", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
# Fragment write operations without tenant (provider context)
fragment_client.create_fragment(new_fragment, level=Level.SUB_ACCOUNT)
fragment_client.update_fragment(new_fragment, level=Level.SUB_ACCOUNT)
fragment_client.delete_fragment("my-fragment", level=Level.SUB_ACCOUNT)
# Destination write operations with tenant (subscriber context)
new_dest = Destination(name="my-dest", type="HTTP", url="https://api.example.com")
client.create_destination(new_dest, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
client.update_destination(new_dest, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
client.delete_destination("my-dest", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
# Destination write operations without tenant (provider context)
client.create_destination(new_dest, level=Level.SUB_ACCOUNT)
client.update_destination(new_dest, level=Level.SUB_ACCOUNT)
client.delete_destination("my-dest", level=Level.SUB_ACCOUNT)
# Certificate write operations with tenant (subscriber context)
from sap_cloud_sdk.destination import create_certificate_client
from sap_cloud_sdk.destination._models import Certificate
new_cert = Certificate(name="my-cert.pem", content="base64-encoded-content", type="PEM")
certificate_client.create_certificate(new_cert, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
certificate_client.update_certificate(new_cert, level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
certificate_client.delete_certificate("my-cert.pem", level=Level.SUB_ACCOUNT, tenant="tenant-subdomain")
# Certificate write operations without tenant (provider context)
certificate_client.create_certificate(new_cert, level=Level.SUB_ACCOUNT)
certificate_client.update_certificate(new_cert, level=Level.SUB_ACCOUNT)
certificate_client.delete_certificate("my-cert.pem", level=Level.SUB_ACCOUNT)-
Level:
- SERVICE_INSTANCE: Operates on instance destinations
- SUB_ACCOUNT: Operates on subaccount destinations
-
AccessStrategy (applies to subaccount reads):
- SUBSCRIBER_ONLY: Only subscriber (tenant required)
- PROVIDER_ONLY: Only provider (no tenant)
- SUBSCRIBER_FIRST: Try subscriber, fallback to provider (tenant required)
- PROVIDER_FIRST: Try provider, fallback to subscriber (tenant required)
The client produced by create_client() exposes the following operations:
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 list_instance_destinations(self, 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]: ...
# V1 Admin API - Write operations
def create_destination(self, dest: Destination, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...
def update_destination(self, dest: Destination, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...
def delete_destination(self, name: str, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...
# V2 Runtime API - Destination consumption with automatic token retrieval
def get_destination(self, name: str, level: Optional[Level] = None, options: Optional[ConsumptionOptions] = None, proxy_enabled: Optional[bool] = None) -> Optional[Destination | TransparentProxyDestination]: ...The fragment client produced by create_fragment_client() exposes the following operations:
class FragmentClient:
def get_instance_fragment(self, name: str) -> Optional[Fragment]: ...
def get_subaccount_fragment(self, name: str, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None) -> Optional[Fragment]: ...
def list_instance_fragments(self) -> List[Fragment]: ...
def list_subaccount_fragments(self, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None) -> List[Fragment]: ...
def create_fragment(self, fragment: Fragment, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...
def update_fragment(self, fragment: Fragment, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...
def delete_fragment(self, name: str, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...The certificate client produced by create_certificate_client() exposes the following operations:
class CertificateClient:
def get_instance_certificate(self, name: str) -> Optional[Certificate]: ...
def get_subaccount_certificate(self, name: str, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None) -> Optional[Certificate]: ...
def list_instance_certificates(self, filter: Optional[ListOptions] = None) -> PagedResult[Certificate]: ...
def list_subaccount_certificates(self, access_strategy: AccessStrategy = AccessStrategy.SUBSCRIBER_FIRST, tenant: Optional[str] = None, filter: Optional[ListOptions] = None) -> PagedResult[Certificate]: ...
def create_certificate(self, certificate: Certificate, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...
def update_certificate(self, certificate: Certificate, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...
def delete_certificate(self, name: str, level: Optional[Level] = Level.SUB_ACCOUNT, tenant: Optional[str] = None) -> None: ...Destination(name: str, type: str, url?: str, proxy_type?: str, authentication?: str, description?: str, properties?: dict[str, str], auth_tokens?: list[AuthToken], certificates?: list[Certificate])auth_tokensandcertificatesare populated by the v2 consumption API
ConsumptionOptions- Options for v2 destination consumption, controls HTTP headers sent to the Destination Service:fragment_name?: str- Fragment to merge into the destination (X-fragment-name)fragment_optional?: bool- IfTrue, a missing fragment does not cause an error (X-fragment-optional)tenant?: str- Tenant subdomain for token retrieval (X-tenant)user_token?: str- User JWT for OAuth2UserTokenExchange / OAuth2JWTBearer / OAuth2SAMLBearerAssertion (X-user-token)subject_token?: str- Subject token for OAuth2TokenExchange (X-subject-token)subject_token_type?: str- Format of the subject token (X-subject-token-type), e.g."urn:ietf:params:oauth:token-type:access_token"actor_token?: str- Actor token for OAuth2TokenExchange (X-actor-token)actor_token_type?: str- Format of the actor token (X-actor-token-type)saml_assertion?: str- Client-provided SAML assertion for OAuth2SAMLBearerAssertion withSAMLAssertionProvider=ClientProvided(X-samlAssertion)refresh_token?: str- Refresh token for OAuth2RefreshToken destinations (X-refresh-token)code?: str- Authorization code for OAuth2AuthorizationCode destinations (X-code)redirect_uri?: str- Redirect URI for OAuth2AuthorizationCode destinations (X-redirect-uri)code_verifier?: str- PKCE code verifier for OAuth2AuthorizationCode destinations (X-code-verifier)chain_name?: str- Name of a predefined destination chain (X-chain-name)chain_vars?: dict[str, str]- Variables for the destination chain; each entry is sent asX-chain-var-<key>
AuthToken(type: str, value: str, http_header: dict, expires_in?: str, error?: str, scope?: str, refresh_token?: str)- Authentication token from v2 APIFragment(name: str, properties: dict[str, str])Certificate(name: str, content: str, type: str)DestinationConfig(url, token_url, client_id, client_secret, identityzone)TransparentProxy(proxy_name: str, namespace: str)- Configuration for transparent proxy routingTransparentProxyDestination(name: str, url: str, headers: dict[str, str])- Destination configured for transparent proxy accessListOptions(name?: str, $skip?: int, $top?: int)- For pagination and filtering list operationsPagedResult[T](items: list[T], pagination?: PaginationInfo)- Contains results and optional pagination metadataPaginationInfo(next_cursor?: str, total_count?: int)- Pagination metadata from response headers
Notes:
- Unknown string-valued destination fields are captured into
Destination.propertiespreserving their original key casing and are included when serializing viaDestination.to_dict. Non-string unknown fields are ignored. - Fragment properties are stored as string key-value pairs in
Fragment.properties. - Certificate
contentshould be base64-encoded. Supported certificate types include PEM, JKS, P12, etc. - The v2 consumption API returns tokens in the
auth_tokensfield with ready-to-use HTTP headers inhttp_headerdict.
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.
There are three ways to configure transparent proxy support:
Enable proxy by default for all destination lookups when creating the client:
from sap_cloud_sdk.destination import create_client
# Default: use_default_proxy=False
# Turning it to true will use TransparentProxy from APPFND_CONHOS_TRANSP_PROXY environment variable
client = create_client(instance="default", use_default_proxy=True)
# All get operations will use the proxy by default
dest = client.get_instance_destination("my-destination")
# Returns TransparentProxyDestinationThe environment variable APPFND_CONHOS_TRANSP_PROXY should be set with the format {proxy_name}.{namespace}:
export APPFND_CONHOS_TRANSP_PROXY="connectivity-proxy.my-namespace"This setting might be automatically configured depending on the runtime
You can set or update the proxy configuration after client creation using the set_proxy() method:
from sap_cloud_sdk.destination import create_client, TransparentProxy
# Create client first
client = create_client(instance="default")
# Set custom proxy configuration
transparent_proxy = TransparentProxy(proxy_name="my-destination", namespace="my-namespace")
client.set_proxy(transparent_proxy)Override the client's default proxy setting for individual requests:
# Client created with use_default_proxy=True (uses proxy by default)
client = create_client(instance="default", use_default_proxy=True)
# Override to NOT use proxy for this specific request
dest = client.get_instance_destination("my-destination", proxy_enabled=False)
# Returns regular Destination
# Or explicitly enable proxy (even if client default is False)
client2 = create_client(instance="default", use_default_proxy=False)
dest2 = client2.get_instance_destination("my-destination", proxy_enabled=True)
# Returns TransparentProxyDestinationfrom sap_cloud_sdk.destination import create_client, TransparentProxy, AccessStrategy
# Example 1: Using environment variable with default proxy
client = create_client(instance="default", use_default_proxy=True)
dest = client.get_instance_destination("my-destination")
# Uses proxy from APPFND_CONHOS_TRANSP_PROXY
# Example 2: Explicit proxy configuration with set_proxy()
client = create_client(instance="default")
transparent_proxy = TransparentProxy(proxy_name="my-proxy", namespace="my-namespace")
client.set_proxy(transparent_proxy)
dest = client.get_instance_destination("my-destination", proxy_enabled=True)
# Example 3: Update proxy after creation
client = create_client(instance="default")
transparent_proxy = TransparentProxy(proxy_name="my-destination", namespace="my-namespace")
client.set_proxy(transparent_proxy)
dest = client.get_instance_destination("my-destination", proxy_enabled=True)
# Example 4: Subaccount destination with proxy
client = create_client(instance="default", use_default_proxy=True)
dest = client.get_subaccount_destination(
name="my-destination",
access_strategy=AccessStrategy.PROVIDER_ONLY
)
# Uses proxy by default (client's use_default_proxy=True)
# Example 5: Override client default per request
client = create_client(instance="default", use_default_proxy=True)
regular_dest = client.get_instance_destination("my-destination", proxy_enabled=False)
# Returns regular Destination (overrides client default)
# Example 6: V2 API (get_destination) with proxy support
client = create_client(instance="default", use_default_proxy=True)
dest = client.get_destination("my-api", proxy_enabled=True)
# Returns TransparentProxyDestination
# Example 7: V2 API with ConsumptionOptions and proxy disabled
client = create_client(instance="default", use_default_proxy=True)
options = ConsumptionOptions(fragment_name="production", tenant="tenant-1")
dest = client.get_destination("my-api", options=options, proxy_enabled=False)
# Returns regular Destination with merged fragment and tenant context
# Example 8: V2 API with level parameter for optimized lookup
client = create_client(instance="default")
# Search only at instance level
dest = client.get_destination("my-api", level=Level.SERVICE_INSTANCE)
# Search only at subaccount level
dest = client.get_destination("my-api", level=Level.SUB_ACCOUNT)
# No level specified - searches at instance level as default
dest = client.get_destination("my-api")
# Example 9: Combine level with options
options = ConsumptionOptions(fragment_name="production", tenant="tenant-1")
dest = client.get_destination("my-api", level=Level.SUB_ACCOUNT, options=options)
# Example 10: Optional fragment (no error if fragment does not exist)
options = ConsumptionOptions(fragment_name="maybe-exists", fragment_optional=True)
dest = client.get_destination("my-api", options=options)
# Example 11: User token exchange (OAuth2UserTokenExchange / OAuth2JWTBearer)
options = ConsumptionOptions(user_token="<encoded-jwt>", tenant="tenant-1")
dest = client.get_destination("my-api", options=options)
# Example 12: OAuth2TokenExchange with subject and actor tokens
options = ConsumptionOptions(
subject_token="<subject-token>",
subject_token_type="urn:ietf:params:oauth:token-type:access_token",
actor_token="<actor-token>",
actor_token_type="urn:ietf:params:oauth:token-type:access_token",
)
dest = client.get_destination("my-api", options=options)
# Example 13: Client-provided SAML assertion (SAMLAssertionProvider=ClientProvided)
options = ConsumptionOptions(saml_assertion="<base64-encoded-saml>")
dest = client.get_destination("my-api", options=options)
# Example 14: OAuth2RefreshToken
options = ConsumptionOptions(refresh_token="<refresh-token>")
dest = client.get_destination("my-api", options=options)
# Example 15: OAuth2AuthorizationCode with PKCE
options = ConsumptionOptions(
code="<authorization-code>",
redirect_uri="https://myapp/callback",
code_verifier="<pkce-code-verifier>",
)
dest = client.get_destination("my-api", options=options)
# Example 16: Destination chain with chain variables
options = ConsumptionOptions(
chain_name="my-predefined-chain",
chain_vars={"subject_token": "<token>", "subject_token_type": "access_token"},
)
dest = client.get_destination("my-api", options=options)When proxy_enabled=True (either as client default or per-request override), the methods return a TransparentProxyDestination instead of a regular Destination:
# TransparentProxyDestination has these properties:
# - name: str - The destination name
# - url: str - The proxy URL (e.g., "http://connectivity-proxy.my-namespace")
# - headers: dict[str, str] - Required headers including "X-destination-name"
client = create_client(instance="default", use_default_proxy=True)
proxy_dest = client.get_instance_destination("my-destination")
print(proxy_dest.name) # "my-destination"
print(proxy_dest.url) # "http://connectivity-proxy.my-namespace"
print(proxy_dest.headers) # {"X-destination-name": "my-destination"}The TransparentProxyDestination class provides a set_header() method to add or update headers required by the transparent proxy. Use the TransparentProxyHeader enum to ensure type-safe header names:
from sap_cloud_sdk.destination import (
create_client,
TransparentProxyHeader
)
# Get a transparent proxy destination
client = create_client(instance="default", use_default_proxy=True)
proxy_dest = client.get_instance_destination("my-destination")
# Set additional headers using the enum
proxy_dest.set_header(
TransparentProxyHeader.AUTHORIZATION,
"Bearer token123"
)
# Access the updated headers
print(proxy_dest.headers)
# {
# "X-destination-name": "my-destination",
# "Authorization": "Bearer token123"
# }Available Headers (TransparentProxyHeader enum):
X_DESTINATION_NAME- Header for specifying the destination name (automatically set byfrom_proxy())AUTHORIZATION- Header for authorization (e.g., "Bearer token", "Basic base64credentials")X_FRAGMENT_NAME- Header for specifying the fragment nameX_TENANT_SUBDOMAIN- Header for tenant subdomainX_TENANT_ID- Header for tenant IDX_FRAGMENT_OPTIONAL- Header for optional fragment flagX_DESTINATION_LEVEL- Header for destination levelX_FRAGMENT_LEVEL- Header for fragment levelX_TOKEN_SERVICE_TENANT- Header for token service tenantX_CLIENT_ASSERTION- Header for client assertionX_CLIENT_ASSERTION_TYPE- Header for client assertion typeX_CLIENT_ASSERTION_DESTINATION_NAME- Header for client assertion destination nameX_SUBJECT_TOKEN_TYPE- Header for subject token typeX_ACTOR_TOKEN- Header for actor tokenX_ACTOR_TOKEN_TYPE- Header for actor token typeX_REDIRECT_URI- Header for redirect URIX_CODE_VERIFIER- Header for code verifierX_CHAIN_NAME- Header for chain nameX_CHAIN_VAR_SUBJECT_TOKEN- Header for chain variable subject tokenX_CHAIN_VAR_SUBJECT_TOKEN_TYPE- Header for chain variable subject token typeX_CHAIN_VAR_SAML_PROVIDER_DESTINATION_NAME- Header for chain variable SAML provider destination name
The set_header() method accepts:
header: ATransparentProxyHeaderenum valuevalue: The string value for the header
This ensures only valid headers are used with transparent proxy destinations.
- If
use_default_proxy=Truebut no proxy configuration is available in the environment variable,load_transparent_proxy()returnsNoneand proxy functionality is disabled - The actual destination configuration is retrieved by the proxy service, not by the SDK
- When
proxy_enabledis not specified in get methods, the client's default setting (fromuse_default_proxy) is used - Proxy support is available for all three get methods:
get_instance_destination()- V1 API for instance-level destinationsget_subaccount_destination()- V1 API for subaccount-level destinations with access strategiesget_destination()- V2 API for runtime consumption with automatic token retrieval
When a mocks/<resource>.json file is present at the repository root, the factory functions automatically return a local in-memory client backed by that file instead of connecting to the SAP BTP Destination Service. No credentials or network access are required.
| Factory | Mock file |
|---|---|
create_client() |
mocks/destination.json |
create_fragment_client() |
mocks/fragments.json |
create_certificate_client() |
mocks/certificates.json |
WARNING: Local mode is for local development only. Local clients perform no authentication and store data in plain JSON files on disk. Never use local mode in a deployed or production environment. A warning is logged at
WARNINGlevel every time a local client is returned by a factory.
Mock files may contain sensitive data (URLs, credentials, certificates). Add them to .gitignore to prevent accidental commits:
mocks/destination.json
mocks/fragments.json
mocks/certificates.json
mocks/destination.json
{
"subaccount": [
{
"name": "my-destination",
"type": "HTTP",
"url": "https://example.com",
"authentication": "NoAuthentication"
},
{
"tenant": "my-tenant",
"name": "subscriber-destination",
"type": "HTTP",
"url": "https://subscriber.example.com",
"authentication": "NoAuthentication"
}
],
"instance": [
{
"name": "instance-destination",
"type": "HTTP",
"url": "https://instance.example.com"
}
]
}mocks/fragments.json
{
"subaccount": [
{
"FragmentName": "my-fragment",
"URL": "https://example.com",
"Authentication": "NoAuthentication"
}
],
"instance": []
}mocks/certificates.json
{
"subaccount": [
{
"Name": "my-cert.pem",
"Content": "LS0tLS1CRUdJTi...",
"Type": "PEM"
}
],
"instance": []
}Entries with a "tenant" field are treated as subscriber-specific. Entries without "tenant" are provider entries.
- Mount path:
/etc/secrets/appfnd/destination/{instance}/ - Keys:
clientid,clientsecret,url(auth base),uri(service base),identityzone - Fallback env vars:
CLOUD_SDK_CFG_DESTINATION_{INSTANCE}_{FIELD_KEY}(uppercased) - The config loader normalizes to a unified binding:
DestinationConfig(url=..., token_url=..., client_id=..., client_secret=..., identityzone=...)
- Environment variable:
APPFND_CONHOS_TRANSP_PROXY - Format:
{proxy_name}.{namespace}(e.g.,connectivity-proxy.my-namespace) - The proxy configuration is loaded and validated when the client is created
- Proxy reachability is tested via HTTP HEAD request to
http://{proxy_name}.{namespace}
The OAuth2 token URL is derived from service binding (DestinationConfig.token_url). For subscriber context, when a tenant is provided, the token provider constructs the subscriber token URL by replacing the identityzone segment with the tenant sub-domain.
DestinationNotFoundError: mapped from HTTP 404 where applicableDestinationOperationError: general operation failuresHttpError: HTTP-related or local store read/write errors withstatus_codeandresponse_textwhen applicable
- Current implementation omits explicit HTTP retries/timeouts for simplicity.
- The v2 consumption API (
get_destination) is supported for runtime scenarios requiring automatic token retrieval.