scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

chore(ci): merge oss branch (#33784)

Browse source 
* fix(embeddings): accept encoding_format='float' for vertex_ai/gemini embeddings (#33617)

OpenAI SDKs (and litellm's own client since ~1.84) send
encoding_format='float' by default, but the vertex embedding config only
supports ['dimensions'], so get_optional_params_embeddings raised
UnsupportedParamsError at the provider default value. Any
OpenAI-compatible client talking to a litellm proxy with vertex
embedding models got a 400 unless the operator set proxy-wide
drop_params: true.

Float lists are exactly what the vertex API returns, so the param is a
no-op: pop it before validation. Other values (e.g. 'base64') keep the
existing unsupported-param behavior (dropped with drop_params, raise
otherwise).

Fixes #33173

Co-authored-by: Mihidum Hettiyahandi <55163074+mihidumh@users.noreply.github.com>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* feat(guardrails): add Singulr guardrail integration for LiteLLM gateway (#31302)

* singulr guardrail support for litellm gateway

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* fix comments

* improvement

* fix: resolve review comments and implement requested improvements

* fix:Guardrail bypass through uninspected messages

* fix:tool text scanning

* fix: Legacy function definitions bypass scanning by adding indirect message scaning

* chore: remove unintended basedpyright budget file

* fix:Response schema bypasses guardrail scanning (response_format.json_schema)

* chore: restore basedpyright-code-budget.json and update lint baselines

Restores the file deleted in c698b88686 to match upstream litellm_internal_staging.
Regenerates basedpyright and ruff-strict budget baselines via make lint-budget-update.

* fix: scan system messages as indirect prompt injection in Singulr guardrail

* chore: restore lint budget files to upstream baseline

* fix: resolve ruff UP006 and I001 violations in singulr guardrail

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* resolve review comments on Singulr guardrail

* fix: scan tool call results as indirect prompt injection in Singulr guardrail

* Apply suggestion from @greptile-apps[bot]

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* minor

* formating fix

* refactor: shift extraction logic to singulr side

* refactor:keep precall hook only

* fix:formatting

* fix:linting

* improve config description

* Trigger CI

* fix

* fix:field description

* fix:errors due to change in field names

* style: apply ruff line-wrap formatting to singulr guardrail

* fix:exception

* fix:formatting

* fix playground

* improved

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>

* fix

* fix ci issues

* remove uv.lock from pr

* fix

* fix:resolved comments

* chore: trigger CI

* remove uv.lock

* fix

* fix linting

* fix linting

* fix linting

* remove doc strings

* remove test fixes

* chore: retrigger CI

* change in singulr api contract

* remove some ut

* send litellm call_id to singulr

---------

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
Co-authored-by: aniket-kardile <aniket.kardile@singulr.ai>
Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>

* Fix non-conformant UUIDv7 generation in native Opik integration (#31294)

create_uuid7() encoded the timestamp in units of 16 seconds instead of
milliseconds, so the top 48 bits came out ~4096x the real unix-ms. Opik's
backend validates the embedded UUIDv7 timestamp on ingestion (OPIK-7067);
the bad encoding decoded to ~year 2201 and every trace/span batch was
rejected with HTTP 400.

Rewrite create_uuid7() to be RFC 9562 conformant (top 48 bits = unix-ms),
using the standard library only so no new dependency is added. Add unit
tests covering UUIDv7 validity and millisecond timestamp encoding.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(proxy): expose uvicorn concurrency limit (#33077)

Expose uvicorn's limit_concurrency as a --limit_concurrency CLI flag and
LIMIT_CONCURRENCY environment variable. Uvicorn counts both active tasks and
accepted connections and returns HTTP 503 once the configured limit is reached.

Reject non-positive limits at CLI parse time and only add the setting to the
uvicorn startup arguments. Because idle connections also consume capacity,
deployments should use upstream connection/header timeouts and per-client
connection limits.

* test: reorder test_utils tail to keep the daily merge conflict-free (#33788)

The daily OSS branch and litellm_internal_staging each appended an
independent test block at the very end of tests/test_litellm/test_utils.py,
so merging the two collides on that shared end-of-file position even though
the additions are unrelated (this branch adds the vertex embedding
encoding-format tests; staging adds the per-model prompt-cache-minimum
tests). Moving this branch's new TestVertexEmbeddingEncodingFormat class
above test_gemini_image_models_do_not_support_reasoning, which both branches
share, gives the two additions different anchors, so git applies both
without a conflict and without pulling staging into this branch. Pure
reorder; no test bodies change

---------

Co-authored-by: Mateo Wang <277851410+mateo-berri@users.noreply.github.com>
Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Mihidum Hettiyahandi <55163074+mihidumh@users.noreply.github.com>
Co-authored-by: madan-singulr <150280287+madan-singulr@users.noreply.github.com>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
Co-authored-by: aniket-kardile <aniket.kardile@singulr.ai>
Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>
Co-authored-by: Aliaksandr Kuzmik <98702584+alexkuzmik@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Salva Madrid <50212436+salvamadrid@users.noreply.github.com>
This commit is contained in:
yuneng-jiang 2026-07-17 16:22:13 -07:00 • committed by GitHub
parent 89c87ae59a
commit 04a5ebb94d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1081 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
litellm/integrations/opik/utils.py
View file

@ -1,40 +1,38 @@
import configparser
import os
import time
import uuid
from typing import Any, Dict, Final, List, Optional, Tuple
CONFIG_FILE_PATH_DEFAULT: Final[str] = "~/.opik.config"
def create_uuid7():
ns = time.time_ns()
last = [0, 0, 0, 0]
def create_uuid7() -> str:
"""Generate an RFC 9562 conformant UUIDv7 string.
# Simple uuid7 implementation
sixteen_secs = 16_000_000_000
t1, rest1 = divmod(ns, sixteen_secs)
t2, rest2 = divmod(rest1 << 16, sixteen_secs)
t3, _ = divmod(rest2 << 12, sixteen_secs)
t3 |= 7 << 12 # Put uuid version in top 4 bits, which are 0 in t3
The top 48 bits encode the Unix timestamp in milliseconds. Opik's backend
validates this embedded timestamp on ingestion (it must fall within a window
around "now"), so the encoding has to be correct or trace/span batches are
rejected with HTTP 400. Implemented with the standard library only, so no
extra dependency is added to litellm. See ``opik.id_helpers`` for the
reference implementation.
"""
unix_ts_ms = int(time.time() * 1000)
# The next two bytes are an int (t4) with two bits for
# the variant 2 and a 14 bit sequence counter which increments
# if the time is unchanged.
if t1 == last[0] and t2 == last[1] and t3 == last[2]:
# Stop the seq counter wrapping past 0x3FFF.
# This won't happen in practice, but if it does,
# uuids after the 16383rd with that same timestamp
# will not longer be correctly ordered but
# are still unique due to the 6 random bytes.
if last[3] < 0x3FFF:
last[3] += 1
else:
last[:] = (t1, t2, t3, 0)
t4 = (2 << 14) | last[3] # Put variant 0b10 in top two bits
# Fill the 16-byte buffer with random data, then overwrite the structured
# parts (timestamp, version, variant) defined by the UUIDv7 layout.
uuid_bytes = bytearray(os.urandom(16))
# Six random bytes for the lower part of the uuid
rand = os.urandom(6)
return f"{t1:>08x}-{t2:>04x}-{t3:>04x}-{t4:>04x}-{rand.hex()}"
# First 48 bits (6 bytes): Unix timestamp in milliseconds.
uuid_bytes[0:6] = unix_ts_ms.to_bytes(6, byteorder="big")
# Version 7 in the top 4 bits of byte 6.
uuid_bytes[6] = 0x70 | (uuid_bytes[6] & 0x0F)
# Variant 0b10 in the top 2 bits of byte 8.
uuid_bytes[8] = 0x80 | (uuid_bytes[8] & 0x3F)
return str(uuid.UUID(bytes=bytes(uuid_bytes)))
def _read_opik_config_file() -> Dict[str, str]:

50
litellm/proxy/guardrails/guardrail_hooks/singulr/__init__.py Normal file
View file

@ -0,0 +1,50 @@
"""
Author: Madan Singhal
Date: 23/06/26
"""
from typing import TYPE_CHECKING
from litellm.types.guardrails import SupportedGuardrailIntegrations
from .singulr import SingulrGuardrail
if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(
litellm_params: "LitellmParams",
guardrail: "Guardrail",
):
import litellm
_cb = SingulrGuardrail(
singulr_api_base=getattr(litellm_params, "singulr_api_base", None) or litellm_params.api_base,
singulr_api_key=getattr(litellm_params, "singulr_api_key", None) or litellm_params.api_key,
singulr_application_id=getattr(litellm_params, "singulr_application_id", None),
singulr_guardrail_id=getattr(litellm_params, "singulr_guardrail_id", None),
block_on_error=getattr(litellm_params, "block_on_error", None),
timeout=litellm_params.timeout,
guardrail_name=guardrail.get(
"guardrail_name",
"",
),
event_hook=litellm_params.mode,
default_on=litellm_params.default_on,
)
litellm.logging_callback_manager.add_litellm_callback(
_cb,
)
return _cb
guardrail_initializer_registry = {
SupportedGuardrailIntegrations.SINGULR.value: initialize_guardrail,
}
guardrail_class_registry = {
SupportedGuardrailIntegrations.SINGULR.value: SingulrGuardrail,
}

216
litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py Normal file
View file

@ -0,0 +1,216 @@
import os
from typing import Any
from urllib.parse import urlparse
import httpx
import pydantic
from litellm._logging import verbose_proxy_logger
from litellm.exceptions import GuardrailRaisedException
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.litellm_core_utils.litellm_logging import (
Logging as LiteLLMLoggingObj,
)
from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.proxy.guardrails.guardrail_hooks.base import (
GuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailPayload,
SingulrGuardrailRequest,
SingulrGuardrailResponse,
)
from litellm.types.utils import GenericGuardrailAPIInputs
_DEFAULT_API_BASE = "http://localhost:8003"
_GUARD_ENDPOINT = "/api/v1/ai-gateway/litellm"
_DEFAULT_TIMEOUT = 30.0
class SingulrGuardrail(CustomGuardrail):
def __init__(
self,
singulr_api_key: str | None = None,
singulr_api_base: str | None = None,
singulr_application_id: str | None = None,
singulr_guardrail_id: str | None = None,
block_on_error: bool | None = None,
timeout: float | None = None,
**kwargs: Any,
) -> None:
self.singulr_api_key = singulr_api_key or os.environ.get("SINGULR_API_KEY")
self.singulr_api_base = (singulr_api_base or os.environ.get("SINGULR_API_BASE") or _DEFAULT_API_BASE).rstrip(
"/"
)
parsed = urlparse(self.singulr_api_base)
if parsed.scheme == "http" and parsed.hostname not in (
"localhost",
"127.0.0.1",
):
raise ValueError(
f"Singulr: api_base {self.singulr_api_base} uses plain HTTP for a "
"non-local endpoint. Guardrail payloads contain the API token, full "
"conversation content, and the guardrail decision, so this endpoint "
"must use HTTPS."
)
self.singulr_application_id = singulr_application_id or os.environ.get("SINGULR_ENFORCEMENT_ENTITY_ID")
self.singulr_guardrail_id = singulr_guardrail_id or os.environ.get("SINGULR_GUARDRAIL_ID")
if block_on_error is None:
env = os.environ.get("SINGULR_BLOCK_ON_ERROR", "true")
self.block_on_error = env.lower() in ("true", "1", "yes")
else:
self.block_on_error = block_on_error
self.timeout = _DEFAULT_TIMEOUT if timeout is None else timeout
self.async_handler = get_async_httpx_client(
llm_provider=httpxSpecialProvider.GuardrailCallback,
)
if "supported_event_hooks" not in kwargs:
kwargs["supported_event_hooks"] = [
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
]
super().__init__(**kwargs)
@staticmethod
def get_config_model() -> type["GuardrailConfigModel"] | None:
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailConfigModel,
)
return SingulrGuardrailConfigModel
def _build_payload(
self,
request_data: dict[str, Any],
inputs: GenericGuardrailAPIInputs,
input_type: str,
) -> dict[str, Any]:
if not request_data:
texts = inputs.get("texts", [])
payload = SingulrGuardrailPayload(
input_type=input_type,
is_playground_request=True,
playground_text=texts[0] if texts else None,
)
else:
response = request_data.get("response")
singulr_req_object = SingulrGuardrailRequest(
model=request_data.get("model"),
messages=request_data.get("messages"),
tools=request_data.get("tools"),
model_response=response.model_dump(mode="json") if input_type == "response" and response else None,
litellm_metadata=request_data.get("litellm_metadata"),
)
payload = SingulrGuardrailPayload(
litellm_call_id=request_data.get("litellm_call_id"),
request_data=singulr_req_object,
input_type=input_type,
)
return payload.model_dump(mode="json")
def _build_headers(self) -> dict[str, str]:
return dict(
(header, value)
for header, value in (
("Content-Type", "application/json"),
("X-Singulr-Gateway-Token", self.singulr_api_key),
(
"X-Singulr-Enforcement-Entity-Id",
self.singulr_application_id or "",
),
("X-Singulr-Guardrail-Id", self.singulr_guardrail_id or ""),
)
if value
)
async def _call_api(self, payload: dict[str, Any]) -> SingulrGuardrailResponse | None:
endpoint = f"{self.singulr_api_base}{_GUARD_ENDPOINT}"
verbose_proxy_logger.debug("Singulr: %s", endpoint)
try:
response = await self.async_handler.post(
url=endpoint,
headers=self._build_headers(),
json=payload,
timeout=self.timeout,
)
response.raise_for_status()
result = SingulrGuardrailResponse.model_validate(response.json())
verbose_proxy_logger.debug("Singulr: result=%s", result)
return result
except httpx.HTTPStatusError as exc:
verbose_proxy_logger.error(
"Singulr API returned HTTP %s: %s",
exc.response.status_code,
str(exc),
)
if self.block_on_error:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=(f"Singulr API returned HTTP {exc.response.status_code}: {exc.response.text}"),
) from exc
return None
except httpx.TransportError as exc:
verbose_proxy_logger.error("Singulr API unreachable: %s", str(exc))
if self.block_on_error:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"Singulr API unreachable (block_on_error=True): {exc}",
) from exc
return None
except (ValueError, pydantic.ValidationError) as exc:
verbose_proxy_logger.error("Singulr API returned an invalid response: %s", str(exc))
if self.block_on_error:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"Singulr API returned an invalid response: {exc}",
) from exc
return None
@log_guardrail_information
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict,
input_type: str,
logging_obj: "LiteLLMLoggingObj | None" = None,
) -> GenericGuardrailAPIInputs:
payload = self._build_payload(request_data, inputs, input_type)
if not payload:
return inputs
result = await self._call_api(payload)
if result is None:
return inputs
verbose_proxy_logger.debug(
"Singulr: should_block=%s blocking_due_to=%s",
result.should_block,
result.blocking_due_to,
)
if result.should_block:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"Blocked by Singulr: {result.blocking_due_to or 'unknown'}",
)
return inputs

16
litellm/proxy/proxy_cli.py
View file

@ -802,6 +802,19 @@ class ProxyInitializationHelpers:
),
envvar="MAX_REQUESTS_BEFORE_RESTART_JITTER",
)
@click.option(
"--limit_concurrency",
default=None,
type=click.IntRange(min=1),
help=(
"Set uvicorn's concurrency limit. Uvicorn counts both active tasks and "
"accepted connections and returns HTTP 503 after the limit is reached. "
"Idle connections can consume capacity, so use upstream connection/header "
"timeouts and per-client connection limits. Only applies to uvicorn "
"(ignored under --run_gunicorn / --run_hypercorn / --run_granian)."
),
envvar="LIMIT_CONCURRENCY",
)
@click.option(
"--enforce_prisma_migration_check",
is_flag=True,
@ -870,6 +883,7 @@ def run_server(
timeout_worker_healthcheck,
max_requests_before_restart,
max_requests_before_restart_jitter: Optional[int],
limit_concurrency: Optional[int],
enforce_prisma_migration_check: bool,
use_v2_migration_resolver: bool,
reload: bool,
@ -1243,6 +1257,8 @@ def run_server(
if max_requests_before_restart is not None:
uvicorn_args["limit_max_requests"] = max_requests_before_restart
if run_gunicorn is False and run_hypercorn is False and run_granian is False:
if limit_concurrency is not None:
uvicorn_args["limit_concurrency"] = limit_concurrency
if max_requests_before_restart_jitter is not None:
ProxyInitializationHelpers._apply_uvicorn_max_requests_jitter(
uvicorn_args=uvicorn_args,

5
litellm/types/guardrails.py
View file

@ -53,6 +53,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.vigil_guard import (
from litellm.types.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
CiscoAIDefenseGuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.headroom import (
HeadroomGuardrailConfigModel,
)
@ -125,6 +128,7 @@ class SupportedGuardrailIntegrations(Enum):
RUBRIK = "rubrik"
VIGIL_GUARD = "vigil_guard"
REPELLOAI = "repelloai"
SINGULR = "singulr"
HEADROOM = "headroom"
COMPRESR = "compresr"
@ -932,6 +936,7 @@ class LitellmParams(
HiddenlayerGuardrailConfigModel,
QostodianNexusConfigModel,
VigilGuardGuardrailConfigModel,
SingulrGuardrailConfigModel,
):
guardrail: str = Field(description="The type of guardrail integration to use")
mode: Union[str, List[str], Mode] = Field(

63
litellm/types/proxy/guardrails/guardrail_hooks/singulr.py Normal file
View file

@ -0,0 +1,63 @@
from typing import Any, Optional
from pydantic import BaseModel, Field
from .base import GuardrailConfigModel
class SingulrGuardrailRequest(BaseModel):
model: Optional[str] = None
messages: Optional[list[dict[str, Any]]] = None
tools: Optional[list[dict[str, Any]]] = None
model_response: Optional[dict[str, Any]] = None
litellm_metadata: Optional[dict[str, Any]] = None
class SingulrGuardrailPayload(BaseModel):
litellm_call_id: Optional[str] = None
request_data: Optional[SingulrGuardrailRequest] = None
input_type: str
is_playground_request: Optional[bool] = None
playground_text: Optional[str] = None
class SingulrGuardrailResponse(BaseModel):
"""Response returned by the Singulr guardrail API."""
should_block: bool = False
blocking_due_to: Optional[str] = None
class SingulrGuardrailConfigModel(GuardrailConfigModel):
singulr_api_key: Optional[str] = Field(
default=None,
description="The Singulr API key. Generate API key from Singulr Platform.",
)
singulr_api_base: Optional[str] = Field(
default=None,
description="The Singulr API base URL. Get base URL from Singulr Platform.",
)
singulr_application_id: Optional[str] = Field(
default=None,
description="The Singulr application ID. Get application ID from Singulr Platform.",
)
singulr_guardrail_id: Optional[str] = Field(
default=None,
description="The Singulr Guardrail ID. Get guardrail ID from Singulr Platform.",
)
block_on_error: Optional[bool] = Field(
default=None,
description=(
"Whether to block requests when the Singulr Guardrails API is unavailable "
"or returns an error. If enabled, requests fail closed. "
"If disabled, requests continue without guardrail enforcement (fail open)."
),
)
@staticmethod
def ui_friendly_name() -> str:
return "Singulr"

6
litellm/utils.py
View file

@ -3198,6 +3198,12 @@ def get_optional_params_embeddings(
non_default_params=non_default_params, optional_params={}, kwargs=kwargs
)
elif custom_llm_provider == "vertex_ai" or custom_llm_provider == "gemini":
# OpenAI SDKs (and litellm's own client) send encoding_format="float"
# by default; float lists are exactly what the vertex API returns, so
# the param is a no-op — don't reject the provider default. Other
# values (e.g. "base64") stay on the unsupported-param path below.
if non_default_params.get("encoding_format") == "float":
non_default_params.pop("encoding_format")
supported_params = get_supported_openai_params(
model=model,
custom_llm_provider="vertex_ai",

29
tests/test_litellm/integrations/test_opik_utils.py Normal file
View file

@ -0,0 +1,29 @@
"""Unit tests for the native Opik integration's UUIDv7 id generation."""
import uuid
from datetime import datetime, timezone
from unittest.mock import patch
from litellm.integrations.opik.utils import create_uuid7
def _timestamp_ms(uuid_str: str) -> int:
"""Return the unix-ms timestamp encoded in a UUIDv7's top 48 bits."""
return uuid.UUID(uuid_str).int >> 80
def test_create_uuid7_is_valid_version_7_uuid():
parsed = uuid.UUID(create_uuid7())
assert parsed.version == 7
assert parsed.variant == uuid.RFC_4122
def test_create_uuid7_encodes_timestamp_in_milliseconds():
fixed = datetime(2026, 6, 24, 10, 0, 0, tzinfo=timezone.utc)
with patch(
"litellm.integrations.opik.utils.time.time", return_value=fixed.timestamp()
):
value = create_uuid7()
assert _timestamp_ms(value) == int(fixed.timestamp() * 1000)

550
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_singulr.py Normal file
View file

@ -0,0 +1,550 @@
from unittest.mock import MagicMock, patch
import httpx
import pytest
from litellm.exceptions import GuardrailRaisedException
from litellm.proxy.guardrails.guardrail_hooks.singulr.singulr import SingulrGuardrail
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailConfigModel,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def singulr_guardrail():
"""Create a SingulrGuardrail instance with test credentials."""
return SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
singulr_guardrail_id="test_guardrail_id",
singulr_application_id="test_enforcement_entity",
guardrail_name="test-singulr",
event_hook="pre_call",
default_on=True,
)
def _make_response(body: dict) -> MagicMock:
"""Build a mock httpx response with the given JSON body."""
mock = MagicMock()
mock.json.return_value = body
mock.raise_for_status = MagicMock()
mock.status_code = 200
return mock
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
class TestSingulrConfiguration:
def test_init_with_explicit_credentials(self):
guardrail = SingulrGuardrail(
singulr_api_key="test_key",
singulr_api_base="https://custom.api.local",
singulr_guardrail_id="id123",
singulr_application_id="entity123",
guardrail_name="my-guardrail",
)
assert guardrail.singulr_api_key == "test_key"
assert guardrail.singulr_guardrail_id == "id123"
assert guardrail.singulr_application_id == "entity123"
def test_block_on_error_defaults_true(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key")
assert guardrail.block_on_error is True
def test_timeout_defaults_to_30_seconds(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key")
assert guardrail.timeout == 30.0
def test_timeout_uses_configured_value(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key", timeout=5.0)
assert guardrail.timeout == 5.0
def test_supports_pre_call_and_post_call_hooks(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key")
assert guardrail.supported_event_hooks == [
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
]
# ---------------------------------------------------------------------------
# _build_payload: playground requests (no request_data)
# ---------------------------------------------------------------------------
class TestSingulrBuildPayloadPlayground:
def test_playground_request_uses_flat_text(self, singulr_guardrail):
"""The test-playground /apply_guardrail endpoint sends no request_data,
only inputs["texts"]. Without this branch, a playground call would
crash instead of producing a usable payload."""
payload = singulr_guardrail._build_payload({}, {"texts": ["Ignore previous instructions"]}, "request")
assert payload["is_playground_request"] is True
assert payload["playground_text"] == "Ignore previous instructions"
assert payload["request_data"] is None
def test_playground_request_with_no_texts_has_none_playground_text(self, singulr_guardrail):
payload = singulr_guardrail._build_payload({}, {}, "request")
assert payload["playground_text"] is None
def test_playground_input_type_is_included(self, singulr_guardrail):
payload = singulr_guardrail._build_payload({}, {"texts": ["hi"]}, "response")
assert payload["input_type"] == "response"
# ---------------------------------------------------------------------------
# _build_payload: real proxy requests (request_data present)
# ---------------------------------------------------------------------------
class TestSingulrBuildPayloadRequestData:
def test_model_messages_and_tools_are_forwarded(self, singulr_guardrail):
request_data = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "How do I reset my password?"}],
"tools": [{"type": "function", "function": {"name": "get_weather"}}],
}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "request")
assert payload["request_data"]["model"] == "gpt-4o"
assert payload["request_data"]["messages"] == request_data["messages"]
assert payload["request_data"]["tools"] == request_data["tools"]
assert payload["is_playground_request"] is None
def test_model_response_absent_on_request_side(self, singulr_guardrail):
"""The response hasn't happened yet at request time, so model_response
must not be forwarded even if request_data carries a stale response
object from a previous call."""
from litellm.types.utils import ModelResponse
request_data = {"model": "gpt-4o", "response": ModelResponse()}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "request")
assert payload["request_data"]["model_response"] is None
def test_model_response_is_forwarded_and_json_serializable(self, singulr_guardrail):
"""Regression: request_data["response"] is a ModelResponse (pydantic)
object containing nested non-JSON-safe values (e.g. a `created`
unix timestamp is fine, but nested pydantic submodels are not plain
dicts). Without mode="json" on both the inner and outer dumps, this
payload cannot be sent via httpx's json= kwarg."""
import json as _json
from litellm.types.utils import Choices, Message, ModelResponse, Usage
response = ModelResponse(
choices=[Choices(message=Message(role="assistant", content="Go to settings."))],
usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
)
request_data = {"model": "gpt-4o", "response": response}
payload = singulr_guardrail._build_payload(request_data, {"texts": ["Go to settings."]}, "response")
# Must not raise - this is what httpx's json= kwarg effectively does.
serialized = _json.dumps(payload)
assert "Go to settings." in serialized
assert payload["request_data"]["model_response"]["choices"][0]["message"]["content"] == "Go to settings."
def test_model_requested_tool_calls_are_forwarded_in_model_response(self, singulr_guardrail):
"""Tool calls the model requests arrive inside response.choices[].message.tool_calls.
They must survive the dump so Singulr can inspect what tools the
model is trying to invoke."""
from litellm.types.utils import Choices, Message, ModelResponse
response = ModelResponse(
choices=[
Choices(
message=Message(
role="assistant",
content=None,
tool_calls=[
{
"id": "call_1",
"type": "function",
"function": {"name": "get_current_time", "arguments": "{}"},
}
],
)
)
],
)
request_data = {"model": "gpt-4o", "response": response}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "response")
tool_calls = payload["request_data"]["model_response"]["choices"][0]["message"]["tool_calls"]
assert tool_calls[0]["function"]["name"] == "get_current_time"
def test_litellm_metadata_is_forwarded(self, singulr_guardrail):
request_data = {"model": "gpt-4o", "litellm_metadata": {"user_api_key_hash": "abc123"}}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "request")
assert payload["request_data"]["litellm_metadata"] == {"user_api_key_hash": "abc123"}
def test_internal_logging_object_is_not_forwarded(self, singulr_guardrail):
"""Regression: request_data can carry internal proxy objects (e.g. the
Logging instance) that aren't JSON-serializable at all. _build_payload
must only pull known request/response fields out of request_data,
not dump it wholesale, or this crashes on every real proxy call."""
import json as _json
class _NotSerializable:
pass
request_data = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
"litellm_logging_obj": _NotSerializable(),
}
payload = singulr_guardrail._build_payload(request_data, {"texts": ["hi"]}, "request")
# Must not raise.
_json.dumps(payload)
assert "litellm_logging_obj" not in payload["request_data"]
# ---------------------------------------------------------------------------
# Allow / block decisions
# ---------------------------------------------------------------------------
class TestSingulrAllowAction:
@pytest.mark.asyncio
async def test_allow_returns_inputs_unchanged(self, singulr_guardrail):
resp = _make_response({"should_block": False})
inputs = {"texts": ["How do I reset my password?"]}
with patch.object(singulr_guardrail.async_handler, "post", return_value=resp):
result = await singulr_guardrail.apply_guardrail(
inputs=inputs,
request_data={"model": "gpt-4o"},
input_type="request",
)
assert result is inputs
class TestSingulrBlockAction:
@pytest.mark.asyncio
async def test_block_raises_guardrail_exception(self, singulr_guardrail):
"""Regression: a should_block=True response must stop the request
instead of silently letting it through."""
resp = _make_response(
{
"should_block": True,
"blocking_due_to": "PII Information detected",
}
)
with patch.object(singulr_guardrail.async_handler, "post", return_value=resp):
with pytest.raises(GuardrailRaisedException) as exc_info:
await singulr_guardrail.apply_guardrail(
inputs={"texts": ["My SSN is 123-45-6789"]},
request_data={"model": "gpt-4o"},
input_type="request",
)
assert "PII Information detected" in str(exc_info.value)
@pytest.mark.asyncio
async def test_block_without_reason_uses_unknown_placeholder(self, singulr_guardrail):
resp = _make_response({"should_block": True})
with patch.object(singulr_guardrail.async_handler, "post", return_value=resp):
with pytest.raises(GuardrailRaisedException, match="unknown"):
await singulr_guardrail.apply_guardrail(
inputs={"texts": ["hi"]},
request_data={},
input_type="request",
)
# ---------------------------------------------------------------------------
# HTTP call wiring (endpoint, timeout, headers)
# ---------------------------------------------------------------------------
class TestSingulrRequestWiring:
@pytest.mark.asyncio
async def test_sends_configured_timeout(self):
"""litellm_params.timeout must reach the httpx call so operators can
tighten or loosen the latency budget instead of being stuck with a
hardcoded 30s regardless of configuration."""
guardrail = SingulrGuardrail(
singulr_api_key="test_key",
singulr_api_base="https://api.test.singulr.ai",
timeout=5.0,
)
resp = _make_response({"should_block": False})
with patch.object(guardrail.async_handler, "post", return_value=resp) as mock_post:
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
assert mock_post.call_args.kwargs["timeout"] == 5.0
class TestSingulrBuildHeaders:
def test_content_type_always_present(self, singulr_guardrail):
assert singulr_guardrail._build_headers()["Content-Type"] == "application/json"
def test_all_optional_headers_included_when_set(self, singulr_guardrail):
headers = singulr_guardrail._build_headers()
assert headers["X-Singulr-Gateway-Token"] == "test_token_1234"
assert headers["X-Singulr-Enforcement-Entity-Id"] == "test_enforcement_entity"
assert headers["X-Singulr-Guardrail-Id"] == "test_guardrail_id"
def test_optional_headers_absent_when_unset(self):
guardrail = SingulrGuardrail(guardrail_name="bare")
headers = guardrail._build_headers()
assert "X-Singulr-Gateway-Token" not in headers
assert "X-Singulr-Enforcement-Entity-Id" not in headers
assert "X-Singulr-Guardrail-Id" not in headers
# ---------------------------------------------------------------------------
# Non-JSON / malformed response handling
# ---------------------------------------------------------------------------
class TestSingulrInvalidResponse:
@pytest.mark.asyncio
async def test_non_json_response_block_on_error_false_returns_inputs(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=False,
)
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
mock_resp.json.side_effect = ValueError("No JSON object could be decoded")
inputs = {"texts": ["test"]}
with patch.object(guardrail.async_handler, "post", return_value=mock_resp):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
input_type="request",
)
assert result is inputs
@pytest.mark.asyncio
async def test_non_json_response_block_on_error_true_raises(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
mock_resp.json.side_effect = ValueError("No JSON object could be decoded")
with patch.object(guardrail.async_handler, "post", return_value=mock_resp):
with pytest.raises(GuardrailRaisedException):
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
@pytest.mark.asyncio
async def test_response_missing_expected_fields_block_on_error_true_raises(self):
"""Regression: a response body that fails SingulrGuardrailResponse
validation (e.g. should_block is a string, not a bool) must raise
GuardrailRaisedException instead of letting pydantic.ValidationError
propagate unhandled."""
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
resp = _make_response({"should_block": "not-a-bool"})
with patch.object(guardrail.async_handler, "post", return_value=resp):
with pytest.raises(GuardrailRaisedException):
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
# ---------------------------------------------------------------------------
# Transport error handling
# ---------------------------------------------------------------------------
class TestSingulrTransportError:
@pytest.mark.asyncio
async def test_remote_protocol_error_block_on_error_false_returns_inputs(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=False,
)
inputs = {"texts": ["test"]}
with patch.object(
guardrail.async_handler,
"post",
side_effect=httpx.RemoteProtocolError("malformed HTTP response"),
):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
input_type="request",
)
assert result is inputs
@pytest.mark.asyncio
async def test_remote_protocol_error_block_on_error_true_raises(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
with patch.object(
guardrail.async_handler,
"post",
side_effect=httpx.RemoteProtocolError("malformed HTTP response"),
):
with pytest.raises(GuardrailRaisedException):
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
# ---------------------------------------------------------------------------
# HTTP status error handling
# ---------------------------------------------------------------------------
class TestSingulrHttpStatusError:
@pytest.mark.asyncio
async def test_http_error_message_names_status_code_not_unreachable(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
mock_response = MagicMock()
mock_response.status_code = 403
mock_response.text = "Forbidden"
exc = httpx.HTTPStatusError("403 Forbidden", request=MagicMock(), response=mock_response)
mock_response.raise_for_status.side_effect = exc
with patch.object(guardrail.async_handler, "post", return_value=mock_response):
with pytest.raises(GuardrailRaisedException) as exc_info:
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
msg = str(exc_info.value)
assert "403" in msg
assert "unreachable" not in msg.lower()
@pytest.mark.asyncio
async def test_http_error_block_on_error_false_returns_inputs(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=False,
)
mock_response = MagicMock()
mock_response.status_code = 500
mock_response.text = "Internal Server Error"
exc = httpx.HTTPStatusError("500", request=MagicMock(), response=mock_response)
mock_response.raise_for_status.side_effect = exc
inputs = {"texts": ["test"]}
with patch.object(guardrail.async_handler, "post", return_value=mock_response):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
input_type="request",
)
assert result is inputs
# ---------------------------------------------------------------------------
# Config model
# ---------------------------------------------------------------------------
class TestSingulrConfigModel:
def test_ui_friendly_name(self):
assert SingulrGuardrailConfigModel.ui_friendly_name() == "Singulr"
# ---------------------------------------------------------------------------
# Initializer and registry
# ---------------------------------------------------------------------------
class TestSingulrInitializer:
def test_guardrail_initializer_registry_has_entry(self):
from litellm.proxy.guardrails.guardrail_hooks.singulr import (
initialize_guardrail,
)
assert callable(initialize_guardrail)
def test_initialize_guardrail_reads_singulr_prefixed_fields(self):
"""Regression: the UI config form (and YAML config) populate the
singulr_-prefixed fields declared on SingulrGuardrailConfigModel, not
the generic api_base/api_key fields. initialize_guardrail must read
those, or a UI-configured singulr_api_base is silently ignored and
the guardrail falls back to the localhost default."""
from litellm.proxy.guardrails.guardrail_hooks.singulr import (
initialize_guardrail,
)
from litellm.types.guardrails import Guardrail, LitellmParams
litellm_params = LitellmParams(
guardrail="singulr",
mode="pre_call",
singulr_api_base="https://configured.singulr.ai",
singulr_api_key="configured_key",
singulr_application_id="configured_app_id",
singulr_guardrail_id="configured_guardrail_id",
)
guardrail: Guardrail = {
"guardrail_name": "test-singulr",
"litellm_params": litellm_params,
}
cb = initialize_guardrail(litellm_params, guardrail)
assert cb.singulr_application_id == "configured_app_id"
assert cb.singulr_guardrail_id == "configured_guardrail_id"
def test_initialize_guardrail_wires_timeout(self):
"""BaseLitellmParams.timeout exists so operators can override the
per-request latency budget. initialize_guardrail must forward it to
SingulrGuardrail instead of leaving every deployment stuck on the
hardcoded default regardless of configuration."""
from litellm.proxy.guardrails.guardrail_hooks.singulr import (
initialize_guardrail,
)
from litellm.types.guardrails import Guardrail, LitellmParams
litellm_params = LitellmParams(
guardrail="singulr",
mode="pre_call",
singulr_api_key="configured_key",
timeout=12.5,
)
guardrail: Guardrail = {
"guardrail_name": "test-singulr",
"litellm_params": litellm_params,
}
cb = initialize_guardrail(litellm_params, guardrail)
assert cb.timeout == 12.5

73
tests/test_litellm/proxy/test_proxy_cli.py
View file

@ -582,6 +582,79 @@ class TestProxyInitializationHelpers:
), f"exit_code={result.exit_code}, output={result.output}"
mock_uvicorn_run.assert_called_once()
@patch("uvicorn.run")
@patch("atexit.register")
@patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database")
@patch(
"litellm.proxy.db.prisma_client.should_update_prisma_schema", return_value=False
)
def test_limit_concurrency_passed_to_uvicorn(
self, mock_should_update, mock_setup_db, mock_atexit_register, mock_uvicorn_run
):
"""--limit_concurrency must reach uvicorn.run so uvicorn sheds load with 503
past the cap; omitted values stay absent and non-positive values are rejected."""
from click.testing import CliRunner
from litellm.proxy.proxy_cli import run_server
runner = CliRunner()
mock_proxy_module = MagicMock(
app=MagicMock(),
ProxyConfig=MagicMock(),
KeyManagementSettings=MagicMock(),
save_worker_config=MagicMock(),
)
clean_env = {
k: v
for k, v in os.environ.items()
if k not in ("DATABASE_URL", "DIRECT_URL")
}
with (
patch.dict(os.environ, clean_env, clear=True),
patch.dict(
"sys.modules",
{
"proxy_server": mock_proxy_module,
"litellm.proxy.proxy_server": mock_proxy_module,
},
),
patch(
"litellm.proxy.proxy_cli.ProxyInitializationHelpers._get_default_unvicorn_init_args"
) as mock_get_args,
):
mock_get_args.side_effect = lambda *a, **k: {
"app": "litellm.proxy.proxy_server:app",
"host": "localhost",
"port": 8000,
}
result = runner.invoke(
run_server, ["--local", "--limit_concurrency", "250"]
)
assert (
result.exit_code == 0
), f"exit_code={result.exit_code}, output={result.output}"
mock_uvicorn_run.assert_called_once()
assert mock_uvicorn_run.call_args.kwargs.get("limit_concurrency") == 250
mock_uvicorn_run.reset_mock()
result = runner.invoke(run_server, ["--local"])
assert (
result.exit_code == 0
), f"exit_code={result.exit_code}, output={result.output}"
mock_uvicorn_run.assert_called_once()
assert "limit_concurrency" not in mock_uvicorn_run.call_args.kwargs
for invalid_value in ("0", "-1"):
mock_uvicorn_run.reset_mock()
result = runner.invoke(
run_server,
["--local", "--limit_concurrency", invalid_value],
)
assert result.exit_code == 2
assert "Invalid value for '--limit_concurrency'" in result.output
mock_uvicorn_run.assert_not_called()
@pytest.mark.parametrize(
"timeout_config,expected_timeout",
[

49
tests/test_litellm/test_utils.py
View file

@ -4717,6 +4717,55 @@ class TestValidateEnvironmentTencent:
assert "TENCENT_API_KEY" in result["missing_keys"]
class TestVertexEmbeddingEncodingFormat:
"""vertex_ai/gemini embeddings must accept encoding_format="float" — it's
the OpenAI SDK default and float lists are exactly what the vertex API
returns. Other values keep the unsupported-param behavior (drop with
drop_params, raise otherwise). Issue #33173."""
def test_encoding_format_float_is_accepted_and_dropped(self):
optional_params = litellm.utils.get_optional_params_embeddings(
model="gemini-embedding-001",
encoding_format="float",
custom_llm_provider="vertex_ai",
)
assert "encoding_format" not in optional_params
def test_encoding_format_float_accepted_for_gemini_provider(self):
optional_params = litellm.utils.get_optional_params_embeddings(
model="gemini-embedding-001",
encoding_format="float",
custom_llm_provider="gemini",
)
assert "encoding_format" not in optional_params
def test_encoding_format_base64_still_rejected_without_drop_params(self):
with pytest.raises(Exception) as excinfo:
litellm.utils.get_optional_params_embeddings(
model="gemini-embedding-001",
encoding_format="base64",
custom_llm_provider="vertex_ai",
)
assert "encoding_format" in str(excinfo.value)
def test_encoding_format_base64_dropped_with_drop_params(self):
optional_params = litellm.utils.get_optional_params_embeddings(
model="gemini-embedding-001",
encoding_format="base64",
custom_llm_provider="vertex_ai",
drop_params=True,
)
assert "encoding_format" not in optional_params
def test_dimensions_still_mapped(self):
optional_params = litellm.utils.get_optional_params_embeddings(
model="gemini-embedding-001",
encoding_format="float",
dimensions=256,
custom_llm_provider="vertex_ai",
)
assert optional_params.get("outputDimensionality") == 256
@pytest.mark.parametrize(
"model",