feat: new methods to return destination instance id and client id for agw

pyproject.toml

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

src/sap_cloud_sdk/agentgateway/_lob.py

@@ -114,6 +114,33 @@ def _fetch_auth_token(
     return raw_token, gateway_url
 
 
+def get_ias_client_id_lob() -> str:
+    """Read the IAS client ID from the IAS destination properties (LoB flow).
+
+    Fetches the IAS destination (``sap-managed-runtime-ias-{landscape}``)
+    at provider subaccount level with ``$skipTokenRetrieval=true`` so only
+    destination properties are returned — no auth token exchange is performed.
+
+    Returns:
+        The IAS client ID string, or ``""`` if the ``clientId`` property is absent.
+
+    Raises:
+        EnvironmentError: If ``APPFND_CONHOS_LANDSCAPE`` is not set.
+        AgentGatewaySDKError: If the IAS destination is not found.
+        Any exception raised by the destination client.
+    """
+    dest_name = _ias_dest_name()
+    client = create_destination_client(instance=_DESTINATION_INSTANCE)
+    dest = client.get_destination(
+        dest_name,
+        level=ConsumptionLevel.PROVIDER_SUBACCOUNT,
+        options=ConsumptionOptions(skip_token_retrieval=True),
+    )
+    if not dest:
+        raise AgentGatewaySDKError(f"IAS destination '{dest_name}' not found")
+    return dest.properties.get("clientId", "")
+
+
 async def fetch_system_auth(
     tenant_subdomain: str,
     token_cache: _TokenCache | None = None,

src/sap_cloud_sdk/agentgateway/agw_client.py

@@ -25,6 +25,7 @@
     fetch_system_auth,
     fetch_user_auth,
     get_agent_cards_lob,
+    get_ias_client_id_lob,
     get_mcp_tools_lob,
 )
 from sap_cloud_sdk.agentgateway._models import (
@@ -295,6 +296,40 @@ async def get_user_auth(
             logger.exception("Unexpected error during user auth exchange")
             raise AgentGatewaySDKError(f"User auth exchange failed: {e}") from e
 
+    @record_metrics(Module.AGENTGATEWAY, Operation.AGENTGATEWAY_GET_IAS_CLIENT_ID)
+    def get_ias_client_id(self) -> str:
+        """Read the IAS client ID.
+
+        Automatically detects agent type (LoB vs Customer) based on
+        credential file presence.
+
+        - Customer agents: Returns ``client_id`` directly from the credentials file.
+        - LoB agents: Fetches the IAS destination
+          (``sap-managed-runtime-ias-{landscape}``) at provider subaccount level
+          and returns the ``clientId`` destination property.
+
+        Returns:
+            The IAS client ID string.
+
+        Raises:
+            AgentGatewaySDKError: If the IAS client ID cannot be resolved.
+        """
+        try:
+            credentials_path = detect_customer_agent_credentials()
+            if credentials_path:
+                logger.info(
+                    "Customer agent credentials detected at '%s'", credentials_path
+                )
+                credentials = load_customer_credentials(credentials_path)
+                return credentials.client_id
+
+            # LoB flow — read clientId from the IAS destination properties
+            return get_ias_client_id_lob()
+        except AgentGatewaySDKError:
+            raise
+        except Exception as e:
+            raise AgentGatewaySDKError(f"Could not resolve IAS client ID: {e}") from e
+
     @record_metrics(Module.AGENTGATEWAY, Operation.AGENTGATEWAY_LIST_MCP_TOOLS)
     async def list_mcp_tools(
         self,

src/sap_cloud_sdk/agentgateway/user-guide.md

@@ -218,6 +218,22 @@ class AgentGatewayClient:
         self,
         filter: AgentCardFilter | None = None,
     ) -> list[Agent]
+
+    def get_ias_client_id(self) -> str
+```
+
+#### `get_ias_client_id()`
+
+Returns the IAS client ID. Automatically detects agent type:
+
+- **Customer agents**: reads `client_id` directly from the mounted credentials file.
+- **LoB agents**: fetches the IAS destination (`sap-managed-runtime-ias-{landscape}`) at provider subaccount level and returns the `clientId` destination property.
+
+Raises `AgentGatewaySDKError` if the value cannot be resolved.
+
+```python
+agw_client = create_client(tenant_subdomain="my-tenant")
+client_id = agw_client.get_ias_client_id()
 ```
 
 ### AgentCardFilter

src/sap_cloud_sdk/core/telemetry/operation.py

@@ -27,6 +27,7 @@ class Operation(str, Enum):
     DESTINATION_UPDATE_DESTINATION = "update_destination"
     DESTINATION_DELETE_DESTINATION = "delete_destination"
     DESTINATION_GET_DESTINATION = "get_destination"
+    DESTINATION_GET_SERVICE_INSTANCE_ID = "get_service_instance_id"
 
     # Destination Label Operations
     DESTINATION_GET_LABELS = "get_destination_labels"
@@ -188,6 +189,7 @@ class Operation(str, Enum):
     AGENTGATEWAY_GET_SYSTEM_AUTH = "get_system_auth"
     AGENTGATEWAY_GET_USER_AUTH = "get_user_auth"
     AGENTGATEWAY_LIST_AGENT_CARDS = "list_agent_cards"
+    AGENTGATEWAY_GET_IAS_CLIENT_ID = "get_ias_client_id"
 
     # Agent Memory Operations
     AGENT_MEMORY_ADD_MEMORY = "add_memory"

src/sap_cloud_sdk/destination/_models.py

@@ -510,6 +510,10 @@ class ConsumptionOptions:
         chain_vars: Key-value pairs for destination chain variables (X-chain-var-<name>).
             Each entry is sent as a separate "X-chain-var-<key>" header. Only applicable
             when chain_name is provided.
+        skip_token_retrieval: When True, instructs the Destination Service to skip the
+            OAuth2 token exchange and return only the destination configuration properties
+            ($skipTokenRetrieval query parameter). Useful when only destination metadata
+            is needed and token retrieval would be wasteful or cause unnecessary errors.
 
     Example:
         ```python
@@ -560,6 +564,7 @@ class ConsumptionOptions:
     code_verifier: Optional[str] = None
     chain_name: Optional[str] = None
     chain_vars: Optional[dict] = None
+    skip_token_retrieval: bool = False
 
 
 @dataclass
@@ -1000,3 +1005,8 @@ def set_header(self, header: TransparentProxyHeader, value: str) -> None:
             ```
         """
         self.headers[header.value] = value
+
+
+@dataclass
+class _DestinationInstanceConfig:
+    instanceid: str = ""

src/sap_cloud_sdk/destination/client.py

@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+import logging
 import warnings
-from typing import List, Optional, Callable, TypeVar
+from typing import Any, Dict, List, Optional, Callable, TypeVar
 
 from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
+from sap_cloud_sdk.core.secret_resolver import read_from_mount_and_fallback_to_env_var
 from sap_cloud_sdk.destination._http import DestinationHttp, API_V1, API_V2
 from sap_cloud_sdk.destination._models import (
     AccessStrategy,
@@ -18,6 +20,7 @@
     PatchLabels,
     TransparentProxy,
     TransparentProxyDestination,
+    _DestinationInstanceConfig,
 )
 from sap_cloud_sdk.destination.config import load_transparent_proxy
 from sap_cloud_sdk.destination.exceptions import (
@@ -31,6 +34,8 @@
 
 T = TypeVar("T")
 
+logger = logging.getLogger(__name__)
+
 _SUBACCOUNT_COLLECTION = "subaccountDestinations"
 _INSTANCE_COLLECTION = "instanceDestinations"
 
@@ -426,7 +431,13 @@ def get_destination(
                 else f"{API_V2}/destinations/{name}"
             )
 
-            resp = self._http.get(path, headers=headers, tenant_subdomain=tenant)
+            params: Dict[str, Any] = {}
+            if options and options.skip_token_retrieval:
+                params["$skipTokenRetrieval"] = "true"
+
+            resp = self._http.get(
+                path, headers=headers, tenant_subdomain=tenant, params=params or None
+            )
             data = resp.json()
 
             # Parse v2 response: destinationConfiguration + authTokens + certificates
@@ -665,6 +676,35 @@ def patch_destination_labels(
                 f"failed to patch labels for destination '{name}': {e}"
             )
 
+    @record_metrics(Module.DESTINATION, Operation.DESTINATION_GET_SERVICE_INSTANCE_ID)
+    def get_service_instance_id(self) -> str:
+        """Read the destination service instance ID from mounted secrets.
+
+        Resolves ``instanceid`` via the common secret resolver
+        (base mount ``/etc/secrets/appfnd``, env var prefix ``CLOUD_SDK_CFG``,
+        module ``destination``, instance ``default``).
+
+        Returns:
+            The instance ID string.
+
+        Raises:
+            DestinationOperationError: If the instance ID cannot be resolved from secrets.
+        """
+        try:
+            config = _DestinationInstanceConfig()
+            read_from_mount_and_fallback_to_env_var(
+                base_volume_mount="/etc/secrets/appfnd",
+                base_var_name="CLOUD_SDK_CFG",
+                module="destination",
+                instance="default",
+                target=config,
+            )
+            return config.instanceid
+        except Exception as e:
+            raise DestinationOperationError(
+                "Could not resolve destination instance ID from secrets"
+            ) from e
+
     # ---------- Internal helpers ----------
 
     def _get_destination(

src/sap_cloud_sdk/destination/user-guide.md

@@ -737,3 +737,22 @@ The OAuth2 token URL is derived from service binding (`DestinationConfig.token_u
 
 - Current implementation omits explicit HTTP retries/timeouts for simplicity.
 - The v2 consumption API (`get_destination`) is supported for runtime scenarios requiring automatic token retrieval.
+
+## Utilities
+
+### `get_service_instance_id()`
+
+Reads the destination service instance ID from the mounted secret binding. This is useful when you need the instance ID for programmatic use (e.g., to pass it to another service or for logging).
+
+```python
+from sap_cloud_sdk.destination import get_service_instance_id
+
+instance_id = get_service_instance_id()
+```
+
+The function reads the `instanceid` field from the secret resolved via the standard mount/env fallback:
+
+- **Mount path**: `/etc/secrets/appfnd/destination/default/instanceid`
+- **Env var fallback**: `CLOUD_SDK_CFG_DESTINATION_DEFAULT_INSTANCEID`
+
+Raises `DestinationOperationError` if the secret cannot be resolved (e.g., running locally without a binding).

tests/agentgateway/integration/agw_auth.feature

@@ -55,3 +55,7 @@ Feature: Agent Gateway Auth Integration
     When I call list_mcp_tools
     And I call call_mcp_tool with the sample MCP tool and the user token
     Then the tool result should be a non-empty string
+
+  Scenario: Get IAS client ID returns a non-empty string
+    When I call get_ias_client_id
+    Then the ias_client_id should be a non-empty string

tests/agentgateway/integration/test_agw_bdd.py

@@ -54,6 +54,7 @@ def __init__(self):
         self.tools: Optional[list[MCPTool]] = None
         self.tool_result: Optional[str] = None
         self.sample_mcp_tool_name: Optional[str] = None
+        self.ias_client_id: Optional[str] = None
 
 
 @pytest.fixture
@@ -266,3 +267,17 @@ def tool_result_is_non_empty_string(context: ScenarioContext):
     assert context.tool_result is not None
     assert isinstance(context.tool_result, str)
     assert context.tool_result.strip(), "Expected a non-empty tool result"
+
+
+@when("I call get_ias_client_id")
+def call_get_ias_client_id(context: ScenarioContext, agw_client: AgentGatewayClient):
+    """Call get_ias_client_id and store the result."""
+    context.ias_client_id = agw_client.get_ias_client_id()
+
+
+@then("the ias_client_id should be a non-empty string")
+def ias_client_id_non_empty(context: ScenarioContext):
+    """Verify the IAS client ID is a non-empty string."""
+    assert context.ias_client_id is not None
+    assert isinstance(context.ias_client_id, str)
+    assert context.ias_client_id.strip(), "Expected a non-empty IAS client ID"

tests/agentgateway/unit/test_agw_client.py

@@ -980,3 +980,70 @@ async def test_wraps_unexpected_error(self):
             agw_client = create_client(tenant_subdomain="my-tenant")
             with pytest.raises(AgentGatewaySDKError, match="Agent card discovery failed"):
                 await agw_client.list_agent_cards()
+
+
+# ============================================================
+# Test: get_ias_client_id
+# ============================================================
+
+_DEST_CREATE_PATCH = "sap_cloud_sdk.destination.create_client"
+_IAS_DEST_NAME_PATCH = "sap_cloud_sdk.agentgateway._lob._ias_dest_name"
+_GET_IAS_CLIENT_ID_LOB_PATCH = "sap_cloud_sdk.agentgateway.agw_client.get_ias_client_id_lob"
+_DETECT_CREDS_PATCH = "sap_cloud_sdk.agentgateway.agw_client.detect_customer_agent_credentials"
+_LOAD_CREDS_PATCH = "sap_cloud_sdk.agentgateway.agw_client.load_customer_credentials"
+
+_NO_CUSTOMER_CREDS = patch(_DETECT_CREDS_PATCH, return_value=None)
+
+
+class TestGetIasClientId:
+    """Tests for AgentGatewayClient.get_ias_client_id()."""
+
+    # --- Customer flow ---
+
+    def test_customer_returns_client_id_from_credentials(self):
+        mock_creds = MagicMock()
+        mock_creds.client_id = "customer-client-id"
+
+        with (
+            patch(_DETECT_CREDS_PATCH, return_value="/etc/ums/credentials/credentials"),
+            patch(_LOAD_CREDS_PATCH, return_value=mock_creds),
+        ):
+            result = create_client().get_ias_client_id()
+
+        assert result == "customer-client-id"
+
+    def test_customer_raises_on_load_failure(self):
+        with (
+            patch(_DETECT_CREDS_PATCH, return_value="/etc/ums/credentials/credentials"),
+            patch(_LOAD_CREDS_PATCH, side_effect=Exception("parse error")),
+        ):
+            with pytest.raises(AgentGatewaySDKError, match="Could not resolve IAS client ID"):
+                create_client().get_ias_client_id()
+
+    # --- LoB flow ---
+
+    @_NO_CUSTOMER_CREDS
+    def test_lob_returns_client_id_from_destination_properties(self, _mock_detect):
+        with patch(_GET_IAS_CLIENT_ID_LOB_PATCH, return_value="lob-client-id"):
+            result = create_client(tenant_subdomain="my-tenant").get_ias_client_id()
+
+        assert result == "lob-client-id"
+
+    @_NO_CUSTOMER_CREDS
+    def test_lob_raises_when_destination_not_found(self, _mock_detect):
+        with patch(_GET_IAS_CLIENT_ID_LOB_PATCH, side_effect=AgentGatewaySDKError("IAS destination 'sap-managed-runtime-ias-eu10' not found")):
+            with pytest.raises(AgentGatewaySDKError, match="IAS destination"):
+                create_client(tenant_subdomain="my-tenant").get_ias_client_id()
+
+    @_NO_CUSTOMER_CREDS
+    def test_lob_returns_empty_string_when_property_absent(self, _mock_detect):
+        with patch(_GET_IAS_CLIENT_ID_LOB_PATCH, return_value=""):
+            result = create_client(tenant_subdomain="my-tenant").get_ias_client_id()
+
+        assert result == ""
+
+    @_NO_CUSTOMER_CREDS
+    def test_lob_raises_on_exception(self, _mock_detect):
+        with patch(_GET_IAS_CLIENT_ID_LOB_PATCH, side_effect=EnvironmentError("APPFND_CONHOS_LANDSCAPE not set")):
+            with pytest.raises(AgentGatewaySDKError, match="Could not resolve IAS client ID"):
+                create_client(tenant_subdomain="my-tenant").get_ias_client_id()