From d28d766bf6deb550951357403be8f4dc945ec6b0 Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Wed, 15 Jul 2026 13:14:43 +0200 Subject: [PATCH 1/2] fix(agentgateway): address sdk-review findings (CON-01, EL-01, EL-02, PT-08) - Extract _TOKEN_FIELD_CLIENT_ID and _TOKEN_FIELD_ACCESS_TOKEN constants in _customer.py - Extract _LOG_TRANSPARENT_MODE constant in agw_client.py - Add `from e` to raise in token request error handlers - Add type annotations and justify ImportError suppression in _telemetry_compat.py --- src/sap_cloud_sdk/agentgateway/_customer.py | 20 ++++++++++++-------- src/sap_cloud_sdk/agentgateway/agw_client.py | 10 ++++++---- src/sap_cloud_sdk/core/_telemetry_compat.py | 10 +++++++--- 3 files changed, 25 insertions(+), 15 deletions(-) diff --git a/src/sap_cloud_sdk/agentgateway/_customer.py b/src/sap_cloud_sdk/agentgateway/_customer.py index c3b5f289..a4b09c47 100644 --- a/src/sap_cloud_sdk/agentgateway/_customer.py +++ b/src/sap_cloud_sdk/agentgateway/_customer.py @@ -59,6 +59,10 @@ _GRANT_TYPE_CLIENT_CREDENTIALS = "client_credentials" _GRANT_TYPE_JWT_BEARER = "urn:ietf:params:oauth:grant-type:jwt-bearer" +# Token response field keys +_TOKEN_FIELD_CLIENT_ID = "client_id" +_TOKEN_FIELD_ACCESS_TOKEN = "access_token" + def _cache_scope_key(credentials: CustomerCredentials) -> str: """Build a cache scope key for customer-flow tokens.""" @@ -324,7 +328,7 @@ def _request_token_mtls( ssl_context = _create_ssl_context(credentials.certificate, credentials.private_key) data = { - "client_id": credentials.client_id, + _TOKEN_FIELD_CLIENT_ID: credentials.client_id, "grant_type": grant_type, "resource": _AGW_RESOURCE_URN, } @@ -363,7 +367,7 @@ def _request_token_mtls( ) token_data = response.json() - access_token = token_data.get("access_token") + access_token = token_data.get(_TOKEN_FIELD_ACCESS_TOKEN) if not access_token: raise AgentGatewaySDKError( @@ -374,7 +378,7 @@ def _request_token_mtls( return token_data except httpx.RequestError as e: - raise AgentGatewaySDKError(f"Token request failed: {e}") + raise AgentGatewaySDKError(f"Token request failed: {e}") from e def _request_token_transparent( @@ -401,7 +405,7 @@ def _request_token_transparent( AgentGatewaySDKError: If token request fails. """ data = { - "client_id": credentials.client_id, + _TOKEN_FIELD_CLIENT_ID: credentials.client_id, "grant_type": grant_type, "resource": _AGW_RESOURCE_URN, } @@ -437,7 +441,7 @@ def _request_token_transparent( ) token_data = response.json() - access_token = token_data.get("access_token") + access_token = token_data.get(_TOKEN_FIELD_ACCESS_TOKEN) if not access_token: raise AgentGatewaySDKError( @@ -448,7 +452,7 @@ def _request_token_transparent( return token_data except httpx.RequestError as e: - raise AgentGatewaySDKError(f"Token request failed: {e}") + raise AgentGatewaySDKError(f"Token request failed: {e}") from e def get_system_token_mtls( @@ -499,7 +503,7 @@ def get_system_token_mtls( extra_data={"response_type": "token"}, ) - access_token = token_data["access_token"] + access_token = token_data[_TOKEN_FIELD_ACCESS_TOKEN] if token_cache: token_cache.set_system_token( @@ -571,7 +575,7 @@ def exchange_user_token( }, ) - access_token = token_data["access_token"] + access_token = token_data[_TOKEN_FIELD_ACCESS_TOKEN] if token_cache: token_cache.set_user_token( diff --git a/src/sap_cloud_sdk/agentgateway/agw_client.py b/src/sap_cloud_sdk/agentgateway/agw_client.py index 1c120adb..efa178b7 100644 --- a/src/sap_cloud_sdk/agentgateway/agw_client.py +++ b/src/sap_cloud_sdk/agentgateway/agw_client.py @@ -42,6 +42,8 @@ logger = logging.getLogger(__name__) +_LOG_TRANSPARENT_MODE = "Transparent mode credentials detected" + class AgentGatewayClient: """Client for discovering and invoking MCP tools via SAP Agent Gateway. @@ -197,7 +199,7 @@ async def get_system_auth(self) -> AuthResult: # Check for transparent mode if detect_transparent_credentials(): - logger.info("Transparent mode credentials detected") + logger.info(_LOG_TRANSPARENT_MODE) credentials = load_customer_credentials_from_env() loop = asyncio.get_running_loop() token = await loop.run_in_executor( @@ -283,7 +285,7 @@ async def get_user_auth( # Check for transparent mode if detect_transparent_credentials(): - logger.info("Transparent mode credentials detected") + logger.info(_LOG_TRANSPARENT_MODE) credentials = load_customer_credentials_from_env() loop = asyncio.get_running_loop() token = await loop.run_in_executor( @@ -406,7 +408,7 @@ async def list_mcp_tools( # Check for transparent mode if detect_transparent_credentials(): - logger.info("Transparent mode credentials detected") + logger.info(_LOG_TRANSPARENT_MODE) credentials = load_customer_credentials_from_env() return await get_mcp_tools_customer( credentials, auth.access_token, self._config.timeout @@ -560,7 +562,7 @@ async def call_mcp_tool( # Check for transparent mode if detect_transparent_credentials(): - logger.debug("Transparent mode credentials detected") + logger.debug(_LOG_TRANSPARENT_MODE) return await call_mcp_tool_customer( tool, auth.access_token, self._config.timeout, **kwargs diff --git a/src/sap_cloud_sdk/core/_telemetry_compat.py b/src/sap_cloud_sdk/core/_telemetry_compat.py index cbc4185f..962a03cd 100644 --- a/src/sap_cloud_sdk/core/_telemetry_compat.py +++ b/src/sap_cloud_sdk/core/_telemetry_compat.py @@ -4,18 +4,22 @@ When telemetry packages are not installed, it provides no-op implementations. """ +from typing import Any, Callable, TypeVar + +_F = TypeVar("_F", bound=Callable[..., Any]) + try: from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics TELEMETRY_AVAILABLE = True except ImportError: + # Telemetry packages not installed — provide no-op implementations TELEMETRY_AVAILABLE = False - # Provide no-op implementations when telemetry is not available Module = None # type: ignore Operation = None # type: ignore - def record_metrics(*args, **kwargs): # type: ignore + def record_metrics(*args: Any, **kwargs: Any) -> Any: # type: ignore """No-op decorator when telemetry is not available.""" - def decorator(func): + def decorator(func: _F) -> _F: return func if args and callable(args[0]): # Called without parentheses: @record_metrics From 87089cba89491104387a9d5057428114d1941902 Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Wed, 15 Jul 2026 13:33:01 +0200 Subject: [PATCH 2/2] fix(agentgateway): replace remaining client_id literals with constant Use _TOKEN_FIELD_CLIENT_ID in required_vars and required_fields mappings to eliminate remaining bare 'client_id' string literals (PY-CON-01). --- src/sap_cloud_sdk/agentgateway/_customer.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/sap_cloud_sdk/agentgateway/_customer.py b/src/sap_cloud_sdk/agentgateway/_customer.py index a4b09c47..bc602be0 100644 --- a/src/sap_cloud_sdk/agentgateway/_customer.py +++ b/src/sap_cloud_sdk/agentgateway/_customer.py @@ -151,7 +151,7 @@ def load_customer_credentials_from_env( # Check required environment variables required_vars = { - _INTEGRATION_CLIENT_ID_ENV: "client_id", + _INTEGRATION_CLIENT_ID_ENV: _TOKEN_FIELD_CLIENT_ID, _INTEGRATION_AUTH_URL_ENV: "auth_url", _INTEGRATION_GATEWAY_URL_ENV: "gateway_url", } @@ -204,7 +204,7 @@ def load_customer_credentials(path: str) -> CustomerCredentials: # Credential file uses camelCase, we use snake_case required_fields = { _CredentialFields.TOKEN_SERVICE_URL: "token_service_url", - _CredentialFields.CLIENT_ID: "client_id", + _CredentialFields.CLIENT_ID: _TOKEN_FIELD_CLIENT_ID, _CredentialFields.CERTIFICATE: "certificate", _CredentialFields.PRIVATE_KEY: "private_key", _CredentialFields.GATEWAY_URL: "gateway_url",