Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
[project]
name = "sap-cloud-sdk"
version = "0.48.3"

version = "0.49.0"
description = "SAP Cloud SDK for Python"
readme = "README.md"
license = "Apache-2.0"
Expand Down
161 changes: 139 additions & 22 deletions src/sap_cloud_sdk/aicore/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,26 @@

logger = logging.getLogger(__name__)

# When set, the infrastructure sidecar adds the mTLS certificate transparently.
# The SDK calls the XSUAA token endpoint over plain HTTPS with only client_id.
# No client_secret or certificate material is required in the service binding.
TRANSPARENT_TLS_ENV_VAR = "AICORE_TRANSPARENT_TLS"

Comment thread
tiagoek marked this conversation as resolved.
# Option 3 — transparent proxy routing.
# Deployer injects these; agent code is identical in all environments.
_PROXY_URL_ENV = "AICORE_PROXY_URL"
_PROXY_API_KEY_ENV = "AICORE_PROXY_API_KEY"
_DESTINATION_NAME_ENV = "AICORE_DESTINATION_NAME"


def _is_transparent_tls() -> bool:
"""Return True when transparent TLS proxy mode is active."""
return os.environ.get(TRANSPARENT_TLS_ENV_VAR, "").strip().lower() in (
"1",
"true",
"yes",
)


