From 608f9c21ad1d5119b0171f17450ba4408b85715c Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Mon, 19 Oct 2026 10:33:40 +0200 Subject: [PATCH 1/9] fix: content filtering with stream --- src/sap_cloud_sdk/aicore/completion.py | 75 ++++++++-- src/sap_cloud_sdk/aicore/filtering/_patch.py | 4 +- src/sap_cloud_sdk/aicore/user-guide.md | 36 +++-- tests/aicore/filtering/unit/test_filtering.py | 58 ++++++++ tests/aicore/unit/test_completion.py | 136 ++++++++++++++++++ 5 files changed, 284 insertions(+), 25 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index 70dc6266..080a8203 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -40,9 +40,15 @@ ... These are intentionally thin — every other keyword argument is forwarded -verbatim to ``litellm.completion`` / ``litellm.acompletion``, including -``stream=True`` (the streaming iterator is returned unchanged). No -telemetry is recorded here; the wrappers fire on every LLM call and a +verbatim to ``litellm.completion`` / ``litellm.acompletion``. When content +filtering is inactive, or when only input filtering is configured, +``stream=True`` is allowed — input filter rejections surface as +:class:`ContentFilteredError` through the stream iterator regardless. +When output filtering is active, the wrappers wrap the returned iterator +so that a ``finish_reason == "content_filter"`` chunk raises +:class:`ContentFilteredError` rather than being yielded silently. + +No telemetry is recorded here; the wrappers fire on every LLM call and a counter at this level would be both noisy and uninformative about adoption of the SDK. """ @@ -58,10 +64,46 @@ from sap_cloud_sdk.core.telemetry.module import Module from sap_cloud_sdk.core.telemetry.operation import Operation from .filtering.filters import _parse_input_filter_error +from .filtering import _patch as _filtering_patch logger = logging.getLogger(__name__) +def _output_filtering_active() -> bool: + """Return True only when output filtering is configured. + + Input-filter rejections already surface correctly through the stream + iterator (the server sends an in-stream error event). Output filtering + needs extra handling because the server signals it via a normal chunk + with ``finish_reason == "content_filter"``, which would otherwise be + silently yielded to the caller. + """ + cfg = _filtering_patch._active_cfg + return cfg is not None and cfg.output_filtering is not None + + +def _wrap_stream_for_output_filter(iterator: Any) -> Any: + """Wrap a sync stream iterator to raise ContentFilteredError on filtered chunks.""" + from .filtering.exceptions import ContentFilteredError + + for chunk in iterator: + choices = getattr(chunk, "choices", None) or [] + if choices and getattr(choices[0], "finish_reason", None) == "content_filter": + raise ContentFilteredError(direction="output", details={}, request_id=None) + yield chunk + + +async def _awrap_stream_for_output_filter(iterator: Any) -> Any: + """Wrap an async stream iterator to raise ContentFilteredError on filtered chunks.""" + from .filtering.exceptions import ContentFilteredError + + async for chunk in iterator: + choices = getattr(chunk, "choices", None) or [] + if choices and getattr(choices[0], "finish_reason", None) == "content_filter": + raise ContentFilteredError(direction="output", details={}, request_id=None) + yield chunk + + @record_metrics(Module.AICORE, Operation.AICORE_REACTIVE_RELOAD) def _reload_reactive() -> None: # Local import avoids circular dep: completion ← __init__ ← completion @@ -94,35 +136,50 @@ def completion(*args: Any, **kwargs: Any) -> Any: Model strings (e.g. ``sap/``) 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. + + When output filtering is active and ``stream=True``, the returned iterator + is wrapped so that a ``finish_reason == "content_filter"`` chunk raises + :class:`ContentFilteredError` instead of being yielded silently. """ try: - return litellm.completion(*args, **kwargs) + result = litellm.completion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - return litellm.completion(*args, **kwargs) + result = litellm.completion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc + if kwargs.get("stream") and _output_filtering_active(): + return _wrap_stream_for_output_filter(result) + return result async def acompletion(*args: Any, **kwargs: Any) -> Any: """Async wrapper around :func:`litellm.acompletion`. - Same credential-rotation semantics as :func:`completion`. - Model strings are passed verbatim to LiteLLM in all routing modes. + Same credential-rotation and output-filter streaming semantics as + :func:`completion`. Model strings are passed verbatim to LiteLLM in all + routing modes. + + When output filtering is active and ``stream=True``, the returned async + iterator is wrapped so that a ``finish_reason == "content_filter"`` chunk + raises :class:`ContentFilteredError` instead of being yielded silently. """ try: - return await litellm.acompletion(*args, **kwargs) + result = await litellm.acompletion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - return await litellm.acompletion(*args, **kwargs) + result = await litellm.acompletion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc + if kwargs.get("stream") and _output_filtering_active(): + return _awrap_stream_for_output_filter(result) + return result __all__ = ["completion", "acompletion"] diff --git a/src/sap_cloud_sdk/aicore/filtering/_patch.py b/src/sap_cloud_sdk/aicore/filtering/_patch.py index c3c1ab1d..cdf730be 100644 --- a/src/sap_cloud_sdk/aicore/filtering/_patch.py +++ b/src/sap_cloud_sdk/aicore/filtering/_patch.py @@ -75,8 +75,8 @@ def transform_request( modules = body["config"]["modules"] if isinstance(modules, list): - # Fallback mode (list of configs) — inject into primary config only. - modules[0]["filtering"] = filtering_dict + for m in modules: + m["filtering"] = filtering_dict else: modules["filtering"] = filtering_dict diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md index 095a9204..09d01ab2 100644 --- a/src/sap_cloud_sdk/aicore/user-guide.md +++ b/src/sap_cloud_sdk/aicore/user-guide.md @@ -360,10 +360,12 @@ except ContentFilteredError as e: return "Your request was blocked by content safety policy." ``` -The wrapper forwards every argument verbatim to `litellm.completion` -(including `stream=True`), and only intercepts the wrapped -`APIConnectionError` shape that LiteLLM produces for input-filter -rejections. All other exceptions surface unchanged. +The wrapper forwards every argument verbatim to `litellm.completion`, +including `stream=True`. When output filtering is active and `stream=True`, +the returned iterator is wrapped so that a `finish_reason == "content_filter"` +chunk raises `ContentFilteredError` instead of being silently yielded. Input +filtering works server-side regardless of streaming — no extra handling needed. +All other exceptions surface unchanged. `ContentFilteredError` exposes three attributes — `direction`, `details`, `request_id`. The `details` field contains severity scalars from the server, @@ -444,21 +446,27 @@ print(response.choices[0].message.content) ### Streaming Responses +Use `sap_cloud_sdk.aicore.completion` (not `litellm.completion` directly) so +that output-filter rejections surface as `ContentFilteredError` rather than +being silently yielded: + ```python -from sap_cloud_sdk.aicore import set_aicore_config -from litellm import completion +from sap_cloud_sdk.aicore import ContentFilteredError, completion, set_aicore_config set_aicore_config() -response = completion( - model="sap/gpt-4", - messages=[{"role": "user", "content": "Tell me a story"}], - stream=True, -) +try: + response = completion( + model="sap/gpt-4", + messages=[{"role": "user", "content": "Tell me a story"}], + stream=True, + ) -for chunk in response: - if chunk.choices[0].delta.content: - print(chunk.choices[0].delta.content, end="") + for chunk in response: + if chunk.choices[0].delta.content: + print(chunk.choices[0].delta.content, end="") +except ContentFilteredError as e: + print(f"Blocked ({e.direction}): {e.details}") ``` ### Using Different Models diff --git a/tests/aicore/filtering/unit/test_filtering.py b/tests/aicore/filtering/unit/test_filtering.py index 0f0fe6c9..b8954a18 100644 --- a/tests/aicore/filtering/unit/test_filtering.py +++ b/tests/aicore/filtering/unit/test_filtering.py @@ -133,3 +133,61 @@ def test_disable_when_never_enabled(self, monkeypatch): disable_filtering() assert litellm.GenAIHubOrchestrationConfig is _ORIGINAL_CONFIG assert _filters_mod._active_cfg is None + + +class TestTransformRequestMultiModule: + """Regression tests: filtering must be injected into every fallback module, not just modules[0].""" + + def _make_cfg(self) -> ContentFiltering: + return ContentFiltering( + input_filtering=InputFiltering(filters=[AzureContentFilter()]) + ) + + def _call_transform_request(self, modules_value): + """Call the filtering injection logic directly, bypassing the LiteLLM base class.""" + from sap_cloud_sdk.aicore.filtering import _patch as _p + from unittest.mock import patch + + body = {"config": {"modules": modules_value}} + instance = FilteringOrchestrationConfig() + + with patch( + "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", + return_value=body, + ): + return instance.transform_request( + model="sap/x", + messages=[], + optional_params={}, + litellm_params={}, + headers={}, + ) + + def test_single_module_dict_gets_filtering(self, monkeypatch): + _clear_aicore_env(monkeypatch) + set_filtering(self._make_cfg()) + modules = {"model_id": "primary-llm"} + result = self._call_transform_request(modules) + assert "filtering" in result["config"]["modules"] + + def test_multi_module_list_all_get_filtering(self, monkeypatch): + _clear_aicore_env(monkeypatch) + set_filtering(self._make_cfg()) + modules = [ + {"model_id": "primary-llm"}, + {"model_id": "fallback-llm-1"}, + {"model_id": "fallback-llm-2"}, + ] + result = self._call_transform_request(modules) + for i, m in enumerate(result["config"]["modules"]): + assert "filtering" in m, f"modules[{i}] missing filtering block" + + def test_multi_module_filtering_dict_is_identical_across_modules(self, monkeypatch): + _clear_aicore_env(monkeypatch) + set_filtering(self._make_cfg()) + modules = [{"model_id": "a"}, {"model_id": "b"}] + result = self._call_transform_request(modules) + assert ( + result["config"]["modules"][0]["filtering"] + == result["config"]["modules"][1]["filtering"] + ) diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 8238e922..5b89cb1c 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -275,6 +275,142 @@ def test_non_auth_error_does_not_trigger_reload(self): mock_reload.assert_not_called() +# --------------------------------------------------------------------------- +# Streaming guard — stream=True must not bypass output filtering +# --------------------------------------------------------------------------- + + +class TestStreamingGuard: + """Output-filter streaming: ContentFilteredError raised on content_filter chunk. + Input-only filtering: stream=True passes through unmodified. + """ + + @pytest.fixture(autouse=True) + def restore_filtering(self): + from sap_cloud_sdk.aicore.filtering._patch import _install + _install(None) + yield + _install(None) + + def _activate_input_only(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, InputFiltering, set_filtering, + ) + set_filtering(ContentFiltering( + input_filtering=InputFiltering(filters=[AzureContentFilter()]) + )) + + def _activate_output(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, OutputFiltering, set_filtering, + ) + set_filtering(ContentFiltering( + output_filtering=OutputFiltering(filters=[AzureContentFilter()]) + )) + + def _activate_input_and_output(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, InputFiltering, OutputFiltering, + set_filtering, + ) + set_filtering(ContentFiltering( + input_filtering=InputFiltering(filters=[AzureContentFilter()]), + output_filtering=OutputFiltering(filters=[AzureContentFilter()]), + )) + + def _make_chunk(self, finish_reason=None): + """Build a minimal object that looks like an OpenAI stream chunk.""" + choice = MagicMock() + choice.finish_reason = finish_reason + chunk = MagicMock() + chunk.choices = [choice] + return chunk + + # --- input-only: stream allowed, iterator returned as-is ---------------- + + def test_no_filtering_stream_allowed(self): + sentinel = object() + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel): + result = completion(model="sap/x", messages=[], stream=True) + assert result is sentinel + + def test_input_only_filtering_stream_allowed(self): + self._activate_input_only() + sentinel = object() + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel): + result = completion(model="sap/x", messages=[], stream=True) + assert result is sentinel + + # --- output filtering: iterator wrapped, normal chunks pass through ------ + + def test_output_filter_stream_normal_chunks_pass_through(self): + self._activate_output() + chunks = [self._make_chunk(finish_reason=None), self._make_chunk(finish_reason="stop")] + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=iter(chunks)): + stream = completion(model="sap/x", messages=[], stream=True) + collected = list(stream) + assert len(collected) == 2 + + def test_output_filter_stream_raises_on_content_filter_chunk(self): + from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError + self._activate_output() + chunks = [ + self._make_chunk(finish_reason=None), + self._make_chunk(finish_reason="content_filter"), + ] + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=iter(chunks)): + stream = completion(model="sap/x", messages=[], stream=True) + with pytest.raises(ContentFilteredError) as ei: + list(stream) + assert ei.value.direction == "output" + + def test_input_and_output_stream_raises_on_content_filter_chunk(self): + from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError + self._activate_input_and_output() + chunks = [self._make_chunk(finish_reason="content_filter")] + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=iter(chunks)): + stream = completion(model="sap/x", messages=[], stream=True) + with pytest.raises(ContentFilteredError): + list(stream) + + # --- async variants ------------------------------------------------------ + + def test_acompletion_input_only_stream_allowed(self): + sentinel = object() + + async def _fake_acompletion(*_, **__): # noqa: S7503 + return sentinel + + with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion): + result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) + assert result is sentinel + + def test_acompletion_output_filter_stream_raises_on_content_filter_chunk(self): + from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError + self._activate_output() + + chunks = [self._make_chunk(finish_reason="content_filter")] + + async def fake_iter(): + for c in chunks: + yield c + + async def fake(*args, **kwargs): + return fake_iter() + + with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=fake): + stream = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) + + async def collect(): + results = [] + async for chunk in stream: + results.append(chunk) + return results + + with pytest.raises(ContentFilteredError): + asyncio.run(collect()) + + # --------------------------------------------------------------------------- # Reactive reload on AuthenticationError — async # --------------------------------------------------------------------------- From 689db5a3a8e1ef6bbad8118d56fb34aa8b674b29 Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Mon, 19 Oct 2026 15:13:57 +0200 Subject: [PATCH 2/9] fix(aicore): extract request_id from filtered stream chunk The streaming ContentFilteredError was always raised with request_id=None. The SAP orchestration v2 SSE format includes request_id in every chunk (surfaced as chunk.id by the LiteLLM SAP handler), so extract it from the filtered chunk for auditability. --- src/sap_cloud_sdk/aicore/completion.py | 8 ++++++-- tests/aicore/unit/test_completion.py | 6 ++++-- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index 080a8203..bada323c 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -89,7 +89,9 @@ def _wrap_stream_for_output_filter(iterator: Any) -> Any: for chunk in iterator: choices = getattr(chunk, "choices", None) or [] if choices and getattr(choices[0], "finish_reason", None) == "content_filter": - raise ContentFilteredError(direction="output", details={}, request_id=None) + raise ContentFilteredError( + direction="output", details={}, request_id=getattr(chunk, "id", None) + ) yield chunk @@ -100,7 +102,9 @@ async def _awrap_stream_for_output_filter(iterator: Any) -> Any: async for chunk in iterator: choices = getattr(chunk, "choices", None) or [] if choices and getattr(choices[0], "finish_reason", None) == "content_filter": - raise ContentFilteredError(direction="output", details={}, request_id=None) + raise ContentFilteredError( + direction="output", details={}, request_id=getattr(chunk, "id", None) + ) yield chunk diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 5b89cb1c..e4bf1cb0 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -318,12 +318,13 @@ def _activate_input_and_output(self): output_filtering=OutputFiltering(filters=[AzureContentFilter()]), )) - def _make_chunk(self, finish_reason=None): + def _make_chunk(self, finish_reason=None, request_id=None): """Build a minimal object that looks like an OpenAI stream chunk.""" choice = MagicMock() choice.finish_reason = finish_reason chunk = MagicMock() chunk.choices = [choice] + chunk.id = request_id return chunk # --- input-only: stream allowed, iterator returned as-is ---------------- @@ -356,13 +357,14 @@ def test_output_filter_stream_raises_on_content_filter_chunk(self): self._activate_output() chunks = [ self._make_chunk(finish_reason=None), - self._make_chunk(finish_reason="content_filter"), + self._make_chunk(finish_reason="content_filter", request_id="req-stream-123"), ] with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=iter(chunks)): stream = completion(model="sap/x", messages=[], stream=True) with pytest.raises(ContentFilteredError) as ei: list(stream) assert ei.value.direction == "output" + assert ei.value.request_id == "req-stream-123" def test_input_and_output_stream_raises_on_content_filter_chunk(self): from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError From 50e274d9c191bac2b93cf7b11d7426c78100121a Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Mon, 19 Oct 2026 16:11:07 +0200 Subject: [PATCH 3/9] fix(aicore): block stream=True when output filtering is active MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wrapping the stream iterator only protected callers using the SDK's completion() wrapper. ChatLiteLLM and LangGraph agents call litellm.completion directly, bypassing the wrapper entirely, so content_filter chunks would be silently yielded. Moving the guard into FilteringOrchestrationConfig.transform_request makes it universal — it fires before the HTTP request is sent regardless of the caller, because transform_request runs at the LiteLLM transport layer. --- src/sap_cloud_sdk/aicore/completion.py | 81 ++--------- src/sap_cloud_sdk/aicore/filtering/_patch.py | 28 ++++ src/sap_cloud_sdk/aicore/user-guide.md | 17 ++- tests/aicore/unit/test_completion.py | 137 ++++++++++--------- 4 files changed, 127 insertions(+), 136 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index bada323c..4fb91668 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -40,13 +40,13 @@ ... These are intentionally thin — every other keyword argument is forwarded -verbatim to ``litellm.completion`` / ``litellm.acompletion``. When content -filtering is inactive, or when only input filtering is configured, -``stream=True`` is allowed — input filter rejections surface as -:class:`ContentFilteredError` through the stream iterator regardless. -When output filtering is active, the wrappers wrap the returned iterator -so that a ``finish_reason == "content_filter"`` chunk raises -:class:`ContentFilteredError` rather than being yielded silently. +verbatim to ``litellm.completion`` / ``litellm.acompletion``. When output +filtering is active, ``stream=True`` is not supported — the filtering +transport patch raises :class:`ContentFilteredError` upfront before the +request is sent. This applies universally to all callers (including +``ChatLiteLLM`` and LangGraph agents). Input filtering works with streaming +— filter rejections surface as :class:`ContentFilteredError` through the +stream iterator regardless. No telemetry is recorded here; the wrappers fire on every LLM call and a counter at this level would be both noisy and uninformative about @@ -64,50 +64,10 @@ from sap_cloud_sdk.core.telemetry.module import Module from sap_cloud_sdk.core.telemetry.operation import Operation from .filtering.filters import _parse_input_filter_error -from .filtering import _patch as _filtering_patch logger = logging.getLogger(__name__) -def _output_filtering_active() -> bool: - """Return True only when output filtering is configured. - - Input-filter rejections already surface correctly through the stream - iterator (the server sends an in-stream error event). Output filtering - needs extra handling because the server signals it via a normal chunk - with ``finish_reason == "content_filter"``, which would otherwise be - silently yielded to the caller. - """ - cfg = _filtering_patch._active_cfg - return cfg is not None and cfg.output_filtering is not None - - -def _wrap_stream_for_output_filter(iterator: Any) -> Any: - """Wrap a sync stream iterator to raise ContentFilteredError on filtered chunks.""" - from .filtering.exceptions import ContentFilteredError - - for chunk in iterator: - choices = getattr(chunk, "choices", None) or [] - if choices and getattr(choices[0], "finish_reason", None) == "content_filter": - raise ContentFilteredError( - direction="output", details={}, request_id=getattr(chunk, "id", None) - ) - yield chunk - - -async def _awrap_stream_for_output_filter(iterator: Any) -> Any: - """Wrap an async stream iterator to raise ContentFilteredError on filtered chunks.""" - from .filtering.exceptions import ContentFilteredError - - async for chunk in iterator: - choices = getattr(chunk, "choices", None) or [] - if choices and getattr(choices[0], "finish_reason", None) == "content_filter": - raise ContentFilteredError( - direction="output", details={}, request_id=getattr(chunk, "id", None) - ) - yield chunk - - @record_metrics(Module.AICORE, Operation.AICORE_REACTIVE_RELOAD) def _reload_reactive() -> None: # Local import avoids circular dep: completion ← __init__ ← completion @@ -140,50 +100,35 @@ def completion(*args: Any, **kwargs: Any) -> Any: Model strings (e.g. ``sap/``) 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. - - When output filtering is active and ``stream=True``, the returned iterator - is wrapped so that a ``finish_reason == "content_filter"`` chunk raises - :class:`ContentFilteredError` instead of being yielded silently. """ try: - result = litellm.completion(*args, **kwargs) + return litellm.completion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - result = litellm.completion(*args, **kwargs) + return litellm.completion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc - if kwargs.get("stream") and _output_filtering_active(): - return _wrap_stream_for_output_filter(result) - return result async def acompletion(*args: Any, **kwargs: Any) -> Any: """Async wrapper around :func:`litellm.acompletion`. - Same credential-rotation and output-filter streaming semantics as - :func:`completion`. Model strings are passed verbatim to LiteLLM in all - routing modes. - - When output filtering is active and ``stream=True``, the returned async - iterator is wrapped so that a ``finish_reason == "content_filter"`` chunk - raises :class:`ContentFilteredError` instead of being yielded silently. + Same credential-rotation semantics as :func:`completion`. + Model strings are passed verbatim to LiteLLM in all routing modes. """ try: - result = await litellm.acompletion(*args, **kwargs) + return await litellm.acompletion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - result = await litellm.acompletion(*args, **kwargs) + return await litellm.acompletion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc - if kwargs.get("stream") and _output_filtering_active(): - return _awrap_stream_for_output_filter(result) - return result __all__ = ["completion", "acompletion"] diff --git a/src/sap_cloud_sdk/aicore/filtering/_patch.py b/src/sap_cloud_sdk/aicore/filtering/_patch.py index cdf730be..ceeece89 100644 --- a/src/sap_cloud_sdk/aicore/filtering/_patch.py +++ b/src/sap_cloud_sdk/aicore/filtering/_patch.py @@ -26,6 +26,21 @@ back into :class:`ContentFilteredError` before it reaches caller code; the :func:`_parse_input_filter_error` helper in :mod:`.filters` does the JSON parsing. + +Streaming + output filtering +---------------------------- +The ``transform_response`` hook is never called for streaming responses — +LiteLLM routes those through a separate iterator path. This means output +filter rejections (``finish_reason == "content_filter"`` chunks) would be +silently yielded to the caller rather than raising +:class:`ContentFilteredError`. + +To prevent this bypass, ``transform_request`` raises +:class:`ContentFilteredError` upfront when both streaming and output +filtering are requested simultaneously. This guarantee is universal — it +applies to all callers (``sap_cloud_sdk.aicore.completion``, +``ChatLiteLLM``, LangGraph, etc.) because ``transform_request`` runs at +the LiteLLM transport layer. """ from __future__ import annotations @@ -73,6 +88,19 @@ def transform_request( if not filtering_dict: return body + # Streaming + output filtering: transform_response is never called for + # streams, so content_filter chunks would be silently yielded. Block + # the combination upfront — applies to all callers universally. + if ( + _active_cfg.output_filtering is not None + and body.get("config", {}).get("stream", {}).get("enabled") + ): + raise ContentFilteredError( + direction="output", + details={}, + request_id=None, + ) + modules = body["config"]["modules"] if isinstance(modules, list): for m in modules: diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md index 09d01ab2..4fea8150 100644 --- a/src/sap_cloud_sdk/aicore/user-guide.md +++ b/src/sap_cloud_sdk/aicore/user-guide.md @@ -361,10 +361,12 @@ except ContentFilteredError as e: ``` The wrapper forwards every argument verbatim to `litellm.completion`, -including `stream=True`. When output filtering is active and `stream=True`, -the returned iterator is wrapped so that a `finish_reason == "content_filter"` -chunk raises `ContentFilteredError` instead of being silently yielded. Input -filtering works server-side regardless of streaming — no extra handling needed. +including `stream=True`. When output filtering is active, `stream=True` is +not supported — a `ContentFilteredError(direction="output")` is raised before +the request is sent. This block applies universally to all callers (including +`ChatLiteLLM` and LangGraph agents) because it fires inside the LiteLLM +transport patch. Input filtering works with streaming — filter rejections +surface as `ContentFilteredError` regardless. All other exceptions surface unchanged. `ContentFilteredError` exposes three attributes — `direction`, `details`, @@ -446,9 +448,10 @@ print(response.choices[0].message.content) ### Streaming Responses -Use `sap_cloud_sdk.aicore.completion` (not `litellm.completion` directly) so -that output-filter rejections surface as `ContentFilteredError` rather than -being silently yielded: +`stream=True` is supported when no output filtering is configured. If output +filtering is active, streaming is blocked upfront with +`ContentFilteredError(direction="output")` — this applies to all callers +including `ChatLiteLLM` and LangGraph agents. ```python from sap_cloud_sdk.aicore import ContentFilteredError, completion, set_aicore_config diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index e4bf1cb0..239eff03 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -281,7 +281,7 @@ def test_non_auth_error_does_not_trigger_reload(self): class TestStreamingGuard: - """Output-filter streaming: ContentFilteredError raised on content_filter chunk. + """stream=True is blocked upfront when output filtering is active. Input-only filtering: stream=True passes through unmodified. """ @@ -318,16 +318,7 @@ def _activate_input_and_output(self): output_filtering=OutputFiltering(filters=[AzureContentFilter()]), )) - def _make_chunk(self, finish_reason=None, request_id=None): - """Build a minimal object that looks like an OpenAI stream chunk.""" - choice = MagicMock() - choice.finish_reason = finish_reason - chunk = MagicMock() - chunk.choices = [choice] - chunk.id = request_id - return chunk - - # --- input-only: stream allowed, iterator returned as-is ---------------- + # --- input-only or no filtering: stream=True allowed -------------------- def test_no_filtering_stream_allowed(self): sentinel = object() @@ -342,75 +333,99 @@ def test_input_only_filtering_stream_allowed(self): result = completion(model="sap/x", messages=[], stream=True) assert result is sentinel - # --- output filtering: iterator wrapped, normal chunks pass through ------ + def test_acompletion_input_only_stream_allowed(self): + sentinel = object() - def test_output_filter_stream_normal_chunks_pass_through(self): - self._activate_output() - chunks = [self._make_chunk(finish_reason=None), self._make_chunk(finish_reason="stop")] - with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=iter(chunks)): - stream = completion(model="sap/x", messages=[], stream=True) - collected = list(stream) - assert len(collected) == 2 + async def _fake_acompletion(*_, **__): # noqa: S7503 + return sentinel + + with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion): + result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) + assert result is sentinel - def test_output_filter_stream_raises_on_content_filter_chunk(self): + # --- output filtering + stream=True: blocked upfront -------------------- + + def test_output_filter_stream_blocked(self): from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError + from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig + from unittest.mock import patch as _patch + self._activate_output() - chunks = [ - self._make_chunk(finish_reason=None), - self._make_chunk(finish_reason="content_filter", request_id="req-stream-123"), - ] - with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=iter(chunks)): - stream = completion(model="sap/x", messages=[], stream=True) - with pytest.raises(ContentFilteredError) as ei: - list(stream) + + body = {"config": {"stream": {"enabled": True}, "modules": {"model_id": "x"}}} + instance = FilteringOrchestrationConfig() + + with _patch( + "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", + return_value=body, + ): + with pytest.raises(ContentFilteredError) as ei: + instance.transform_request( + model="sap/x", messages=[], optional_params={}, + litellm_params={}, headers={}, + ) assert ei.value.direction == "output" - assert ei.value.request_id == "req-stream-123" - def test_input_and_output_stream_raises_on_content_filter_chunk(self): + def test_input_and_output_filter_stream_blocked(self): from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError + from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig + from unittest.mock import patch as _patch + self._activate_input_and_output() - chunks = [self._make_chunk(finish_reason="content_filter")] - with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=iter(chunks)): - stream = completion(model="sap/x", messages=[], stream=True) - with pytest.raises(ContentFilteredError): - list(stream) - # --- async variants ------------------------------------------------------ + body = {"config": {"stream": {"enabled": True}, "modules": {"model_id": "x"}}} + instance = FilteringOrchestrationConfig() - def test_acompletion_input_only_stream_allowed(self): - sentinel = object() - - async def _fake_acompletion(*_, **__): # noqa: S7503 - return sentinel + with _patch( + "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", + return_value=body, + ): + with pytest.raises(ContentFilteredError) as ei: + instance.transform_request( + model="sap/x", messages=[], optional_params={}, + litellm_params={}, headers={}, + ) + assert ei.value.direction == "output" - with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion): - result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) - assert result is sentinel + def test_output_filter_no_stream_not_blocked(self): + from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig + from unittest.mock import patch as _patch - def test_acompletion_output_filter_stream_raises_on_content_filter_chunk(self): - from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError self._activate_output() - chunks = [self._make_chunk(finish_reason="content_filter")] + body = {"config": {"modules": {"model_id": "x"}}} + instance = FilteringOrchestrationConfig() - async def fake_iter(): - for c in chunks: - yield c + with _patch( + "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", + return_value=body, + ): + result = instance.transform_request( + model="sap/x", messages=[], optional_params={}, + litellm_params={}, headers={}, + ) + assert "filtering" in result["config"]["modules"] - async def fake(*args, **kwargs): - return fake_iter() + def test_acompletion_output_filter_stream_blocked(self): + from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError + from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig + from unittest.mock import patch as _patch - with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=fake): - stream = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) + self._activate_output() - async def collect(): - results = [] - async for chunk in stream: - results.append(chunk) - return results + body = {"config": {"stream": {"enabled": True}, "modules": {"model_id": "x"}}} + instance = FilteringOrchestrationConfig() - with pytest.raises(ContentFilteredError): - asyncio.run(collect()) + with _patch( + "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", + return_value=body, + ): + with pytest.raises(ContentFilteredError) as ei: + instance.transform_request( + model="sap/x", messages=[], optional_params={}, + litellm_params={}, headers={}, + ) + assert ei.value.direction == "output" # --------------------------------------------------------------------------- From 321b0ac16f54e3de0c431977bf4bb12061863b90 Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Mon, 19 Oct 2026 16:39:16 +0200 Subject: [PATCH 4/9] chore(aicore): fix ruff lint/format issues in filtering patch and tests Remove unused imports flagged by ruff (F401) and reformat the multi-line if condition in _patch.py to satisfy ruff format. --- src/sap_cloud_sdk/aicore/filtering/_patch.py | 7 +++---- tests/aicore/filtering/unit/test_filtering.py | 1 - tests/aicore/unit/test_completion.py | 2 +- 3 files changed, 4 insertions(+), 6 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/filtering/_patch.py b/src/sap_cloud_sdk/aicore/filtering/_patch.py index ceeece89..3f3d7575 100644 --- a/src/sap_cloud_sdk/aicore/filtering/_patch.py +++ b/src/sap_cloud_sdk/aicore/filtering/_patch.py @@ -91,10 +91,9 @@ def transform_request( # Streaming + output filtering: transform_response is never called for # streams, so content_filter chunks would be silently yielded. Block # the combination upfront — applies to all callers universally. - if ( - _active_cfg.output_filtering is not None - and body.get("config", {}).get("stream", {}).get("enabled") - ): + if _active_cfg.output_filtering is not None and body.get("config", {}).get( + "stream", {} + ).get("enabled"): raise ContentFilteredError( direction="output", details={}, diff --git a/tests/aicore/filtering/unit/test_filtering.py b/tests/aicore/filtering/unit/test_filtering.py index b8954a18..6cb16024 100644 --- a/tests/aicore/filtering/unit/test_filtering.py +++ b/tests/aicore/filtering/unit/test_filtering.py @@ -145,7 +145,6 @@ def _make_cfg(self) -> ContentFiltering: def _call_transform_request(self, modules_value): """Call the filtering injection logic directly, bypassing the LiteLLM base class.""" - from sap_cloud_sdk.aicore.filtering import _patch as _p from unittest.mock import patch body = {"config": {"modules": modules_value}} diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 239eff03..88458c42 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -23,7 +23,7 @@ import asyncio import json -from unittest.mock import MagicMock, call, patch +from unittest.mock import MagicMock, patch import litellm import pytest From db6f33adb1d8c0e0a0c4bb876c79b024be27643a Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Mon, 19 Oct 2026 16:41:27 +0200 Subject: [PATCH 5/9] docs(aicore): fix missing blank line in filtering paragraph in user-guide --- src/sap_cloud_sdk/aicore/user-guide.md | 1 + 1 file changed, 1 insertion(+) diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md index 4fea8150..6304ec28 100644 --- a/src/sap_cloud_sdk/aicore/user-guide.md +++ b/src/sap_cloud_sdk/aicore/user-guide.md @@ -367,6 +367,7 @@ the request is sent. This block applies universally to all callers (including `ChatLiteLLM` and LangGraph agents) because it fires inside the LiteLLM transport patch. Input filtering works with streaming — filter rejections surface as `ContentFilteredError` regardless. + All other exceptions surface unchanged. `ContentFilteredError` exposes three attributes — `direction`, `details`, From dde2870a31b5526c4309a34b56733172820bf2c9 Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Wed, 21 Oct 2026 15:40:22 +0200 Subject: [PATCH 6/9] fix(aicore): read stream flag from optional_params before super() pops it LiteLLM's base transform_request pops "stream" from optional_params before returning, and _add_stream_param_to_request_body sets body["config"]["stream"] only *after* transform_request returns. The previous guard checked the body key that didn't exist yet, silently skipping the block. Fix: capture optional_params.get("stream") before calling super() so the guard fires correctly for all callers including ChatLiteLLM with streaming=True. --- src/sap_cloud_sdk/aicore/filtering/_patch.py | 12 +++++++++--- tests/aicore/unit/test_completion.py | 15 +++++++++------ 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/filtering/_patch.py b/src/sap_cloud_sdk/aicore/filtering/_patch.py index 3f3d7575..2fc1ffbf 100644 --- a/src/sap_cloud_sdk/aicore/filtering/_patch.py +++ b/src/sap_cloud_sdk/aicore/filtering/_patch.py @@ -73,6 +73,9 @@ def transform_request( litellm_params: dict, headers: dict, ) -> dict: + # Capture stream flag before super() pops it from optional_params. + is_streaming = bool(optional_params.get("stream")) + body = super().transform_request( model=model, messages=messages, @@ -91,9 +94,12 @@ def transform_request( # Streaming + output filtering: transform_response is never called for # streams, so content_filter chunks would be silently yielded. Block # the combination upfront — applies to all callers universally. - if _active_cfg.output_filtering is not None and body.get("config", {}).get( - "stream", {} - ).get("enabled"): + # Note: body["config"]["stream"] is set by _add_stream_param_to_request_body + # *after* transform_request returns, so we read stream from optional_params + # before super() pops it above. + if _active_cfg.output_filtering is not None and ( + is_streaming or body.get("config", {}).get("stream", {}).get("enabled") + ): raise ContentFilteredError( direction="output", details={}, diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 88458c42..6b5f1026 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -352,7 +352,10 @@ def test_output_filter_stream_blocked(self): self._activate_output() - body = {"config": {"stream": {"enabled": True}, "modules": {"model_id": "x"}}} + # stream=True is in optional_params — LiteLLM pops it before calling + # transform_request, so the body returned by super() never has + # body["config"]["stream"]. We must read it from optional_params first. + body = {"config": {"modules": {"model_id": "x"}}} instance = FilteringOrchestrationConfig() with _patch( @@ -361,7 +364,7 @@ def test_output_filter_stream_blocked(self): ): with pytest.raises(ContentFilteredError) as ei: instance.transform_request( - model="sap/x", messages=[], optional_params={}, + model="sap/x", messages=[], optional_params={"stream": True}, litellm_params={}, headers={}, ) assert ei.value.direction == "output" @@ -373,7 +376,7 @@ def test_input_and_output_filter_stream_blocked(self): self._activate_input_and_output() - body = {"config": {"stream": {"enabled": True}, "modules": {"model_id": "x"}}} + body = {"config": {"modules": {"model_id": "x"}}} instance = FilteringOrchestrationConfig() with _patch( @@ -382,7 +385,7 @@ def test_input_and_output_filter_stream_blocked(self): ): with pytest.raises(ContentFilteredError) as ei: instance.transform_request( - model="sap/x", messages=[], optional_params={}, + model="sap/x", messages=[], optional_params={"stream": True}, litellm_params={}, headers={}, ) assert ei.value.direction == "output" @@ -413,7 +416,7 @@ def test_acompletion_output_filter_stream_blocked(self): self._activate_output() - body = {"config": {"stream": {"enabled": True}, "modules": {"model_id": "x"}}} + body = {"config": {"modules": {"model_id": "x"}}} instance = FilteringOrchestrationConfig() with _patch( @@ -422,7 +425,7 @@ def test_acompletion_output_filter_stream_blocked(self): ): with pytest.raises(ContentFilteredError) as ei: instance.transform_request( - model="sap/x", messages=[], optional_params={}, + model="sap/x", messages=[], optional_params={"stream": True}, litellm_params={}, headers={}, ) assert ei.value.direction == "output" From 817c0f7885ffddd837e12c90b3da1d74b3854bec Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Thu, 22 Oct 2026 12:24:40 +0200 Subject: [PATCH 7/9] docs(aicore): clarify stream+output-filter limitation is SDK-level, not API-level The SAP AI Core API supports output filtering with streaming; the current block is a SDK limitation because mid-stream filter rejections (ValueError from LiteLLM's iterator) cannot be translated to ContentFilteredError for agent frameworks that call litellm directly. Update _patch.py docstring, exceptions.py docstring, and user-guide.md to reflect this accurately. --- src/sap_cloud_sdk/aicore/filtering/_patch.py | 34 ++++++++++--------- .../aicore/filtering/exceptions.py | 8 +++-- src/sap_cloud_sdk/aicore/user-guide.md | 21 +++++++----- 3 files changed, 37 insertions(+), 26 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/filtering/_patch.py b/src/sap_cloud_sdk/aicore/filtering/_patch.py index 2fc1ffbf..9d1fb7b5 100644 --- a/src/sap_cloud_sdk/aicore/filtering/_patch.py +++ b/src/sap_cloud_sdk/aicore/filtering/_patch.py @@ -29,18 +29,21 @@ Streaming + output filtering ---------------------------- -The ``transform_response`` hook is never called for streaming responses — -LiteLLM routes those through a separate iterator path. This means output -filter rejections (``finish_reason == "content_filter"`` chunks) would be -silently yielded to the caller rather than raising +The SAP AI Core Orchestration v2 API supports output filtering with streaming +— the server filters each chunk and signals a rejection via an in-stream error +event (no ``[DONE]``). LiteLLM surfaces this as a ``ValueError`` raised from +inside its streaming iterator, **not** via ``transform_response``. + +The SDK cannot intercept that iterator-level error universally: agent +frameworks (``ChatLiteLLM``, LangGraph, etc.) call ``litellm.completion`` +directly and never pass through ``sap_cloud_sdk.aicore.completion``, so the +``ValueError`` would surface as an untyped exception rather than :class:`ContentFilteredError`. -To prevent this bypass, ``transform_request`` raises -:class:`ContentFilteredError` upfront when both streaming and output -filtering are requested simultaneously. This guarantee is universal — it -applies to all callers (``sap_cloud_sdk.aicore.completion``, -``ChatLiteLLM``, LangGraph, etc.) because ``transform_request`` runs at -the LiteLLM transport layer. +Until the SDK wraps the streaming iterator for all call paths, +``transform_request`` raises :class:`ContentFilteredError` upfront when both +streaming and output filtering are active. This guarantee is universal because +``transform_request`` runs at the LiteLLM transport layer for all callers. """ from __future__ import annotations @@ -91,12 +94,11 @@ def transform_request( if not filtering_dict: return body - # Streaming + output filtering: transform_response is never called for - # streams, so content_filter chunks would be silently yielded. Block - # the combination upfront — applies to all callers universally. - # Note: body["config"]["stream"] is set by _add_stream_param_to_request_body - # *after* transform_request returns, so we read stream from optional_params - # before super() pops it above. + # Streaming + output filtering: the API supports it, but mid-stream + # filter rejections surface as ValueError inside LiteLLM's iterator — + # a path the SDK cannot intercept universally (agent frameworks call + # litellm directly, bypassing sap_cloud_sdk.aicore.completion). + # Block upfront until iterator wrapping is implemented for all callers. if _active_cfg.output_filtering is not None and ( is_streaming or body.get("config", {}).get("stream", {}).get("enabled") ): diff --git a/src/sap_cloud_sdk/aicore/filtering/exceptions.py b/src/sap_cloud_sdk/aicore/filtering/exceptions.py index 22430561..3368fe89 100644 --- a/src/sap_cloud_sdk/aicore/filtering/exceptions.py +++ b/src/sap_cloud_sdk/aicore/filtering/exceptions.py @@ -20,8 +20,12 @@ class ContentFilteredError(OrchestrationError): Attributes: direction: ``"input"`` if the user prompt was rejected (HTTP 4xx, ``error.location`` startswith ``"Filtering Module - Input Filter"``), - or ``"output"`` if the model response was rejected (HTTP 200, - ``finish_reason == "content_filter"``). + or ``"output"`` if the model response was rejected. In non-streaming + responses this is signalled by ``finish_reason == "content_filter"`` + (HTTP 200). When output filtering is active, streaming is blocked + upfront by the SDK (raised before the request is sent) because + mid-stream filter rejections cannot yet be intercepted uniformly + for all callers. details: Severity scalars from ``intermediate_results.{input,output}_filtering.data``. Safe to log; does NOT include raw prompt or completion content. request_id: ``error.request_id`` (input) or top-level ``request_id`` (output). diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md index 6304ec28..573c7df2 100644 --- a/src/sap_cloud_sdk/aicore/user-guide.md +++ b/src/sap_cloud_sdk/aicore/user-guide.md @@ -362,11 +362,13 @@ except ContentFilteredError as e: The wrapper forwards every argument verbatim to `litellm.completion`, including `stream=True`. When output filtering is active, `stream=True` is -not supported — a `ContentFilteredError(direction="output")` is raised before -the request is sent. This block applies universally to all callers (including -`ChatLiteLLM` and LangGraph agents) because it fires inside the LiteLLM -transport patch. Input filtering works with streaming — filter rejections -surface as `ContentFilteredError` regardless. +not currently supported by the SDK — a `ContentFilteredError(direction="output")` +is raised before the request is sent. The SAP AI Core API does support output +filtering with streaming, but mid-stream filter rejections surface as an untyped +error inside LiteLLM's iterator and cannot yet be intercepted uniformly for all +callers (agent frameworks such as `ChatLiteLLM` and LangGraph call `litellm` +directly, bypassing the SDK's exception translation). Input filtering works +with streaming — filter rejections surface as `ContentFilteredError` regardless. All other exceptions surface unchanged. @@ -450,9 +452,12 @@ print(response.choices[0].message.content) ### Streaming Responses `stream=True` is supported when no output filtering is configured. If output -filtering is active, streaming is blocked upfront with -`ContentFilteredError(direction="output")` — this applies to all callers -including `ChatLiteLLM` and LangGraph agents. +filtering is active, streaming is not currently supported by the SDK — +`ContentFilteredError(direction="output")` is raised before the request is +sent. The underlying API supports output filtering with streaming, but the SDK +cannot yet translate mid-stream filter rejections into `ContentFilteredError` +uniformly for all callers (agent frameworks call `litellm` directly and would +receive an untyped error). Input-only filtering works with streaming. ```python from sap_cloud_sdk.aicore import ContentFilteredError, completion, set_aicore_config From 38cb255082f7455acbe4a02ab76bfaa8cf57f209 Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Thu, 29 Oct 2026 16:04:40 -0300 Subject: [PATCH 8/9] fix(aicore): support output filtering with stream=True via iterator wrapping Remove the upfront ContentFilteredError block for stream+output-filter. Instead wrap the iterator returned by litellm when output filtering is active: _FilteringStreamWrapper (sync) and _AsyncFilteringStreamWrapper (async) catch the ValueError that SAPStreamIterator raises for in-stream output-filter rejections and translate it to ContentFilteredError. Add _parse_output_filter_stream_error() to _patch.py to parse the ValueError JSON shape (location: "Filtering Module - Output Filter"). Update TestStreamingGuard (stream+output now returns a wrapper, not a pre-request raise) and add TestStreamOutputFilterIterator + 7 new tests for the iterator wrapping path. Add TestParseOutputFilterStreamError in test_patch.py. Update user-guide.md and docstrings accordingly. --- src/sap_cloud_sdk/aicore/completion.py | 110 ++++++-- src/sap_cloud_sdk/aicore/filtering/_patch.py | 98 ++++--- .../aicore/filtering/exceptions.py | 7 +- src/sap_cloud_sdk/aicore/user-guide.md | 16 +- tests/aicore/filtering/unit/test_patch.py | 59 ++++- tests/aicore/unit/test_completion.py | 249 +++++++++++++----- 6 files changed, 406 insertions(+), 133 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index 4fb91668..d97dc633 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -5,15 +5,30 @@ calls ``raise_for_status()`` on 4xx responses **before** our patched ``transform_response`` runs, so the 400 surfaces in user code as a ``litellm.APIConnectionError`` whose ``str(exc)`` contains the JSON body of -the rejection. Output-filter rejections (HTTP 200 with -``finish_reason == "content_filter"``) go through ``transform_response`` -and surface as :class:`ContentFilteredError`. +the rejection. Output-filter rejections in buffered (non-streaming) mode +(HTTP 200 with ``finish_reason == "content_filter"``) go through +``transform_response`` and surface as :class:`ContentFilteredError`. That asymmetry would force callers to catch two exception types — these wrappers fix it by catching the wrapped exception inside the SDK and re-raising as :class:`ContentFilteredError` so callers can rely on a single exception type for "filter blocked you." +Streaming + output filtering +---------------------------- +When output filtering is active and ``stream=True``, the v2 API streams +chunks normally and signals a filter rejection via an in-stream error event +(no ``[DONE]``). LiteLLM's ``SAPStreamIterator`` surfaces this as a +``ValueError`` raised from inside the iterator. These wrappers detect when +output filtering is active and return a *filtering wrapper* around the +iterator that catches that ``ValueError`` and re-raises it as +:class:`ContentFilteredError`. + +Agent frameworks (``ChatLiteLLM``, LangGraph, etc.) that call +``litellm.completion`` directly — bypassing this wrapper — will see the raw +``ValueError`` from the iterator. They must use +:func:`sap_cloud_sdk.aicore.completion` to get the typed :class:`ContentFilteredError`. + Credential rotation handling ---------------------------- When a credential (client_secret) is rotated while the pod is running, @@ -40,13 +55,7 @@ ... These are intentionally thin — every other keyword argument is forwarded -verbatim to ``litellm.completion`` / ``litellm.acompletion``. When output -filtering is active, ``stream=True`` is not supported — the filtering -transport patch raises :class:`ContentFilteredError` upfront before the -request is sent. This applies universally to all callers (including -``ChatLiteLLM`` and LangGraph agents). Input filtering works with streaming -— filter rejections surface as :class:`ContentFilteredError` through the -stream iterator regardless. +verbatim to ``litellm.completion`` / ``litellm.acompletion``. No telemetry is recorded here; the wrappers fire on every LLM call and a counter at this level would be both noisy and uninformative about @@ -56,6 +65,7 @@ from __future__ import annotations import logging +from collections.abc import AsyncIterator, Iterator from typing import Any import litellm @@ -64,6 +74,8 @@ from sap_cloud_sdk.core.telemetry.module import Module from sap_cloud_sdk.core.telemetry.operation import Operation from .filtering.filters import _parse_input_filter_error +from .filtering import _patch as _filtering_patch +from .filtering._patch import _parse_output_filter_stream_error logger = logging.getLogger(__name__) @@ -90,6 +102,59 @@ def _maybe_translate_filter_error(exc: BaseException) -> BaseException: return blocked if blocked is not None else exc +def _output_filtering_active() -> bool: + cfg = _filtering_patch._active_cfg + return cfg is not None and cfg.output_filtering is not None + + +class _FilteringStreamWrapper: + """Sync iterator wrapper that converts in-stream output-filter errors.""" + + def __init__(self, stream: Iterator) -> None: + self._stream = stream + + def __iter__(self) -> Iterator: + return self + + def __next__(self) -> Any: + try: + return next(self._stream) + except StopIteration: + raise + except ValueError as exc: + translated = _parse_output_filter_stream_error(exc) + if translated is not None: + raise translated from exc + raise + + def __getattr__(self, name: str) -> Any: + return getattr(self._stream, name) + + +class _AsyncFilteringStreamWrapper: + """Async iterator wrapper that converts in-stream output-filter errors.""" + + def __init__(self, stream: AsyncIterator) -> None: + self._stream = stream + + def __aiter__(self) -> AsyncIterator: + return self + + async def __anext__(self) -> Any: + try: + return await self._stream.__anext__() + except StopAsyncIteration: + raise + except (ValueError, Exception) as exc: + translated = _parse_output_filter_stream_error(exc) + if translated is not None: + raise translated from exc + raise + + def __getattr__(self, name: str) -> Any: + return getattr(self._stream, name) + + def completion(*args: Any, **kwargs: Any) -> Any: """Wrapper around :func:`litellm.completion` that normalises filter errors and handles credential rotation transparently. @@ -97,38 +162,53 @@ def completion(*args: Any, **kwargs: Any) -> Any: On ``AuthenticationError`` (e.g. rotated client_secret or mTLS cert), reloads credentials from the mounted secret volume and retries once. + When output filtering is active and ``stream=True``, wraps the returned + iterator so that in-stream output-filter rejections surface as + :class:`ContentFilteredError` rather than a raw ``ValueError``. + Model strings (e.g. ``sap/``) 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. """ + is_streaming = kwargs.get("stream", False) try: - return litellm.completion(*args, **kwargs) + result = litellm.completion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - return litellm.completion(*args, **kwargs) + result = litellm.completion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc + if is_streaming and _output_filtering_active(): + return _FilteringStreamWrapper(result) + return result + async def acompletion(*args: Any, **kwargs: Any) -> Any: """Async wrapper around :func:`litellm.acompletion`. - Same credential-rotation semantics as :func:`completion`. + Same credential-rotation and output-filter stream-wrapping semantics as + :func:`completion`. Model strings are passed verbatim to LiteLLM in all routing modes. """ + is_streaming = kwargs.get("stream", False) try: - return await litellm.acompletion(*args, **kwargs) + result = await litellm.acompletion(*args, **kwargs) except litellm.AuthenticationError: _reload_reactive() - return await litellm.acompletion(*args, **kwargs) + result = await litellm.acompletion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: raise raise translated from exc + if is_streaming and _output_filtering_active(): + return _AsyncFilteringStreamWrapper(result) + return result + __all__ = ["completion", "acompletion"] diff --git a/src/sap_cloud_sdk/aicore/filtering/_patch.py b/src/sap_cloud_sdk/aicore/filtering/_patch.py index 9d1fb7b5..bd08e210 100644 --- a/src/sap_cloud_sdk/aicore/filtering/_patch.py +++ b/src/sap_cloud_sdk/aicore/filtering/_patch.py @@ -4,14 +4,14 @@ - Injects ``modules.filtering`` into every v2 completion request body via ``transform_request``. -- Detects filter rejections in responses and raises ``ContentFilteredError`` - via ``transform_response``. +- Detects filter rejections in buffered responses and raises + ``ContentFilteredError`` via ``transform_response``. The patch is applied by :func:`_install` (called by ``set_filtering`` / ``disable_filtering`` in :mod:`.filters`) and is idempotent — calling it multiple times with the same config is safe. -Two filter-rejection shapes (from the v2 API) are handled: +Two filter-rejection shapes (from the v2 API) are handled in buffered mode: - Input rejection: HTTP 4xx with ``error.location`` startswith ``"Filtering Module - Input Filter"``. @@ -29,25 +29,27 @@ Streaming + output filtering ---------------------------- -The SAP AI Core Orchestration v2 API supports output filtering with streaming -— the server filters each chunk and signals a rejection via an in-stream error -event (no ``[DONE]``). LiteLLM surfaces this as a ``ValueError`` raised from -inside its streaming iterator, **not** via ``transform_response``. - -The SDK cannot intercept that iterator-level error universally: agent -frameworks (``ChatLiteLLM``, LangGraph, etc.) call ``litellm.completion`` -directly and never pass through ``sap_cloud_sdk.aicore.completion``, so the -``ValueError`` would surface as an untyped exception rather than -:class:`ContentFilteredError`. - -Until the SDK wraps the streaming iterator for all call paths, -``transform_request`` raises :class:`ContentFilteredError` upfront when both -streaming and output filtering are active. This guarantee is universal because -``transform_request`` runs at the LiteLLM transport layer for all callers. +The SAP AI Core Orchestration v2 API supports output filtering with streaming. +The server filters each chunk and signals a rejection via an in-stream error +event (no ``[DONE]``). LiteLLM's ``SAPStreamIterator`` surfaces this error +event as a ``ValueError(json.dumps(event_obj))`` from inside the iterator. + +:func:`sap_cloud_sdk.aicore.completion` / :func:`acompletion` wrap the +returned stream iterator when output filtering is active, catching that +``ValueError`` and translating it to :class:`ContentFilteredError`. This gives +callers a single, typed exception for filter rejections in both streaming and +non-streaming modes. + +Note: agent frameworks (``ChatLiteLLM``, LangGraph, etc.) that call +``litellm.completion`` directly — bypassing the SDK's completion wrapper — +will see the raw ``ValueError`` from the iterator. They must either use +:func:`sap_cloud_sdk.aicore.completion` to get the translated error, or +handle the ``ValueError`` themselves. """ from __future__ import annotations +import json import logging from typing import Any @@ -76,9 +78,6 @@ def transform_request( litellm_params: dict, headers: dict, ) -> dict: - # Capture stream flag before super() pops it from optional_params. - is_streaming = bool(optional_params.get("stream")) - body = super().transform_request( model=model, messages=messages, @@ -94,20 +93,6 @@ def transform_request( if not filtering_dict: return body - # Streaming + output filtering: the API supports it, but mid-stream - # filter rejections surface as ValueError inside LiteLLM's iterator — - # a path the SDK cannot intercept universally (agent frameworks call - # litellm directly, bypassing sap_cloud_sdk.aicore.completion). - # Block upfront until iterator wrapping is implemented for all callers. - if _active_cfg.output_filtering is not None and ( - is_streaming or body.get("config", {}).get("stream", {}).get("enabled") - ): - raise ContentFilteredError( - direction="output", - details={}, - request_id=None, - ) - modules = body["config"]["modules"] if isinstance(modules, list): for m in modules: @@ -207,3 +192,46 @@ def _install(cfg: Any) -> None: # cfg: ContentFiltering | None else: litellm.GenAIHubOrchestrationConfig = FilteringOrchestrationConfig logger.info("content filtering active (FilteringOrchestrationConfig)") + + +def _parse_output_filter_stream_error( + exc: BaseException, +) -> ContentFilteredError | None: + """Try to parse an in-stream output-filter rejection from a ValueError. + + LiteLLM's ``SAPStreamIterator`` raises ``ValueError(json.dumps(event_obj))`` + when the server sends an error event inside the stream. The v2 error shape + for an output-filter rejection is:: + + {"error": {"request_id": "...", "code": 500, + "message": "...", + "location": "Filtering Module - Output Filter", + "intermediate_results": {"output_filtering": {"data": {...}}}}} + + Returns a :class:`ContentFilteredError` when the pattern matches, otherwise + ``None`` so the caller can re-raise the original exception unchanged. + """ + if not isinstance(exc, (ValueError, Exception)): + return None + msg = str(exc) + brace = msg.find("{") + if brace == -1: + return None + try: + payload = json.loads(msg[brace:]) + except (ValueError, TypeError): + return None + err = payload.get("error", {}) + if not isinstance(err, dict): + return None + location = err.get("location") or "" + if not location.startswith("Filtering Module - Output Filter"): + return None + data = ( + err.get("intermediate_results", {}).get("output_filtering", {}).get("data", {}) + ) + return ContentFilteredError( + direction="output", + details=data, + request_id=err.get("request_id"), + ) diff --git a/src/sap_cloud_sdk/aicore/filtering/exceptions.py b/src/sap_cloud_sdk/aicore/filtering/exceptions.py index 3368fe89..51673d28 100644 --- a/src/sap_cloud_sdk/aicore/filtering/exceptions.py +++ b/src/sap_cloud_sdk/aicore/filtering/exceptions.py @@ -22,10 +22,9 @@ class ContentFilteredError(OrchestrationError): ``error.location`` startswith ``"Filtering Module - Input Filter"``), or ``"output"`` if the model response was rejected. In non-streaming responses this is signalled by ``finish_reason == "content_filter"`` - (HTTP 200). When output filtering is active, streaming is blocked - upfront by the SDK (raised before the request is sent) because - mid-stream filter rejections cannot yet be intercepted uniformly - for all callers. + (HTTP 200). In streaming responses, the server sends an in-stream + error event which :func:`sap_cloud_sdk.aicore.completion` translates + into this exception from inside the iterator. details: Severity scalars from ``intermediate_results.{input,output}_filtering.data``. Safe to log; does NOT include raw prompt or completion content. request_id: ``error.request_id`` (input) or top-level ``request_id`` (output). diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md index 573c7df2..8b7c1f40 100644 --- a/src/sap_cloud_sdk/aicore/user-guide.md +++ b/src/sap_cloud_sdk/aicore/user-guide.md @@ -451,13 +451,15 @@ print(response.choices[0].message.content) ### Streaming Responses -`stream=True` is supported when no output filtering is configured. If output -filtering is active, streaming is not currently supported by the SDK — -`ContentFilteredError(direction="output")` is raised before the request is -sent. The underlying API supports output filtering with streaming, but the SDK -cannot yet translate mid-stream filter rejections into `ContentFilteredError` -uniformly for all callers (agent frameworks call `litellm` directly and would -receive an untyped error). Input-only filtering works with streaming. +`stream=True` is fully supported, including when output filtering is active. +When the server rejects a streamed response via an in-stream filter event, the +SDK raises `ContentFilteredError(direction="output")` from inside the iterator +— the same exception type used for buffered (non-streaming) rejections. + +**Important:** this translation only applies when you use +`sap_cloud_sdk.aicore.completion` / `acompletion`. If agent frameworks such as +`ChatLiteLLM` or LangGraph call `litellm.completion` directly, mid-stream +output-filter rejections surface as a raw `ValueError` from the iterator. ```python from sap_cloud_sdk.aicore import ContentFilteredError, completion, set_aicore_config diff --git a/tests/aicore/filtering/unit/test_patch.py b/tests/aicore/filtering/unit/test_patch.py index 1d6555d0..34140aba 100644 --- a/tests/aicore/filtering/unit/test_patch.py +++ b/tests/aicore/filtering/unit/test_patch.py @@ -1,4 +1,4 @@ -"""Unit tests for the LiteLLM patch (FilteringOrchestrationConfig, _install, _parse_input_filter_error).""" +"""Unit tests for the LiteLLM patch (FilteringOrchestrationConfig, _install, _parse_input_filter_error, _parse_output_filter_stream_error).""" import json import pytest @@ -17,6 +17,7 @@ FilteringOrchestrationConfig, _install, _ORIGINAL_CONFIG, + _parse_output_filter_stream_error, ) from sap_cloud_sdk.aicore.filtering.config import load_from_env from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError @@ -284,3 +285,59 @@ def test_install_idempotent(self): _install(cfg) # second call — no error assert litellm.GenAIHubOrchestrationConfig is FilteringOrchestrationConfig _install(None) + + +# --------------------------------------------------------------------------- +# _parse_output_filter_stream_error tests +# --------------------------------------------------------------------------- + + +OUTPUT_FILTER_STREAM_ERROR = { + "error": { + "request_id": "req-stream-output", + "code": 500, + "message": "Content filtered by output filter.", + "location": "Filtering Module - Output Filter", + "intermediate_results": { + "output_filtering": { + "data": { + "azure_content_safety": {"Hate": 0, "Violence": 6} + } + } + }, + } +} + + +class TestParseOutputFilterStreamError: + def _make_exc(self, payload: dict) -> ValueError: + return ValueError(json.dumps(payload)) + + def test_extracts_output_filter(self): + exc = self._make_exc(OUTPUT_FILTER_STREAM_ERROR) + blocked = _parse_output_filter_stream_error(exc) + assert blocked is not None + assert blocked.direction == "output" + assert blocked.request_id == "req-stream-output" + assert blocked.details["azure_content_safety"]["Violence"] == 6 + + def test_returns_none_for_non_output_location(self): + body = {"error": {"location": "Filtering Module - Input Filter", "message": "x"}} + assert _parse_output_filter_stream_error(self._make_exc(body)) is None + + def test_returns_none_for_non_filter_error(self): + body = {"error": {"location": "Model Module", "message": "model error"}} + assert _parse_output_filter_stream_error(self._make_exc(body)) is None + + def test_returns_none_for_no_brace(self): + assert _parse_output_filter_stream_error(ValueError("network error")) is None + + def test_returns_none_for_malformed_json(self): + assert _parse_output_filter_stream_error(ValueError("{not valid json}")) is None + + def test_returns_none_for_non_exception(self): + assert _parse_output_filter_stream_error(KeyboardInterrupt()) is None + + def test_stream_with_no_output_filter_location_passes_through(self): + body = {"code": 500, "message": "some server error"} + assert _parse_output_filter_stream_error(self._make_exc(body)) is None diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 6b5f1026..fc9cadf1 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -5,7 +5,8 @@ whether the rejection happened on input (litellm wraps it in ``APIConnectionError`` because the 4xx triggers ``raise_for_status()`` before our transport patch runs) or output (already raised by the -transport patch as :class:`ContentFilteredError`). Test focus: +transport patch as :class:`ContentFilteredError` in buffered mode, or +raised from the stream iterator in streaming mode). Test focus: - Successful calls pass through verbatim. - An input-filter-shaped ``APIConnectionError`` is re-raised as @@ -17,6 +18,9 @@ - ``acompletion`` exhibits the same behaviour on the async path. - On ``AuthenticationError``, credentials are reloaded and the call is retried once (credential rotation without pod restart). +- With output filtering active and ``stream=True``, the returned iterator + catches in-stream ``ValueError`` filter errors and raises + :class:`ContentFilteredError`. """ from __future__ import annotations @@ -276,12 +280,13 @@ def test_non_auth_error_does_not_trigger_reload(self): # --------------------------------------------------------------------------- -# Streaming guard — stream=True must not bypass output filtering +# Streaming guard — stream=True + output filtering now supported # --------------------------------------------------------------------------- class TestStreamingGuard: - """stream=True is blocked upfront when output filtering is active. + """stream=True is now allowed with output filtering active. + The returned iterator wraps filter errors into ContentFilteredError. Input-only filtering: stream=True passes through unmodified. """ @@ -318,7 +323,7 @@ def _activate_input_and_output(self): output_filtering=OutputFiltering(filters=[AzureContentFilter()]), )) - # --- input-only or no filtering: stream=True allowed -------------------- + # --- no filtering or input-only: stream=True passes through -------------------- def test_no_filtering_stream_allowed(self): sentinel = object() @@ -336,99 +341,201 @@ def test_input_only_filtering_stream_allowed(self): def test_acompletion_input_only_stream_allowed(self): sentinel = object() - async def _fake_acompletion(*_, **__): # noqa: S7503 + async def _fake_acompletion(*_, **__): return sentinel with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion): result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) assert result is sentinel - # --- output filtering + stream=True: blocked upfront -------------------- + # --- output filtering + stream=True: iterator returned (not blocked upfront) -- - def test_output_filter_stream_blocked(self): - from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError - from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig - from unittest.mock import patch as _patch + def test_output_filter_stream_returns_wrapper(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper self._activate_output() + fake_stream = iter([]) + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=fake_stream): + result = completion(model="sap/x", messages=[], stream=True) + assert isinstance(result, _FilteringStreamWrapper) - # stream=True is in optional_params — LiteLLM pops it before calling - # transform_request, so the body returned by super() never has - # body["config"]["stream"]. We must read it from optional_params first. - body = {"config": {"modules": {"model_id": "x"}}} - instance = FilteringOrchestrationConfig() + def test_input_and_output_filter_stream_returns_wrapper(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper - with _patch( - "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", - return_value=body, - ): - with pytest.raises(ContentFilteredError) as ei: - instance.transform_request( - model="sap/x", messages=[], optional_params={"stream": True}, - litellm_params={}, headers={}, - ) - assert ei.value.direction == "output" + self._activate_input_and_output() + fake_stream = iter([]) + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=fake_stream): + result = completion(model="sap/x", messages=[], stream=True) + assert isinstance(result, _FilteringStreamWrapper) - def test_input_and_output_filter_stream_blocked(self): - from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError - from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig - from unittest.mock import patch as _patch + def test_output_filter_no_stream_not_wrapped(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper - self._activate_input_and_output() + self._activate_output() + sentinel = object() + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel): + result = completion(model="sap/x", messages=[]) + assert result is sentinel + assert not isinstance(result, _FilteringStreamWrapper) - body = {"config": {"modules": {"model_id": "x"}}} - instance = FilteringOrchestrationConfig() + def test_acompletion_output_filter_stream_returns_async_wrapper(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper - with _patch( - "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", - return_value=body, - ): - with pytest.raises(ContentFilteredError) as ei: - instance.transform_request( - model="sap/x", messages=[], optional_params={"stream": True}, - litellm_params={}, headers={}, - ) + self._activate_output() + + async def _fake_stream_aiter(): + return + yield # make it an async generator + + async def _fake_acompletion(*_, **__): + return _fake_stream_aiter() + + with patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=_fake_acompletion): + result = asyncio.run(acompletion(model="sap/x", messages=[], stream=True)) + assert isinstance(result, _AsyncFilteringStreamWrapper) + + +# --------------------------------------------------------------------------- +# Stream output-filter iterator — ValueError from stream → ContentFilteredError +# --------------------------------------------------------------------------- + + +def _output_filter_stream_error_json() -> str: + """In-stream error event JSON for an output-filter rejection (v2 shape).""" + return json.dumps({ + "error": { + "request_id": "req-stream-out", + "code": 500, + "message": "Content filtered by output filter.", + "location": "Filtering Module - Output Filter", + "intermediate_results": { + "output_filtering": { + "data": { + "azure_content_safety": {"Violence": 6, "Hate": 0} + } + } + }, + } + }) + + +class TestStreamOutputFilterIterator: + """The filtering stream wrapper catches output-filter ValueError from the iterator.""" + + @pytest.fixture(autouse=True) + def restore_filtering(self): + from sap_cloud_sdk.aicore.filtering._patch import _install + _install(None) + yield + _install(None) + + def _activate_output(self): + from sap_cloud_sdk.aicore.filtering import ( + AzureContentFilter, ContentFiltering, OutputFiltering, set_filtering, + ) + set_filtering(ContentFiltering( + output_filtering=OutputFiltering(filters=[AzureContentFilter()]) + )) + + # --- sync wrapper --- + + def test_sync_wrapper_raises_content_filtered_on_output_filter_error(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper + + def _bad_stream(): + raise ValueError(_output_filter_stream_error_json()) + yield # make it a generator + + wrapper = _FilteringStreamWrapper(_bad_stream()) + with pytest.raises(ContentFilteredError) as ei: + next(wrapper) assert ei.value.direction == "output" + assert ei.value.request_id == "req-stream-out" + assert ei.value.details["azure_content_safety"]["Violence"] == 6 - def test_output_filter_no_stream_not_blocked(self): - from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig - from unittest.mock import patch as _patch + def test_sync_wrapper_yields_normal_chunks(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper - self._activate_output() + chunks = ["chunk1", "chunk2"] + wrapper = _FilteringStreamWrapper(iter(chunks)) + assert list(wrapper) == chunks - body = {"config": {"modules": {"model_id": "x"}}} - instance = FilteringOrchestrationConfig() + def test_sync_wrapper_passes_through_non_filter_value_error(self): + from sap_cloud_sdk.aicore.completion import _FilteringStreamWrapper - with _patch( - "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", - return_value=body, - ): - result = instance.transform_request( - model="sap/x", messages=[], optional_params={}, - litellm_params={}, headers={}, - ) - assert "filtering" in result["config"]["modules"] + def _bad_stream(): + raise ValueError("some unrelated error") + yield - def test_acompletion_output_filter_stream_blocked(self): - from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError - from sap_cloud_sdk.aicore.filtering._patch import FilteringOrchestrationConfig - from unittest.mock import patch as _patch + wrapper = _FilteringStreamWrapper(_bad_stream()) + with pytest.raises(ValueError, match="some unrelated error"): + next(wrapper) + def test_completion_stream_raises_content_filtered_on_iteration(self): + """End-to-end: completion() + output filtering + stream=True → ContentFilteredError on next().""" self._activate_output() - body = {"config": {"modules": {"model_id": "x"}}} - instance = FilteringOrchestrationConfig() + def _bad_stream(): + raise ValueError(_output_filter_stream_error_json()) + yield - with _patch( - "sap_cloud_sdk.aicore.filtering._patch.GenAIHubOrchestrationConfig.transform_request", - return_value=body, - ): - with pytest.raises(ContentFilteredError) as ei: - instance.transform_request( - model="sap/x", messages=[], optional_params={"stream": True}, - litellm_params={}, headers={}, - ) + with patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=_bad_stream()): + stream = completion(model="sap/x", messages=[], stream=True) + + with pytest.raises(ContentFilteredError) as ei: + next(stream) + assert ei.value.direction == "output" + assert ei.value.request_id == "req-stream-out" + + # --- async wrapper --- + + def test_async_wrapper_raises_content_filtered_on_output_filter_error(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper + + async def _bad_async_stream(): + raise ValueError(_output_filter_stream_error_json()) + yield # make it an async generator + + async def _run(): + wrapper = _AsyncFilteringStreamWrapper(_bad_async_stream()) + async for _ in wrapper: + pass + + with pytest.raises(ContentFilteredError) as ei: + asyncio.run(_run()) assert ei.value.direction == "output" + assert ei.value.request_id == "req-stream-out" + + def test_async_wrapper_yields_normal_chunks(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper + + async def _good_stream(): + for item in ["a", "b", "c"]: + yield item + + async def _run(): + result = [] + wrapper = _AsyncFilteringStreamWrapper(_good_stream()) + async for chunk in wrapper: + result.append(chunk) + return result + + assert asyncio.run(_run()) == ["a", "b", "c"] + + def test_async_wrapper_passes_through_non_filter_value_error(self): + from sap_cloud_sdk.aicore.completion import _AsyncFilteringStreamWrapper + + async def _bad_async_stream(): + raise ValueError("unrelated async error") + yield + + async def _run(): + wrapper = _AsyncFilteringStreamWrapper(_bad_async_stream()) + async for _ in wrapper: + pass + + with pytest.raises(ValueError, match="unrelated async error"): + asyncio.run(_run()) # --------------------------------------------------------------------------- From 8e1a4644af2d4d47d450f3609374a77932908b45 Mon Sep 17 00:00:00 2001 From: Nicole Gomes Date: Thu, 29 Oct 2026 16:15:26 -0300 Subject: [PATCH 9/9] docs(aicore): remove stale stream+output-filter limitation paragraph The paragraph added in docs(aicore): clarify stream+output-filter limitation said stream=True is not supported with output filtering and raises upfront. The final fix(aicore): support output filtering with stream=True via iterator wrapping made this untrue; the new Streaming Responses section already has the correct description. Remove the contradicting paragraph. --- src/sap_cloud_sdk/aicore/user-guide.md | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/user-guide.md b/src/sap_cloud_sdk/aicore/user-guide.md index 8b7c1f40..efc95d38 100644 --- a/src/sap_cloud_sdk/aicore/user-guide.md +++ b/src/sap_cloud_sdk/aicore/user-guide.md @@ -361,14 +361,9 @@ except ContentFilteredError as e: ``` The wrapper forwards every argument verbatim to `litellm.completion`, -including `stream=True`. When output filtering is active, `stream=True` is -not currently supported by the SDK — a `ContentFilteredError(direction="output")` -is raised before the request is sent. The SAP AI Core API does support output -filtering with streaming, but mid-stream filter rejections surface as an untyped -error inside LiteLLM's iterator and cannot yet be intercepted uniformly for all -callers (agent frameworks such as `ChatLiteLLM` and LangGraph call `litellm` -directly, bypassing the SDK's exception translation). Input filtering works -with streaming — filter rejections surface as `ContentFilteredError` regardless. +including `stream=True`. Both input and output filtering work with streaming — +filter rejections surface as `ContentFilteredError` regardless. See the +[Streaming Responses](#streaming-responses) section for details. All other exceptions surface unchanged.