feat: use dedicated fragments for AGW token generation (#137)

Co-authored-by: Max Heidinger <max.heidinger@sap.com>

Author: NicoleMGomes

.env_integration_tests.example

@@ -25,3 +25,7 @@ CLOUD_SDK_CFG_SDM_DEFAULT_UAA='{"url":"https://your-auth-url","clientid":"your-c
 
 CLOUD_SDK_CFG_HANA_AGENT_MEMORY_DEFAULT_APPLICATION_URL=https://your-agent-memory-api-url-here
 CLOUD_SDK_CFG_HANA_AGENT_MEMORY_DEFAULT_UAA='{"url":"https://your-auth-url","clientid":"your-client-id","clientsecret":"your-client-secret"}'
+
+APPFND_CONHOS_LANDSCAPE=your-landscape-here
+TENANT_SUBDOMAIN=your-tenant-subdomain-here
+AGW_USER_TOKEN=your-user-jwt-here

docs/INTEGRATION_TESTS.md

@@ -96,6 +96,29 @@ CLOUD_SDK_CFG_DATA_ANONYMIZATION_DEFAULT_DESTINATION_NAME=your-client-certificat
 
 The destination must be configured with `ClientCertificateAuthentication` and reference a certificate bundle containing the client certificate and private key.
 
+### Agent Gateway Integration Tests
+
+Agent Gateway integration tests use the LoB agent flow via the Destination Service. Configure the following variables in `.env_integration_tests`:
+
+```bash
+# Destination Service (required by the LoB agent flow)
+CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTID=your-destination-client-id-here
+CLOUD_SDK_CFG_DESTINATION_DEFAULT_CLIENTSECRET=your-destination-client-secret-here
+CLOUD_SDK_CFG_DESTINATION_DEFAULT_URL=https://your-destination-auth-url-here
+CLOUD_SDK_CFG_DESTINATION_DEFAULT_URI=https://your-destination-configuration-uri-here
+CLOUD_SDK_CFG_DESTINATION_DEFAULT_IDENTITYZONE=your-identity-zone-here
+
+# Landscape suffix used to resolve the IAS destination name
+APPFND_CONHOS_LANDSCAPE=your-landscape-here
+
+# Tenant subdomain for multi-tenant lookup
+TENANT_SUBDOMAIN=your-tenant-subdomain-here
+
+# User JWT for token exchange scenarios (get_user_auth)
+# If not set, user auth scenarios are automatically skipped
+AGW_USER_TOKEN=your-user-jwt-here
+```
+
 ## Running Integration Tests
 
 ```bash
@@ -108,6 +131,7 @@ uv run pytest tests/core/integration/data_anonymization -v
 uv run pytest tests/objectstore/integration/ -v
 uv run pytest tests/destination/integration/ -v
 uv run pytest tests/agent_memory/integration/ -v
+uv run pytest tests/agentgateway/integration/ -v
 ```
 
 ### BDD Scenarios

pyproject.toml

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

src/sap_cloud_sdk/agentgateway/__init__.py

@@ -52,7 +52,7 @@
     ]
 """
 
-from sap_cloud_sdk.agentgateway._models import MCPTool
+from sap_cloud_sdk.agentgateway._models import AuthResult, MCPTool
 from sap_cloud_sdk.agentgateway.config import ClientConfig
 from sap_cloud_sdk.agentgateway.agw_client import create_client, AgentGatewayClient
 from sap_cloud_sdk.agentgateway.exceptions import (
@@ -69,6 +69,7 @@
     # Configuration
     "ClientConfig",
     # Data models
+    "AuthResult",
     "MCPTool",
     # Exceptions
     "AgentGatewaySDKError",

src/sap_cloud_sdk/agentgateway/_customer.py

@@ -8,7 +8,6 @@
 - Tool invocation: mTLS + jwt-bearer grant → user-scoped token (principal propagation)
 """
 
-import asyncio
 import json
 import logging
 import os
@@ -432,17 +431,18 @@ async def _list_server_tools(
 
 async def get_mcp_tools_customer(
     credentials: CustomerCredentials,
+    system_token: str,
     timeout: float,
-    app_tid: str | None = None,
 ) -> list[MCPTool]:
     """List all MCP tools from servers defined in credentials.
 
     Iterates over all integrationDependencies in the credentials file and
-    discovers tools from each MCP server using mTLS client credentials.
+    discovers tools from each MCP server using a pre-fetched system token.
 
     Args:
         credentials: Customer credentials with integrationDependencies.
-        app_tid: BTP Application Tenant ID of subscriber (optional).
+        system_token: Pre-fetched raw system token for authentication.
+        timeout: HTTP timeout in seconds for MCP server calls.
 
     Returns:
         List of MCPTool objects from all servers.
@@ -459,12 +459,6 @@ async def get_mcp_tools_customer(
 
     logger.info("Discovering tools from %d MCP server(s)", len(dependencies))
 
-    # Get system token for discovery
-    loop = asyncio.get_running_loop()
-    system_token = await loop.run_in_executor(
-        None, get_system_token_mtls, credentials, timeout, app_tid
-    )
-
     tools: list[MCPTool] = []
 
     for dep in dependencies:
@@ -490,53 +484,30 @@ async def get_mcp_tools_customer(
 
 
 async def call_mcp_tool_customer(
-    credentials: CustomerCredentials,
     tool: MCPTool,
-    user_token: str | None,
+    auth_token: str,
     timeout: float,
-    app_tid: str | None = None,
     **kwargs,
 ) -> str:
     """Invoke an MCP tool using customer flow.
 
-    If user_token is provided, exchanges it for an AGW-scoped token to preserve
-    user identity for principal propagation. Otherwise, falls back to system token.
+    Uses a pre-fetched token (either user-scoped or system-scoped) for
+    authentication against the MCP server.
 
     Args:
-        credentials: Customer credentials.
         tool: MCPTool to invoke.
-        user_token: User's JWT token for principal propagation (optional).
-            If None, system token is used instead (no principal propagation).
-        app_tid: BTP Application Tenant ID of subscriber (optional).
+        auth_token: Pre-fetched raw access token for authentication.
+        timeout: HTTP timeout in seconds for the MCP server call.
         **kwargs: Tool input parameters.
 
     Returns:
         Tool execution result as string.
     """
     logger.info("Calling tool '%s' on server '%s'", tool.name, tool.server_name)
 
-    loop = asyncio.get_running_loop()
-
-    if user_token:
-        # Exchange user token for AGW-scoped token (with principal propagation)
-        agw_token = await loop.run_in_executor(
-            None, exchange_user_token, credentials, user_token, timeout, app_tid
-        )
-    else:
-        # TODO: IBD workaround - use system token when user_token is not available.
-        # This bypasses principal propagation. Remove this fallback once IBD
-        # supports proper user token flow.
-        logger.warning(
-            "No user_token provided - using system token for tool invocation. "
-            "Principal propagation will NOT work."
-        )
-        agw_token = await loop.run_in_executor(
-            None, get_system_token_mtls, credentials, timeout, app_tid
-        )
-
     async with httpx.AsyncClient(
         headers={
-            "Authorization": f"Bearer {agw_token}",
+            "Authorization": f"Bearer {auth_token}",
             "x-correlation-id": str(uuid.uuid4()),
         },
         timeout=timeout,