def _get_secret(
env_var_name: str,
Expand Down Expand Up @@ -119,14 +139,25 @@ def _get_aicore_base_url(instance_name: str = "aicore-instance") -> str:
def set_aicore_config(instance_name: str = "aicore-instance") -> None:
"""Load AI Core credentials and activate content filtering.

Loads secrets from files or environment variables and sets them as
process env vars so ``litellm`` picks them up.
Detects which routing mode is active based on environment variables:

- ``AICORE_PROXY_URL`` set → **proxy mode**: routes all LiteLLM calls
through a LiteLLM proxy via ``litellm.api_base``; model strings are
passed verbatim. No AI Core credentials are written to the process
environment.

File mappings based on the Kubernetes secret structure:
clientid → AICORE_CLIENT_ID
clientsecret → AICORE_CLIENT_SECRET
url → AICORE_AUTH_URL
serviceurls (JSON with AI_API_URL) → AICORE_BASE_URL
- ``AICORE_DESTINATION_NAME`` set → **destination mode**: loads AI Core
credentials from a BTP Destination Service destination at startup.
The deployer only needs to inject Destination Service binding credentials;
the AI Core ``client_secret`` never needs to be in the K8s Secret.

- Neither set → **direct mode** (existing behaviour): credentials are
loaded from a mounted K8s secret volume or environment variables.
``AICORE_TRANSPARENT_TLS=true`` suppresses ``client_secret`` and
relies on an mTLS sidecar.

Agent code is identical in all three modes — the deployer controls
routing by choosing which env vars to inject.

After credentials are loaded, content filtering is activated on every
``sap/*`` LiteLLM call at the configured thresholds (default: severity
Expand All @@ -136,44 +167,130 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None:
to turn filtering off at runtime, or set ``AICORE_FILTER_ENABLED=false``
to keep it off entirely.
"""
# Load secrets
proxy_url = os.environ.get(_PROXY_URL_ENV, "")
destination_name = os.environ.get(_DESTINATION_NAME_ENV, "")

if proxy_url:
_configure_proxy_mode(proxy_url)
elif destination_name:
_configure_destination_mode(destination_name)
else:
_configure_direct_mode(instance_name)

set_filtering()


def _configure_proxy_mode(proxy_url: str) -> None:
"""Configure LiteLLM to route calls through an external proxy.

Sets ``litellm.api_base`` / ``litellm.api_key`` globally.
Model strings (e.g. ``sap/<model>``) are passed verbatim — no rewrite.
No AI Core credentials are written to env.
"""
import litellm as _litellm

api_key = os.environ.get(_PROXY_API_KEY_ENV, "")
_litellm.api_base = proxy_url
if api_key:
_litellm.api_key = api_key
logger.info("AI Core proxy mode active — routing via %s", proxy_url)


def _configure_destination_mode(name: str) -> None:
"""Load AI Core credentials from a BTP Destination Service destination.

Calls the Destination Service at startup to resolve the named destination
and extracts ``clientId``, ``clientSecret``, ``tokenServiceURL``, and the
AI Core ``URL`` from the destination configuration properties. These are
written to the standard ``AICORE_*`` env vars so that LiteLLM can fetch
an OAuth token from XSUAA as usual.

Security: The deployer does NOT need to inject ``AICORE_CLIENT_SECRET``
directly — only Destination Service binding credentials are required in
the agent environment.

Raises ``RuntimeError`` if the destination is not found or does not
return ``clientId`` / ``clientSecret``.
"""
from sap_cloud_sdk.destination import create_client # lazy import

client = create_client()
dest = client.get_destination(name)

if dest is None:
raise RuntimeError(
f"AI Core destination '{name}' not found in Destination Service. "
"Check that the destination exists and the binding has access."
)

base_url = dest.url or ""
if base_url and not base_url.endswith("/v2"):
base_url = base_url.rstrip("/") + "/v2"
if base_url:
os.environ["AICORE_BASE_URL"] = base_url
Comment thread
tiagoek marked this conversation as resolved.

resource_group = dest.properties.get("resource_group", "default")
os.environ["AICORE_RESOURCE_GROUP"] = resource_group

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Intentional — same write pattern as _configure_direct_mode (lines 275–291). These writes initialise LiteLLM env vars from service binding credentials; secret_resolver is downstream of this layer. Skill updated (FP-Q-01) to skip write assignments.


client_id = dest.properties.get("clientId", "")
client_secret = dest.properties.get("clientSecret", "")
token_service_url = dest.properties.get("tokenServiceURL", "")

if not client_id or not client_secret:
raise RuntimeError(
f"Destination '{name}' did not return clientId/clientSecret. "
"Ensure the destination uses OAuth2ClientCredentials authentication "
"and the calling app has the Destination Service technical-user scope."
)

os.environ["AICORE_CLIENT_ID"] = client_id

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Intentional — same write pattern as _configure_direct_mode (lines 275–291). These writes initialise LiteLLM env vars from service binding credentials; secret_resolver is downstream of this layer. Skill updated (FP-Q-01) to skip write assignments.

os.environ["AICORE_CLIENT_SECRET"] = client_secret

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Intentional — same write pattern as _configure_direct_mode (lines 275–291). These writes initialise LiteLLM env vars from service binding credentials; secret_resolver is downstream of this layer. Skill updated (FP-Q-01) to skip write assignments.


if token_service_url:
if not token_service_url.endswith("/oauth/token"):
token_service_url = token_service_url.rstrip("/") + "/oauth/token"
Comment thread
tiagoek marked this conversation as resolved.
Comment thread
tiagoek marked this conversation as resolved.
os.environ["AICORE_AUTH_URL"] = token_service_url

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Intentional — same write pattern as _configure_direct_mode (lines 275–291). These writes initialise LiteLLM env vars from service binding credentials; secret_resolver is downstream of this layer. Skill updated (FP-Q-01) to skip write assignments.


logger.info("AI Core destination mode active — credentials loaded from '%s'", name)


def _configure_direct_mode(instance_name: str) -> None:
"""Load AI Core credentials directly from mounted secrets or env vars."""
transparent_tls = _is_transparent_tls()

client_id = _get_secret("AICORE_CLIENT_ID", "clientid", instance_name=instance_name)
client_secret = _get_secret(
"AICORE_CLIENT_SECRET", "clientsecret", instance_name=instance_name
)
auth_url = _get_secret("AICORE_AUTH_URL", "url", instance_name=instance_name)
base_url = _get_aicore_base_url(instance_name)
resource_group = _get_secret(
"AICORE_RESOURCE_GROUP", default="default", instance_name=instance_name
)

# Ensure AICORE_AUTH_URL has /oauth/token suffix
if auth_url and not auth_url.endswith("/oauth/token"):
auth_url = auth_url.rstrip("/") + "/oauth/token"

if base_url and not base_url.endswith("/v2"):
base_url = base_url.rstrip("/") + "/v2"

# Set environment variables for LiteLLM
if client_id:
os.environ["AICORE_CLIENT_ID"] = client_id
if client_secret:
os.environ["AICORE_CLIENT_SECRET"] = client_secret
if auth_url:
os.environ["AICORE_AUTH_URL"] = auth_url
if base_url:
os.environ["AICORE_BASE_URL"] = base_url
if resource_group:
os.environ["AICORE_RESOURCE_GROUP"] = resource_group

# Log configuration completion (excluding sensitive information)
logger.info("AI Core configuration has been set successfully")
if transparent_tls:
os.environ.pop("AICORE_CLIENT_SECRET", None)
logger.info("AI Core transparent TLS mode active — client_secret not required")
else:
client_secret = _get_secret(
"AICORE_CLIENT_SECRET", "clientsecret", instance_name=instance_name
)
if client_secret:
os.environ["AICORE_CLIENT_SECRET"] = client_secret

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Intentional — same write pattern as _configure_direct_mode (lines 275–291). These writes initialise LiteLLM env vars from service binding credentials; secret_resolver is downstream of this layer. Skill updated (FP-Q-01) to skip write assignments.


# Activate content filtering for all sap/* LiteLLM model calls.
# AICORE_FILTER_ENABLED=false disables; AICORE_FILTER_* tune thresholds.
# Errors propagate — filtering misconfiguration should surface at startup
# rather than be swallowed silently.
set_filtering()
logger.info("AI Core configuration has been set successfully")


def _get_secret_dir_mtime(instance_name: str = "aicore-instance") -> float:
Expand Down
12 changes: 8 additions & 4 deletions src/sap_cloud_sdk/aicore/completion.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,9 +88,12 @@ def completion(*args: Any, **kwargs: Any) -> Any:
"""Wrapper around :func:`litellm.completion` that normalises filter errors
and handles credential rotation transparently.

On ``AuthenticationError`` (e.g. rotated client_secret), reloads
credentials from the mounted secret volume and retries once.
All other exceptions surface verbatim after the filter-error translation.
On ``AuthenticationError`` (e.g. rotated client_secret or mTLS cert),
reloads credentials from the mounted secret volume and retries once.

Model strings (e.g. ``sap/<model>``) are passed verbatim to LiteLLM in all
routing modes — proxy routing is handled by ``litellm.api_base`` configured
in :func:`set_aicore_config`, not by rewriting the model name.
"""
try:
return litellm.completion(*args, **kwargs)
Expand All @@ -107,7 +110,8 @@ def completion(*args: Any, **kwargs: Any) -> Any:
async def acompletion(*args: Any, **kwargs: Any) -> Any:
"""Async wrapper around :func:`litellm.acompletion`.

Same translation and credential-rotation semantics as :func:`completion`.
Same credential-rotation semantics as :func:`completion`.
Model strings are passed verbatim to LiteLLM in all routing modes.
"""
try:
return await litellm.acompletion(*args, **kwargs)
Expand Down
54 changes: 54 additions & 0 deletions src/sap_cloud_sdk/aicore/user-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,60 @@ set_aicore_config(instance_name="aicore-production")

---

## Routing Modes

`set_aicore_config()` detects which routing mode to activate based on environment variables.
**Agent code is identical in all three modes** — the deployer controls routing by choosing which
env vars to inject.

### Direct mode (default)

No extra env vars required. Credentials are loaded from the mounted K8s secret volume
(`/etc/secrets/appfnd/aicore/<instance>/`) or from `AICORE_*` environment variables.
This is the standard mode for agents deployed on BTP managed runtime.

```python
set_aicore_config() # reads mounted secret or AICORE_* env vars
```

### Proxy mode (`AICORE_PROXY_URL`)

Set `AICORE_PROXY_URL` to route all LiteLLM calls through a LiteLLM-compatible proxy.
No AI Core credentials are written to the process environment — the proxy handles
authentication. Optionally set `AICORE_PROXY_API_KEY` for proxy-level auth.

```bash
# Injected by the deployer (e.g. K8s ConfigMap / Helm values)
AICORE_PROXY_URL=https://your-litellm-proxy.example.com
AICORE_PROXY_API_KEY=sk-... # optional
```

```python
set_aicore_config() # detects AICORE_PROXY_URL, sets litellm.api_base
# AICORE_CLIENT_SECRET is NOT written to env in this mode
```

Model strings (`sap/<model>`) are passed verbatim — no rewriting.

### Destination mode (`AICORE_DESTINATION_NAME`)

Set `AICORE_DESTINATION_NAME` to load AI Core credentials at startup from a BTP Destination
Service destination. The deployer only needs to inject Destination Service binding credentials;
the AI Core `client_secret` never needs to be in the K8s Secret directly.

```bash
AICORE_DESTINATION_NAME=aicore-destination # name of the BTP destination
```

```python
set_aicore_config() # calls Destination Service, writes AICORE_* env vars
```

The destination must use **OAuth2ClientCredentials** authentication with `clientId`,
`clientSecret`, and `tokenServiceURL` as destination configuration properties.

---

## Credential Rotation

BTP rotates AI Core service binding credentials automatically. The SDK
Expand Down
Loading
Loading