feat(agw): add implicit auditlog usage on agw

pyproject.toml

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

src/sap_cloud_sdk/agentgateway/_audit_events.py

@@ -0,0 +1,9 @@
+"""Audit event name constants for Agent Gateway."""
+
+from enum import StrEnum
+
+
+class McpToolEvent(StrEnum):
+    INVOKED = "MCP_TOOL_INVOKED"
+    COMPLETED = "MCP_TOOL_COMPLETED"
+    FAILED = "MCP_TOOL_FAILED"

src/sap_cloud_sdk/agentgateway/agw_client.py

@@ -34,9 +34,19 @@
     AuthResult,
     MCPTool,
 )
+from sap_cloud_sdk.core.auditlog_ng import AuditClient
+from sap_cloud_sdk.core.auditlog_ng.cross_module_helper import (
+    create_audit_client,
+    send_event as _send_audit_event,
+)
+from sap_cloud_sdk.agentgateway._audit_events import McpToolEvent
 from sap_cloud_sdk.agentgateway._token_cache import _GatewayUrlCache, _TokenCache
 from sap_cloud_sdk.agentgateway.exceptions import AgentGatewaySDKError
-from sap_cloud_sdk.core.telemetry import Module, Operation, record_metrics
+from sap_cloud_sdk.core.telemetry import (
+    Module,
+    Operation,
+    record_metrics,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -123,6 +133,9 @@ def __init__(
         self._token_cache = _TokenCache(self._config)
         self._gateway_url_cache = _GatewayUrlCache()
         self._telemetry_source = _telemetry_source
+        self._audit_client: AuditClient | None = create_audit_client(
+            tenant_subdomain, Module.AGENTGATEWAY, self._config.audit_log_mode
+        )
 
     @staticmethod
     def _resolve_value(
@@ -549,18 +562,65 @@ async def call_mcp_tool(
                     )
                     auth = await self.get_system_auth(app_tid)
 
-                return await call_mcp_tool_customer(
-                    tool, auth.access_token, self._config.timeout, **kwargs
+                _send_audit_event(
+                    self._audit_client,
+                    McpToolEvent.INVOKED,
+                    {"tool": tool.name},
+                    user_token,
+                    self._config.audit_log_mode,
+                )
+                try:
+                    result = await call_mcp_tool_customer(
+                        tool, auth.access_token, self._config.timeout, **kwargs
+                    )
+                except Exception as e:
+                    _send_audit_event(
+                        self._audit_client,
+                        McpToolEvent.FAILED,
+                        {"tool": tool.name, "error_type": type(e).__name__},
+                        user_token,
+                        self._config.audit_log_mode,
+                    )
+                    raise
+                _send_audit_event(
+                    self._audit_client,
+                    McpToolEvent.COMPLETED,
+                    {"tool": tool.name},
+                    user_token,
+                    self._config.audit_log_mode,
                 )
+                return result
 
             # LoB flow - requires user_token and tenant_subdomain
             if app_tid:
                 logger.warning("app_tid parameter ignored for LoB agent flow")
 
             auth = await self.get_user_auth(user_token, app_tid)
-            return await call_mcp_tool_lob(
-                tool, auth.access_token, self._config.timeout, **kwargs
+            _send_audit_event(
+                self._audit_client,
+                McpToolEvent.INVOKED,
+                {"tool": tool.name},
+                user_token,
+            )
+            try:
+                result = await call_mcp_tool_lob(
+                    tool, auth.access_token, self._config.timeout, **kwargs
+                )
+            except Exception as e:
+                _send_audit_event(
+                    self._audit_client,
+                    McpToolEvent.FAILED,
+                    {"tool": tool.name, "error_type": type(e).__name__},
+                    user_token,
+                )
+                raise
+            _send_audit_event(
+                self._audit_client,
+                McpToolEvent.COMPLETED,
+                {"tool": tool.name},
+                user_token,
             )
+            return result
 
         except AgentGatewaySDKError:
             # Re-raise SDK errors as-is

src/sap_cloud_sdk/agentgateway/config.py

@@ -2,6 +2,10 @@
 
 from dataclasses import dataclass
 
+from sap_cloud_sdk.core.auditlog_ng.cross_module_helper import AuditLogMode
+
+__all__ = ["AuditLogMode", "ClientConfig"]
+
 DEFAULT_TIMEOUT_SECONDS = 60.0
 DEFAULT_FALLBACK_TOKEN_TTL_SECONDS = 300.0
 DEFAULT_TOKEN_EXPIRY_BUFFER_SECONDS = 30.0
@@ -22,13 +26,16 @@ class ClientConfig:
             token expiries before a cached token is considered stale.
         max_system_token_cache_size: Maximum number of cached system tokens.
         max_user_token_cache_size: Maximum number of cached user tokens.
+        audit_log_mode: Controls how audit logging failures are handled.
+            Defaults to BEST_EFFORT.
     """
 
     timeout: float = DEFAULT_TIMEOUT_SECONDS
     fallback_token_ttl_seconds: float = DEFAULT_FALLBACK_TOKEN_TTL_SECONDS
     token_expiry_buffer_seconds: float = DEFAULT_TOKEN_EXPIRY_BUFFER_SECONDS
     max_system_token_cache_size: int = DEFAULT_MAX_SYSTEM_TOKEN_CACHE_SIZE
     max_user_token_cache_size: int = DEFAULT_MAX_USER_TOKEN_CACHE_SIZE
+    audit_log_mode: AuditLogMode = AuditLogMode.BEST_EFFORT
 
     def __post_init__(self) -> None:
         if self.token_expiry_buffer_seconds >= self.fallback_token_ttl_seconds:

src/sap_cloud_sdk/agentgateway/user-guide.md

@@ -193,9 +193,44 @@ agw_client = create_client(tenant_subdomain="my-tenant", config=config)
 - `token_expiry_buffer_seconds`: Safety buffer subtracted from explicit token expiries before a cached token is reused. Default: `30.0`.
 - `max_system_token_cache_size`: Maximum number of cached system tokens per client instance. Default: `32`.
 - `max_user_token_cache_size`: Maximum number of cached exchanged user tokens per client instance. Default: `256`.
+- `audit_log_mode`: Controls how audit logging failures are handled. Default: `AuditLogMode.BEST_EFFORT`.
 
 The SDK keeps token caches per `AgentGatewayClient` instance and reuses valid cached tokens for repeated authentication calls. System and user token caches are bounded independently with least-recently-used eviction.
 
+## Implicit Audit Logging
+
+The SDK automatically emits audit log events for every MCP tool invocation when the client is created with a `tenant_subdomain` (LoB flow). No additional configuration is required.
+
+Three events are emitted per invocation:
+
+| Event | When |
+|---|---|
+| `MCP_TOOL_INVOKED` | Before the tool call starts |
+| `MCP_TOOL_COMPLETED` | After the tool call succeeds |
+| `MCP_TOOL_FAILED` | When the tool call raises an exception (includes `error_type`) |
+
+Events are sent as `ZzzCustomEvent` to the SAP Audit Log Service v3 via the `AuditLogV3_Destination` destination.
+
+### AuditLogMode
+
+Use `AuditLogMode` in `ClientConfig` to control failure handling:
+
+```python
+from sap_cloud_sdk.agentgateway import ClientConfig, create_client
+from sap_cloud_sdk.agentgateway.config import AuditLogMode
+
+config = ClientConfig(audit_log_mode=AuditLogMode.STRICT)
+agw_client = create_client(tenant_subdomain="my-tenant", config=config)
+```
+
+| Mode | Behavior |
+|---|---|
+| `BEST_EFFORT` | Audit failures are logged as warnings and never raised. Default. |
+| `STRICT` | Audit failures raise an exception, blocking the operation. |
+| `DISABLED` | Audit logging is skipped entirely. |
+
+Audit logging is only active for LoB agents — Customer agents do not resolve a tenant subdomain at construction time and therefore emit no audit events.
+
 ### AgentGatewayClient
 
 ```python

src/sap_cloud_sdk/core/auditlog_ng/__init__.py

@@ -56,6 +56,11 @@
     AuditLogNGConfig,
     SCHEMA_URL,
 )
+from sap_cloud_sdk.core.auditlog_ng.cross_module_helper import (
+    AuditLogMode,
+    create_audit_client,
+    send_event,
+)
 from sap_cloud_sdk.core.auditlog_ng.exceptions import (
     AuditLogNGError,
     ClientCreationError,
@@ -295,6 +300,10 @@ def create_client(
     "create_client",
     # Client
     "AuditClient",
+    # Audit helpers
+    "AuditLogMode",
+    "create_audit_client",
+    "send_event",
     # Configuration
     "AuditLogNGConfig",
     # Exceptions

src/sap_cloud_sdk/core/auditlog_ng/cross_module_helper.py

@@ -0,0 +1,171 @@
+"""Implicit audit log helpers shared across SDK modules.
+
+Provides managed client creation and a ``ZzzCustomEvent`` send pattern for
+modules that emit audit events implicitly without
+requiring callers to configure auditing directly.
+
+For standard catalog events (``DataAccess``, ``ConfigurationChange``,
+``UserLoginSuccess``, etc.), construct the protobuf directly and call
+``AuditClient.send()``.
+"""
+
+import logging
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Callable
+
+import sap_cloud_sdk.core.auditlog_ng as auditlog_ng
+from sap_cloud_sdk.core.auditlog_ng.client import AuditClient
+from sap_cloud_sdk.core.auditlog_ng.gen.sap.auditlog.auditevent.v2 import (
+    auditevent_pb2 as pb,
+)
+from sap_cloud_sdk.core.telemetry import Module, get_tenant_id
+from sap_cloud_sdk.ias import parse_token
+
+logger = logging.getLogger(__name__)
+
+
+class AuditLogMode(Enum):
+    """Controls how audit logging failures are handled.
+
+    Attributes:
+        DISABLED: Audit logging is skipped entirely.
+        BEST_EFFORT: Failures are logged at WARNING level but never raised.
+            This is the default.
+        STRICT: Failures raise an exception, blocking the operation.
+    """
+
+    DISABLED = "disabled"
+    BEST_EFFORT = "best_effort"
+    STRICT = "strict"
+
+
+def _emit_custom_event(
+    audit_client: AuditClient,
+    tenant_id: str,
+    event_name: str,
+    payload: dict,
+    user_token: str | Callable[[], str] | None = None,
+) -> None:
+    """Build and send a ZzzCustomEvent to the audit log service.
+
+    Args:
+        audit_client: Initialized AuditClient instance.
+        tenant_id: Tenant UUID stamped on the event.
+        event_name: Event identifier (e.g. ``"MCP_TOOL_INVOKED"``).
+        payload: Arbitrary key/value pairs serialized into the custom struct.
+            ``event_name`` is always included automatically.
+        user_token: Optional user JWT. The user initiator ID is resolved from
+            ``scim_id`` or ``sub`` claims via ``_resolve_user_id``.
+    """
+    common = pb.Common()
+    common.timestamp.FromDatetime(datetime.now(timezone.utc))
+    common.tenant_id = tenant_id
+    common.app_context["event_name"] = event_name
+    user_id = _get_user_id(user_token)
+    if user_id:
+        common.user_initiator_id = user_id
+
+    event = pb.ZzzCustomEvent()
+    event.common.CopyFrom(common)
+    event.custom.struct_value.update({"event_name": event_name, **payload})
+    audit_client.send(event)
+
+
+def _resolve_tenant(tenant_subdomain: str | Callable[[], str] | None) -> str | None:
+    if isinstance(tenant_subdomain, str):
+        return tenant_subdomain
+    if callable(tenant_subdomain):
+        return tenant_subdomain()
+    return None
+
+
+def _resolve_user_token(user_token: str | Callable[[], str] | None) -> str | None:
+    if isinstance(user_token, str):
+        return user_token
+    if callable(user_token):
+        return user_token()
+    return None
+
+
+def _get_user_id(user_token: str | Callable[[], str] | None) -> str | None:
+    token = _resolve_user_token(user_token)
+    if not token:
+        return None
+    try:
+        claims = parse_token(token)
+        return claims.scim_id or claims.sub or None
+    except Exception:
+        return None
+
+
+def create_audit_client(
+    tenant_subdomain: str | Callable[[], str] | None,
+    module: Module,
+    mode: AuditLogMode = AuditLogMode.BEST_EFFORT,
+) -> AuditClient | None:
+    """Create an audit client from a LoB destination.
+
+    Args:
+        tenant_subdomain: Tenant subdomain string or callable returning one.
+        module: Telemetry source module identifier.
+        mode: Controls failure handling. Returns None when DISABLED or when
+            tenant is not resolvable.
+
+    Returns:
+        Initialized AuditClient, or None on failure / when disabled.
+    """
+    if mode is AuditLogMode.DISABLED:
+        return None
+    resolved = _resolve_tenant(tenant_subdomain)
+    if not resolved:
+        return None
+    try:
+        return auditlog_ng.create_client(
+            tenant=resolved,
+            _telemetry_source=module,
+        )
+    except Exception:
+        if mode is AuditLogMode.STRICT:
+            raise
+        # BEST_EFFORT: suppress the error and warn instead of propagating
+        logger.warning(
+            "Failed to create audit client — audit events will not be recorded",
+            exc_info=True,
+        )
+        return None
+
+
+def send_event(
+    audit_client: AuditClient | None,
+    event_name: str,
+    payload: dict,
+    user_token: str | Callable[[], str] | None = None,
+    mode: AuditLogMode = AuditLogMode.BEST_EFFORT,
+) -> None:
+    """Send a ZzzCustomEvent to the audit log service.
+
+    Resolves the tenant ID via ``get_tenant_id()`` and applies ``AuditLogMode``
+    semantics around the send. No-op when disabled, client is None, or tenant
+    is not resolvable.
+
+    Args:
+        audit_client: Initialized AuditClient, or None to skip.
+        event_name: Event identifier stamped on the event (e.g. ``"MCP_TOOL_INVOKED"``).
+        payload: Arbitrary key/value pairs included in the custom struct.
+        user_token: Optional user JWT. The user initiator ID is resolved from
+            ``scim_id`` or ``sub`` claims.
+        mode: Controls failure handling.
+    """
+    if mode is AuditLogMode.DISABLED:
+        return
+    tenant_id = get_tenant_id()
+    if audit_client is None or not tenant_id:
+        return
+    try:
+        _emit_custom_event(audit_client, tenant_id, event_name, payload, user_token)
+    except Exception:
+        if mode is AuditLogMode.STRICT:
+            raise
+        # BEST_EFFORT: supress the error and warn instead of propagating
+        logger.warning("Failed to send audit event", exc_info=True)