scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-13 03:59:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

Merge branch 'main' of https://github.com/BerriAI/litellm into pr45663-botfix-3e3e

Browse source 
# Conflicts:
#	litellm/decisions/main.py
This commit is contained in:
yucheng-berri 2026-10-10 15:31:43 -07:00
commit be36290a7d
42 changed files with 2233 additions and 477 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.github/workflows/test-unit.yml vendored
View file

@ -140,6 +140,7 @@ jobs:
tests/unit/batch_completion
tests/unit/decisions
tests/unit/litellm_core_utils
tests/unit/systemone
python-version: "3.12"
workers: 4
reruns: 0

1
litellm/__init__.py
View file

@ -1490,6 +1490,7 @@ from .rust_bridge import rust
from .rag.main import *
from .sandbox.main import *
from .decisions.main import *
from .systemone.main import *
from .tool_loop import ToolLoopMaxRoundsExceeded, arun_tool_loop, run_tool_loop
from .search.main import *
from .realtime_api.main import (

2
litellm/cost_calculator.py
View file

@ -2026,6 +2026,8 @@ def response_cost_calculator(
"asearch",
"decisions",
"adecisions",
"systemone",
"asystemone",
],
optional_params: dict,
cache_hit: bool | None = None,

347
litellm/decisions/call.py Normal file
View file

@ -0,0 +1,347 @@
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field, replace
from typing import Final, TypeAlias
import httpx
from pydantic import ConfigDict, TypeAdapter, ValidationError
from typing_extensions import assert_never
import litellm
from litellm.litellm_core_utils.core_helpers import RESPONSE_COST_HEADER, normalize_drop_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.decisions.transformation import (
BaseDecisionsConfig,
ir_to_systemone_response,
systemone_request_to_ir,
)
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.llms.openai.decisions.transformation import ir_to_openai_response, openai_request_to_ir
from litellm.types.decisions import (
DecisionQuestion,
DecisionsIRRequest,
DecisionsIRResponse,
DecisionsJSON,
DecisionsRequestBody,
DecisionsResponse,
OpenAIDecisionInput,
OpenAIDecisionQuestion,
OpenAIDecisionRequestBody,
OpenAIDecisionResponse,
UnsupportedDecisionsRequest,
)
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
SystemOneQuestions: TypeAlias = Mapping[str, DecisionQuestion | Mapping[str, object]]
OpenAIDecisionQuestions: TypeAlias = Sequence[OpenAIDecisionQuestion | Mapping[str, object]]
DecisionsRequestFormat: TypeAlias = DecisionsRequestBody | OpenAIDecisionRequestBody
_SYSTEMONE_REQUEST_ADAPTER: Final[TypeAdapter[DecisionsRequestBody]] = TypeAdapter(DecisionsRequestBody)
_OPENAI_REQUEST_ADAPTER: Final[TypeAdapter[OpenAIDecisionRequestBody]] = TypeAdapter(OpenAIDecisionRequestBody)
_SAFETY_IDENTIFIER_ADAPTER: Final[TypeAdapter[str | None]] = TypeAdapter(
str | None, config=ConfigDict(title="safety_identifier")
)
_HANDLER: Final = BaseLLMHTTPHandler()
@dataclass(frozen=True, slots=True)
class SystemOneRequestFields:
state: DecisionsJSON | None
questions: SystemOneQuestions | None
def validate(self, safety_identifier: str | None) -> DecisionsRequestBody:
return _SYSTEMONE_REQUEST_ADAPTER.validate_python({"state": self.state, "questions": self.questions})
@dataclass(frozen=True, slots=True)
class OpenAIDecisionRequestFields:
input: OpenAIDecisionInput | None
questions: OpenAIDecisionQuestions | None
def validate(self, safety_identifier: str | None) -> OpenAIDecisionRequestBody:
return _OPENAI_REQUEST_ADAPTER.validate_python(
{"input": self.input, "questions": self.questions, "safety_identifier": safety_identifier}
)
RequestFields: TypeAlias = SystemOneRequestFields | OpenAIDecisionRequestFields
@dataclass(frozen=True, slots=True, repr=False)
class DecisionsCall:
model: str
requested_model: str
custom_llm_provider: str
provider_config: BaseDecisionsConfig
request: DecisionsRequestFormat
ir_request: DecisionsIRRequest
body: Mapping[str, object] = field(repr=False)
api_base: str
api_key: str | None = field(repr=False)
logging_obj: LiteLLMLoggingObj | None
headers: Mapping[str, str]
timeout: float | httpx.Timeout | None
litellm_params: Mapping[str, object] = field(repr=False)
def supported_decisions_providers() -> tuple[str, ...]:
return tuple(
provider.value
for provider in LlmProviders
if ProviderConfigManager.get_provider_decisions_config(model="", provider=provider) is not None
)
def _provider_config(model: str, custom_llm_provider: str) -> BaseDecisionsConfig:
provider: Final = next((member for member in LlmProviders if member.value == custom_llm_provider), None)
provider_config: Final = (
None if provider is None else ProviderConfigManager.get_provider_decisions_config(model, provider)
)
if provider_config is None:
supported: Final = ", ".join(supported_decisions_providers())
raise litellm.BadRequestError(
message=f"Unknown Decisions provider '{custom_llm_provider}'. Supported providers: {supported}",
model=model,
llm_provider=custom_llm_provider,
)
return provider_config
def _ir_request(request: DecisionsRequestFormat, safety_identifier: str | None) -> DecisionsIRRequest:
match request:
case DecisionsRequestBody():
return replace(systemone_request_to_ir(request), safety_identifier=safety_identifier)
case OpenAIDecisionRequestBody():
return openai_request_to_ir(request)
case _:
assert_never(request)
def _drops_params(kwargs: Mapping[str, object]) -> bool:
return litellm.drop_params is True or normalize_drop_params(kwargs.get("drop_params")) is True
def _request_safety_identifier(safety_identifier: object, kwargs: Mapping[str, object]) -> str | None:
if isinstance(safety_identifier, str) or not _drops_params(kwargs):
return _SAFETY_IDENTIFIER_ADAPTER.validate_python(safety_identifier)
return None
@dataclass(frozen=True, slots=True)
class _UnsupportedSafetyIdentifier:
pass
def _provider_ir_request(
request: DecisionsRequestFormat,
*,
safety_identifier: str | None,
provider_config: BaseDecisionsConfig,
kwargs: Mapping[str, object],
) -> DecisionsIRRequest | _UnsupportedSafetyIdentifier:
ir_request: Final = _ir_request(request, safety_identifier)
if ir_request.safety_identifier is None or provider_config.supports_safety_identifier:
return ir_request
if _drops_params(kwargs):
return replace(ir_request, safety_identifier=None)
return _UnsupportedSafetyIdentifier()
def prepare_call(
*,
model: str,
fields: RequestFields,
safety_identifier: str | None,
api_key: str | None,
api_base: str | None,
timeout: float | httpx.Timeout | None,
custom_llm_provider: str | None,
extra_headers: Mapping[str, str] | None,
kwargs: Mapping[str, object],
) -> DecisionsCall:
upstream_model, provider, dynamic_api_key, dynamic_api_base = litellm.get_llm_provider(
model=model,
custom_llm_provider=custom_llm_provider,
api_base=api_base,
api_key=api_key,
)
provider_config: Final = _provider_config(upstream_model, provider)
if not upstream_model:
raise litellm.BadRequestError(
message="A model name is required for the Decisions API",
model=model,
llm_provider=provider,
)
try:
canonical_model: Final = provider_config.canonical_model(upstream_model)
except ValueError as error:
raise litellm.BadRequestError(message=str(error), model=model, llm_provider=provider) from error
try:
request_safety_identifier: Final = _request_safety_identifier(safety_identifier, kwargs)
request: Final = fields.validate(request_safety_identifier)
except ValidationError as error:
raise litellm.BadRequestError(
message=f"Invalid Decisions request: {error}",
model=model,
llm_provider=provider,
) from error
resolved_api_base: Final = provider_config.resolve_api_base(dynamic_api_base or api_base)
if resolved_api_base is None:
raise litellm.BadRequestError(
message=provider_config.missing_api_base_message(provider),
model=model,
llm_provider=provider,
)
resolved_api_key: Final = provider_config.resolve_api_key(dynamic_api_key or api_key)
if resolved_api_key is None and provider_config.api_key_required:
raise litellm.AuthenticationError(
message=f"Missing API key for Decisions provider '{provider}'",
model=model,
llm_provider=provider,
)
ir_request: Final = _provider_ir_request(
request,
safety_identifier=request_safety_identifier,
provider_config=provider_config,
kwargs=kwargs,
)
if isinstance(ir_request, _UnsupportedSafetyIdentifier):
raise litellm.UnsupportedParamsError(
message=(
f"{provider} does not support parameters: ['safety_identifier'], for model={model}. "
"To drop these, set `litellm.drop_params=True` or for proxy:\n\n"
"`litellm_settings:\n drop_params: true`\n"
),
model=model,
llm_provider=provider,
)
body: Final = provider_config.transform_decisions_request(
model=canonical_model, request=ir_request, custom_llm_provider=provider
)
if isinstance(body, UnsupportedDecisionsRequest):
raise litellm.BadRequestError(
message=f"Decisions provider '{provider}' cannot serve this request: {body.reason}",
model=model,
llm_provider=provider,
)
logging_obj: Final = kwargs.get("litellm_logging_obj")
if isinstance(logging_obj, LiteLLMLoggingObj):
logging_obj.update_from_kwargs(
kwargs=dict(kwargs),
model=canonical_model,
litellm_params={
"litellm_call_id": kwargs.get("litellm_call_id"),
"api_base": provider_config.get_complete_url(resolved_api_base, canonical_model),
},
custom_llm_provider=provider,
)
return DecisionsCall(
model=canonical_model,
requested_model=model,
custom_llm_provider=provider,
provider_config=provider_config,
request=request,
ir_request=ir_request,
body=body,
api_base=resolved_api_base,
api_key=resolved_api_key,
logging_obj=logging_obj if isinstance(logging_obj, LiteLLMLoggingObj) else None,
headers=extra_headers or {},
timeout=timeout,
litellm_params=kwargs,
)
def _format_response(response: DecisionsIRResponse, call: DecisionsCall) -> DecisionsResponse | OpenAIDecisionResponse:
formatted: Final = _formatted_response(response, call)
provider_cost: Final = call.provider_config.provider_reported_cost(response)
formatted.set_hidden_params(
{
"model": f"{call.custom_llm_provider}/{call.model}",
"custom_llm_provider": call.custom_llm_provider,
"provider_response_model": f"{call.custom_llm_provider}/{call.model}",
**({"additional_headers": {RESPONSE_COST_HEADER: provider_cost}} if provider_cost is not None else {}),
}
)
return formatted
def _formatted_response(
response: DecisionsIRResponse, call: DecisionsCall
) -> DecisionsResponse | OpenAIDecisionResponse:
match call.request:
case DecisionsRequestBody():
return ir_to_systemone_response(response, call.ir_request)
case OpenAIDecisionRequestBody():
return ir_to_openai_response(response, call.ir_request, call.requested_model)
case _:
assert_never(call.request)
def _map_upstream_exception(error: Exception, call: DecisionsCall) -> Exception:
if isinstance(error, BaseLLMException) and error.status_code_is_synthesized:
provider_label: Final = f"{call.custom_llm_provider[0].upper()}{call.custom_llm_provider[1:]}Exception"
return litellm.APIConnectionError(
message=f"{provider_label} - {error.message}",
llm_provider=call.custom_llm_provider,
model=f"{call.custom_llm_provider}/{call.model}",
)
return litellm.exception_type(
model=f"{call.custom_llm_provider}/{call.model}",
custom_llm_provider=call.custom_llm_provider,
original_exception=error,
)
async def asend(call: DecisionsCall) -> DecisionsResponse | OpenAIDecisionResponse:
try:
response: Final = await _HANDLER.adecisions(
model=call.model,
custom_llm_provider=call.custom_llm_provider,
logging_obj=call.logging_obj,
provider_config=call.provider_config,
request=call.ir_request,
body=call.body,
api_base=call.api_base,
api_key=call.api_key,
headers=call.headers,
timeout=call.timeout,
litellm_params=call.litellm_params,
)
except Exception as error:
raise _map_upstream_exception(error, call) from error
return _format_response(response, call)
def send(call: DecisionsCall) -> DecisionsResponse | OpenAIDecisionResponse:
try:
response: Final = _HANDLER.decisions(
model=call.model,
custom_llm_provider=call.custom_llm_provider,
logging_obj=call.logging_obj,
provider_config=call.provider_config,
request=call.ir_request,
body=call.body,
api_base=call.api_base,
api_key=call.api_key,
headers=call.headers,
timeout=call.timeout,
litellm_params=call.litellm_params,
)
except Exception as error:
raise _map_upstream_exception(error, call) from error
return _format_response(response, call)
def reject_other_format(*, model: str, kwargs: Mapping[str, object], field_name: str, other_function: str) -> None:
if field_name not in kwargs:
return
raise litellm.BadRequestError(
message=f"Requests with '{field_name}' go through litellm.{other_function}",
model=model,
llm_provider="",
)

388
litellm/decisions/main.py
View file

@ -1,151 +1,56 @@
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field, replace
from typing import Final, TypeAlias
import warnings
from collections.abc import Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter, ValidationError
from typing_extensions import assert_never
from pydantic import TypeAdapter
import litellm
from litellm.litellm_core_utils.core_helpers import RESPONSE_COST_HEADER, normalize_drop_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.decisions.transformation import (
BaseDecisionsConfig,
ir_to_systemone_response,
systemone_request_to_ir,
from litellm.decisions.call import (
DecisionsCall,
OpenAIDecisionQuestions,
OpenAIDecisionRequestFields,
RequestFields,
SystemOneQuestions,
SystemOneRequestFields,
asend,
prepare_call,
send,
)
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.llms.openai.decisions.transformation import ir_to_openai_response, openai_request_to_ir
from litellm.types.decisions import (
DecisionQuestion,
DecisionsIRRequest,
DecisionsIRResponse,
DecisionsJSON,
DecisionsRequestBody,
DecisionsResponse,
OpenAIDecisionInput,
OpenAIDecisionQuestion,
OpenAIDecisionRequestBody,
OpenAIDecisionResponse,
UnsupportedDecisionsRequest,
from litellm.types.decisions import DecisionsJSON, DecisionsResponse, OpenAIDecisionInput, OpenAIDecisionResponse
from litellm.utils import client
_STATE_ADAPTER: Final[TypeAdapter[DecisionsJSON | None]] = TypeAdapter(DecisionsJSON | None)
_SYSTEMONE_QUESTIONS_ADAPTER: Final[TypeAdapter[SystemOneQuestions | None]] = TypeAdapter(SystemOneQuestions | None)
_OPENAI_QUESTIONS_ADAPTER: Final[TypeAdapter[OpenAIDecisionQuestions | None]] = TypeAdapter(
OpenAIDecisionQuestions | None
)
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager, client
DecisionsQuestions: TypeAlias = (
Mapping[str, DecisionQuestion | Mapping[str, object]] | Sequence[OpenAIDecisionQuestion | Mapping[str, object]]
)
DecisionsRequestFormat: TypeAlias = DecisionsRequestBody | OpenAIDecisionRequestBody
_SYSTEMONE_REQUEST_ADAPTER: Final[TypeAdapter[DecisionsRequestBody]] = TypeAdapter(DecisionsRequestBody)
_OPENAI_REQUEST_ADAPTER: Final[TypeAdapter[OpenAIDecisionRequestBody]] = TypeAdapter(OpenAIDecisionRequestBody)
_SAFETY_IDENTIFIER_ADAPTER: Final[TypeAdapter[str | None]] = TypeAdapter(
str | None, config=ConfigDict(title="safety_identifier")
)
_HANDLER: Final = BaseLLMHTTPHandler()
@dataclass(frozen=True, slots=True, repr=False)
class _DecisionsCall:
model: str
requested_model: str
custom_llm_provider: str
provider_config: BaseDecisionsConfig
request: DecisionsRequestFormat
ir_request: DecisionsIRRequest
body: Mapping[str, object] = field(repr=False)
api_base: str
api_key: str | None = field(repr=False)
logging_obj: LiteLLMLoggingObj | None
headers: Mapping[str, str]
timeout: float | httpx.Timeout | None
litellm_params: Mapping[str, object] = field(repr=False)
def supported_decisions_providers() -> tuple[str, ...]:
return tuple(
provider.value
for provider in LlmProviders
if ProviderConfigManager.get_provider_decisions_config(model="", provider=provider) is not None
)
def _provider_config(model: str, custom_llm_provider: str) -> BaseDecisionsConfig:
provider: Final = next((member for member in LlmProviders if member.value == custom_llm_provider), None)
provider_config: Final = (
None if provider is None else ProviderConfigManager.get_provider_decisions_config(model, provider)
)
if provider_config is None:
supported: Final = ", ".join(supported_decisions_providers())
raise litellm.BadRequestError(
message=f"Unknown Decisions provider '{custom_llm_provider}'. Supported providers: {supported}",
model=model,
llm_provider=custom_llm_provider,
)
return provider_config
def _validate_request(
*,
state: DecisionsJSON | None,
questions: DecisionsQuestions | None,
def _request_fields(
decision_input: OpenAIDecisionInput | None,
safety_identifier: str | None,
) -> DecisionsRequestFormat:
if decision_input is None:
return _SYSTEMONE_REQUEST_ADAPTER.validate_python({"state": state, "questions": questions})
return _OPENAI_REQUEST_ADAPTER.validate_python(
{"input": decision_input, "questions": questions, "safety_identifier": safety_identifier}
questions: OpenAIDecisionQuestions | SystemOneQuestions | None,
kwargs: Mapping[str, object],
) -> RequestFields:
if "state" not in kwargs:
return OpenAIDecisionRequestFields(
input=decision_input, questions=_OPENAI_QUESTIONS_ADAPTER.validate_python(questions)
)
warnings.warn(
"litellm.decisions(state=...) is deprecated, call litellm.systemone for the System One request format",
UserWarning,
stacklevel=5,
)
return SystemOneRequestFields(
state=_STATE_ADAPTER.validate_python(kwargs["state"]),
questions=_SYSTEMONE_QUESTIONS_ADAPTER.validate_python(questions),
)
def _ir_request(request: DecisionsRequestFormat, safety_identifier: str | None) -> DecisionsIRRequest:
match request:
case DecisionsRequestBody():
return replace(systemone_request_to_ir(request), safety_identifier=safety_identifier)
case OpenAIDecisionRequestBody():
return openai_request_to_ir(request)
case _:
assert_never(request)
def _drops_params(kwargs: Mapping[str, object]) -> bool:
return litellm.drop_params is True or normalize_drop_params(kwargs.get("drop_params")) is True
def _request_safety_identifier(safety_identifier: object, kwargs: Mapping[str, object]) -> str | None:
if isinstance(safety_identifier, str) or not _drops_params(kwargs):
return _SAFETY_IDENTIFIER_ADAPTER.validate_python(safety_identifier)
return None
@dataclass(frozen=True, slots=True)
class _UnsupportedSafetyIdentifier:
pass
def _provider_ir_request(
request: DecisionsRequestFormat,
*,
safety_identifier: str | None,
provider_config: BaseDecisionsConfig,
kwargs: Mapping[str, object],
) -> DecisionsIRRequest | _UnsupportedSafetyIdentifier:
ir_request: Final = _ir_request(request, safety_identifier)
if ir_request.safety_identifier is None or provider_config.supports_safety_identifier:
return ir_request
if _drops_params(kwargs):
return replace(ir_request, safety_identifier=None)
return _UnsupportedSafetyIdentifier()
def _prepare_call(
def _prepare_decisions_call(
*,
model: str,
state: DecisionsJSON | None,
questions: DecisionsQuestions | None,
decision_input: OpenAIDecisionInput | None,
questions: OpenAIDecisionQuestions | SystemOneQuestions | None,
safety_identifier: str | None,
api_key: str | None,
api_base: str | None,
@ -153,174 +58,37 @@ def _prepare_call(
custom_llm_provider: str | None,
extra_headers: Mapping[str, str] | None,
kwargs: Mapping[str, object],
) -> _DecisionsCall:
upstream_model, provider, dynamic_api_key, dynamic_api_base = litellm.get_llm_provider(
) -> DecisionsCall:
return prepare_call(
model=model,
custom_llm_provider=custom_llm_provider,
api_base=api_base,
fields=_request_fields(decision_input, questions, kwargs),
safety_identifier=safety_identifier,
api_key=api_key,
)
provider_config: Final = _provider_config(upstream_model, provider)
if not upstream_model:
raise litellm.BadRequestError(
message="A model name is required for the Decisions API",
model=model,
llm_provider=provider,
)
if state is not None and decision_input is not None:
raise litellm.BadRequestError(
message="Pass either state (System One format) or input (OpenAI format) to the Decisions API, not both",
model=model,
llm_provider=provider,
)
try:
canonical_model: Final = provider_config.canonical_model(upstream_model)
except ValueError as error:
raise litellm.BadRequestError(message=str(error), model=model, llm_provider=provider) from error
try:
request_safety_identifier: Final = _request_safety_identifier(safety_identifier, kwargs)
request: Final = _validate_request(
state=state,
questions=questions,
decision_input=decision_input,
safety_identifier=request_safety_identifier,
)
except ValidationError as error:
raise litellm.BadRequestError(
message=f"Invalid Decisions request: {error}",
model=model,
llm_provider=provider,
) from error
resolved_api_base: Final = provider_config.resolve_api_base(dynamic_api_base or api_base)
if resolved_api_base is None:
raise litellm.BadRequestError(
message=provider_config.missing_api_base_message(provider),
model=model,
llm_provider=provider,
)
resolved_api_key: Final = provider_config.resolve_api_key(dynamic_api_key or api_key)
if resolved_api_key is None and provider_config.api_key_required:
raise litellm.AuthenticationError(
message=f"Missing API key for Decisions provider '{provider}'",
model=model,
llm_provider=provider,
)
ir_request: Final = _provider_ir_request(
request,
safety_identifier=request_safety_identifier,
provider_config=provider_config,
kwargs=kwargs,
)
if isinstance(ir_request, _UnsupportedSafetyIdentifier):
raise litellm.UnsupportedParamsError(
message=(
f"{provider} does not support parameters: ['safety_identifier'], for model={model}. "
"To drop these, set `litellm.drop_params=True` or for proxy:\n\n"
"`litellm_settings:\n drop_params: true`\n"
),
model=model,
llm_provider=provider,
)
body: Final = provider_config.transform_decisions_request(
model=canonical_model, request=ir_request, custom_llm_provider=provider
)
if isinstance(body, UnsupportedDecisionsRequest):
raise litellm.BadRequestError(
message=f"Decisions provider '{provider}' cannot serve this request: {body.reason}",
model=model,
llm_provider=provider,
)
logging_obj: Final = kwargs.get("litellm_logging_obj")
if isinstance(logging_obj, LiteLLMLoggingObj):
logging_obj.update_from_kwargs(
kwargs=dict(kwargs),
model=canonical_model,
litellm_params={
"litellm_call_id": kwargs.get("litellm_call_id"),
"api_base": provider_config.get_complete_url(resolved_api_base, canonical_model),
},
custom_llm_provider=provider,
)
return _DecisionsCall(
model=canonical_model,
requested_model=model,
custom_llm_provider=provider,
provider_config=provider_config,
request=request,
ir_request=ir_request,
body=body,
api_base=resolved_api_base,
api_key=resolved_api_key,
logging_obj=logging_obj if isinstance(logging_obj, LiteLLMLoggingObj) else None,
headers=extra_headers or {},
api_base=api_base,
timeout=timeout,
litellm_params=kwargs,
)
def _format_response(response: DecisionsIRResponse, call: _DecisionsCall) -> DecisionsResponse | OpenAIDecisionResponse:
formatted: Final = _formatted_response(response, call)
provider_cost: Final = call.provider_config.provider_reported_cost(response)
formatted.set_hidden_params(
{
"model": f"{call.custom_llm_provider}/{call.model}",
"custom_llm_provider": call.custom_llm_provider,
"provider_response_model": f"{call.custom_llm_provider}/{call.model}",
**({"additional_headers": {RESPONSE_COST_HEADER: provider_cost}} if provider_cost is not None else {}),
}
)
return formatted
def _formatted_response(
response: DecisionsIRResponse, call: _DecisionsCall
) -> DecisionsResponse | OpenAIDecisionResponse:
match call.request:
case DecisionsRequestBody():
return ir_to_systemone_response(response, call.ir_request)
case OpenAIDecisionRequestBody():
return ir_to_openai_response(response, call.ir_request, call.requested_model)
case _:
assert_never(call.request)
def _map_upstream_exception(error: Exception, call: _DecisionsCall) -> Exception:
if isinstance(error, BaseLLMException) and error.status_code_is_synthesized:
provider_label: Final = f"{call.custom_llm_provider[0].upper()}{call.custom_llm_provider[1:]}Exception"
return litellm.APIConnectionError(
message=f"{provider_label} - {error.message}",
llm_provider=call.custom_llm_provider,
model=f"{call.custom_llm_provider}/{call.model}",
)
return litellm.exception_type(
model=f"{call.custom_llm_provider}/{call.model}",
custom_llm_provider=call.custom_llm_provider,
original_exception=error,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
kwargs={key: value for key, value in kwargs.items() if key != "state"},
)
@client
async def adecisions(
model: str,
state: DecisionsJSON | None = None,
questions: DecisionsQuestions | None = None,
input: OpenAIDecisionInput | None = None,
questions: OpenAIDecisionQuestions | SystemOneQuestions | None = None,
safety_identifier: str | None = None,
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
input: OpenAIDecisionInput | None = None,
safety_identifier: str | None = None,
**kwargs: object,
) -> DecisionsResponse | OpenAIDecisionResponse:
call: Final = _prepare_call(
) -> OpenAIDecisionResponse | DecisionsResponse:
call: Final = _prepare_decisions_call(
model=model,
state=state,
questions=questions,
decision_input=input,
questions=questions,
safety_identifier=safety_identifier,
api_key=api_key,
api_base=api_base,
@ -329,44 +97,26 @@ async def adecisions(
extra_headers=extra_headers,
kwargs=kwargs,
)
try:
response: Final = await _HANDLER.adecisions(
model=call.model,
custom_llm_provider=call.custom_llm_provider,
logging_obj=call.logging_obj,
provider_config=call.provider_config,
request=call.ir_request,
body=call.body,
api_base=call.api_base,
api_key=call.api_key,
headers=call.headers,
timeout=call.timeout,
litellm_params=call.litellm_params,
)
except Exception as error:
raise _map_upstream_exception(error, call) from error
return _format_response(response, call)
return await asend(call)
@client
def decisions(
model: str,
state: DecisionsJSON | None = None,
questions: DecisionsQuestions | None = None,
input: OpenAIDecisionInput | None = None,
questions: OpenAIDecisionQuestions | SystemOneQuestions | None = None,
safety_identifier: str | None = None,
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
input: OpenAIDecisionInput | None = None,
safety_identifier: str | None = None,
**kwargs: object,
) -> DecisionsResponse | OpenAIDecisionResponse:
call: Final = _prepare_call(
) -> OpenAIDecisionResponse | DecisionsResponse:
call: Final = _prepare_decisions_call(
model=model,
state=state,
questions=questions,
decision_input=input,
questions=questions,
safety_identifier=safety_identifier,
api_key=api_key,
api_base=api_base,
@ -375,23 +125,7 @@ def decisions(
extra_headers=extra_headers,
kwargs=kwargs,
)
try:
response: Final = _HANDLER.decisions(
model=call.model,
custom_llm_provider=call.custom_llm_provider,
logging_obj=call.logging_obj,
provider_config=call.provider_config,
request=call.ir_request,
body=call.body,
api_base=call.api_base,
api_key=call.api_key,
headers=call.headers,
timeout=call.timeout,
litellm_params=call.litellm_params,
)
except Exception as error:
raise _map_upstream_exception(error, call) from error
return _format_response(response, call)
return send(call)
__all__ = ["adecisions", "decisions"]

2
litellm/litellm_core_utils/health_check_helpers.py
View file

@ -311,7 +311,7 @@ class HealthCheckHelpers:
**_filter_model_params(model_params=model_params),
document=_ocr_health_check_document(model=model, custom_llm_provider=custom_llm_provider),
),
"evaluation": lambda: litellm.adecisions(
"evaluation": lambda: litellm.asystemone(
**DECISIONS_CALL_PARAMS.validate_python(
{
"state": prompt or "health check",

15
litellm/llms/vertex_ai/common_utils.py
View file

@ -1,6 +1,7 @@
import re
from collections.abc import Mapping
from copy import deepcopy
from datetime import datetime, timezone
from enum import Enum
from functools import lru_cache
from types import MappingProxyType
@ -1000,18 +1001,18 @@ def strip_field(schema, field_name: str):
strip_field(items, field_name)
_VERTEX_DATETIME_FRACTION: Final = re.compile(r"\.\d+(?=Z$)", re.IGNORECASE)
def convert_vertex_datetime_to_openai_datetime(vertex_datetime: str) -> int:
"""
Converts a Vertex AI datetime string to an OpenAI datetime integer
Converts a Vertex AI / GCS RFC 3339 UTC timestamp to Unix epoch seconds
vertex_datetime: str = "2024-12-04T21:53:12.120184Z"
returns: int = 1722729192
returns: int = 1733349192
"""
from datetime import datetime
# Parse the ISO format string to datetime object
dt: Final = datetime.strptime(vertex_datetime, "%Y-%m-%dT%H:%M:%S.%fZ")
# Convert to Unix timestamp (seconds since epoch)
whole_seconds: Final = _VERTEX_DATETIME_FRACTION.sub("", vertex_datetime)
dt: Final = datetime.strptime(whole_seconds, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
return int(dt.timestamp())

2
litellm/proxy/common_request_processing.py
View file

@ -184,6 +184,7 @@ ProxyRouteType: TypeAlias = Literal[
"aocr",
"asearch",
"adecisions",
"asystemone",
"avideo_generation",
"avideo_list",
"avideo_status",
@ -2074,6 +2075,7 @@ class ProxyBaseLLMRequestProcessing:
"aocr",
"asearch",
"adecisions",
"asystemone",
"avideo_generation",
"avideo_list",
"avideo_status",

7
litellm/proxy/decisions_endpoints/endpoints.py
View file

@ -1,6 +1,6 @@
from collections.abc import Mapping
from types import MappingProxyType
from typing import Annotated, Final
from typing import Annotated, Final, Literal
from fastapi import APIRouter, Depends, Request, Response
from fastapi.responses import ORJSONResponse # pyright: ignore[reportDeprecated] # required endpoint contract
@ -109,6 +109,7 @@ async def _process_decisions(
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth,
body_adapter: TypeAdapter[DecisionsRequestBody] | TypeAdapter[OpenAIDecisionRequestBody],
route_type: Literal["adecisions", "asystemone"],
) -> object:
from litellm.proxy.proxy_server import (
general_settings as proxy_general_settings,
@ -146,7 +147,7 @@ async def _process_decisions(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
route_type="adecisions",
route_type=route_type,
proxy_logging_obj=proxy_logging_obj,
llm_router=llm_router,
general_settings=general_settings,
@ -208,6 +209,7 @@ async def systemone(
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
body_adapter=_DECISIONS_REQUEST_BODY_ADAPTER,
route_type="asystemone",
)
@ -250,4 +252,5 @@ async def decisions(
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
body_adapter=_OPENAI_DECISION_REQUEST_BODY_ADAPTER,
route_type="adecisions",
)

2
litellm/proxy/guardrails/guardrail_endpoints.py
View file

@ -1496,7 +1496,7 @@ async def get_guardrail_ui_settings() -> GuardrailUIAddGuardrailSettings:
- PII entity categories for UI organization
- Content filter settings (patterns and categories)
"""
from litellm.decisions.main import supported_decisions_providers
from litellm.decisions.call import supported_decisions_providers
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.patterns import (
PATTERN_CATEGORIES,
get_available_content_categories,

2
litellm/proxy/route_llm_request.py
View file

@ -94,6 +94,7 @@ ROUTE_ENDPOINT_MAPPING: Final = {
"aocr": "/ocr",
"asearch": "/search",
"adecisions": "/decisions",
"asystemone": "/systemone",
"avideo_generation": "/videos",
"avideo_list": "/videos",
"avideo_status": "/videos/{video_id}",
@ -519,6 +520,7 @@ RouteType = Literal[
"aocr",
"asearch",
"adecisions",
"asystemone",
"avideo_generation",
"avideo_list",
"avideo_status",

9
litellm/router.py
View file

@ -2233,6 +2233,11 @@ class Router:
self.adecisions = self.factory_function(adecisions, call_type="adecisions")
self.decisions = self.factory_function(decisions, call_type="decisions")
from litellm.systemone import asystemone, systemone
self.asystemone = self.factory_function(asystemone, call_type="asystemone") # pyright: ignore[reportUnknownMemberType] # factory_function takes a bare Callable
self.systemone = self.factory_function(systemone, call_type="systemone") # pyright: ignore[reportUnknownMemberType] # factory_function takes a bare Callable
def _initialize_video_endpoints(self):
"""Initialize video endpoints."""
from litellm.videos import (
@ -6996,6 +7001,8 @@ class Router:
"search",
"adecisions",
"decisions",
"asystemone",
"systemone",
"aadapter_generate_content",
"avideo_generation",
"video_generation",
@ -7070,6 +7077,7 @@ class Router:
"ocr",
"search",
"decisions",
"systemone",
"video_generation",
"video_list",
"video_status",
@ -7238,6 +7246,7 @@ class Router:
"aocr",
"ocr",
"adecisions",
"asystemone",
"avideo_generation",
"avideo_list",
"avideo_status",

3
litellm/systemone/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from litellm.systemone.main import asystemone, systemone
__all__ = ["asystemone", "systemone"]

107
litellm/systemone/main.py Normal file
View file

@ -0,0 +1,107 @@
from collections.abc import Mapping
from typing import Final
import httpx
from litellm.decisions.call import (
DecisionsCall,
SystemOneQuestions,
SystemOneRequestFields,
asend,
prepare_call,
reject_other_format,
send,
)
from litellm.types.decisions import DecisionsJSON, DecisionsResponse
from litellm.utils import client # pyright: ignore[reportUnknownVariableType] # client is an untyped decorator
def _prepare_systemone_call(
*,
model: str,
state: DecisionsJSON | None,
questions: SystemOneQuestions | None,
safety_identifier: str | None,
api_key: str | None,
api_base: str | None,
timeout: float | httpx.Timeout | None,
custom_llm_provider: str | None,
extra_headers: Mapping[str, str] | None,
kwargs: Mapping[str, object],
) -> DecisionsCall:
reject_other_format(model=model, kwargs=kwargs, field_name="input", other_function="decisions")
return prepare_call(
model=model,
fields=SystemOneRequestFields(state=state, questions=questions),
safety_identifier=safety_identifier,
api_key=api_key,
api_base=api_base,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
kwargs=kwargs,
)
def _systemone_response(response: object) -> DecisionsResponse:
assert isinstance(response, DecisionsResponse)
return response
@client
async def asystemone(
model: str,
state: DecisionsJSON | None = None,
questions: SystemOneQuestions | None = None,
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
safety_identifier: str | None = None,
**kwargs: object, # kwargs-ok: litellm_params forwarded by @client
) -> DecisionsResponse:
call: Final = _prepare_systemone_call(
model=model,
state=state,
questions=questions,
safety_identifier=safety_identifier,
api_key=api_key,
api_base=api_base,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
kwargs=kwargs,
)
return _systemone_response(await asend(call))
@client
def systemone(
model: str,
state: DecisionsJSON | None = None,
questions: SystemOneQuestions | None = None,
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, str] | None = None,
safety_identifier: str | None = None,
**kwargs: object, # kwargs-ok: litellm_params forwarded by @client
) -> DecisionsResponse:
call: Final = _prepare_systemone_call(
model=model,
state=state,
questions=questions,
safety_identifier=safety_identifier,
api_key=api_key,
api_base=api_base,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
extra_headers=extra_headers,
kwargs=kwargs,
)
return _systemone_response(send(call))
__all__ = ["asystemone", "systemone"]

8
litellm/types/utils.py
View file

@ -489,6 +489,8 @@ class CallTypes(str, Enum):
asearch = "asearch"
decisions = "decisions"
adecisions = "adecisions"
systemone = "systemone"
asystemone = "asystemone"
arealtime = "_arealtime"
aresponses_websocket = "_aresponses_websocket"
create_batch = "create_batch"
@ -677,6 +679,8 @@ CallTypesLiteral = Literal[
"asearch",
"decisions",
"adecisions",
"systemone",
"asystemone",
"_arealtime",
"_aresponses_websocket",
"create_batch",
@ -788,8 +792,8 @@ API_ROUTE_TO_CALL_TYPES: Final[Mapping[str, Sequence[CallTypes]]] = {
"/v1/search": [CallTypes.asearch, CallTypes.search],
"/decisions": [CallTypes.adecisions, CallTypes.decisions],
"/v1/decisions": [CallTypes.adecisions, CallTypes.decisions],
"/systemone": [CallTypes.adecisions, CallTypes.decisions],
"/v1/systemone": [CallTypes.adecisions, CallTypes.decisions],
"/systemone": [CallTypes.asystemone, CallTypes.systemone],
"/v1/systemone": [CallTypes.asystemone, CallTypes.systemone],
# Batches
"/batches": [CallTypes.acreate_batch, CallTypes.create_batch],
"/v1/batches": [CallTypes.acreate_batch, CallTypes.create_batch],

8
litellm/utils.py
View file

@ -974,6 +974,12 @@ def _ocr_document_summary(document: object) -> str:
# Runs once per call to check if the user wants to send their data anywhere - PostHog/Sentry/Slack/etc.
_DECISION_CALL_TYPES: Final[frozenset[str]] = frozenset(
(CallTypes.decisions.value, CallTypes.adecisions.value, CallTypes.systemone.value, CallTypes.asystemone.value)
)
def function_setup(
original_function: str,
rules_obj: Rules,
@ -1220,7 +1226,7 @@ def function_setup(
if isinstance(search_query, list)
else search_query
)
elif call_type in (CallTypes.decisions.value, CallTypes.adecisions.value):
elif call_type in _DECISION_CALL_TYPES:
decisions_state: Final = args[1] if len(args) > 1 else kwargs.get("state") or kwargs.get("input") or ""
messages = decisions_state if isinstance(decisions_state, str) else json.dumps(decisions_state, default=str)
elif call_type in (CallTypes.image_edit.value, CallTypes.aimage_edit.value):

258
tests/integration/cost_calculation/cost_tracking_cases.json
View file

@ -642,20 +642,20 @@
"gpt-realtime-mini-2025-12-15": {
"litellm_provider": "openai",
"mode": "realtime",
"input_cost_per_token": 6.0e-07,
"input_cost_per_token": 6e-07,
"output_cost_per_token": 2.4e-06,
"input_cost_per_audio_token": 1.0e-05,
"cache_read_input_token_cost": 6.0e-08,
"cache_read_input_audio_token_cost": 3.0e-07,
"output_cost_per_audio_token": 2.0e-05
"input_cost_per_audio_token": 1e-05,
"cache_read_input_token_cost": 6e-08,
"cache_read_input_audio_token_cost": 3e-07,
"output_cost_per_audio_token": 2e-05
},
"gpt-realtime-2.1": {
"litellm_provider": "openai",
"mode": "realtime",
"input_cost_per_token": 4.0e-06,
"input_cost_per_token": 4e-06,
"input_cost_per_audio_token": 3.2e-05,
"cache_read_input_token_cost": 4.0e-07,
"cache_read_input_audio_token_cost": 4.0e-07,
"cache_read_input_token_cost": 4e-07,
"cache_read_input_audio_token_cost": 4e-07,
"output_cost_per_token": 2.4e-05,
"output_cost_per_audio_token": 6.4e-05
},
@ -769,6 +769,52 @@
"search_context_size_medium": 0.014
},
"web_search_billing_unit": "per_query"
},
"xai/grok-4.7": {
"litellm_provider": "xai",
"mode": "chat",
"input_cost_per_token": 2e-06,
"output_cost_per_token": 6e-06,
"cache_read_input_token_cost": 5e-07,
"max_tokens": 500000,
"max_input_tokens": 500000,
"max_output_tokens": 500000,
"supports_function_calling": true
},
"us.anthropic.claude-opus-5-5": {
"litellm_provider": "bedrock_converse",
"mode": "chat",
"input_cost_per_token": 4.4e-06,
"output_cost_per_token": 2.2e-05,
"cache_read_input_token_cost": 2.2e-07,
"cache_creation_input_token_cost": 5.5e-06,
"max_tokens": 128000,
"max_input_tokens": 1000000,
"max_output_tokens": 128000,
"supports_function_calling": true
},
"claude-opus-5-5": {
"litellm_provider": "anthropic",
"mode": "chat",
"input_cost_per_token": 4e-06,
"output_cost_per_token": 2e-05,
"cache_read_input_token_cost": 2e-07,
"cache_creation_input_token_cost": 5e-06,
"max_tokens": 128000,
"max_input_tokens": 1000000,
"max_output_tokens": 128000,
"supports_function_calling": true
},
"gpt-6.1-sol": {
"litellm_provider": "openai",
"mode": "chat",
"input_cost_per_token": 2e-06,
"output_cost_per_token": 1e-05,
"cache_read_input_token_cost": 1e-07,
"max_tokens": 128000,
"max_input_tokens": 922000,
"max_output_tokens": 128000,
"supports_function_calling": true
}
},
"cases": [
@ -29687,7 +29733,13 @@
}
}
},
"expected": {"spend": 0.0025118, "input_cost": 0.0020792, "output_cost": 0.0004326, "prompt_tokens": 1840, "completion_tokens": 412}
"expected": {
"spend": 0.0025118,
"input_cost": 0.0020792,
"output_cost": 0.0004326,
"prompt_tokens": 1840,
"completion_tokens": 412
}
},
{
"name": "deepseek-deepseek-v4-chat-prompt_cache_hit",
@ -32102,6 +32154,190 @@
"prompt_tokens": 100,
"completion_tokens": 100
}
},
{
"name": "xai-grok-4.7-input_text",
"covers": "quota_management.spend_tracking.cost_matrix.logs_cost",
"model": "xai/grok-4.7",
"request": {
"model": "$MODEL",
"messages": [
{
"role": "user",
"content": "Say hello."
}
],
"stream": false,
"allowed_openai_params": []
},
"response": {
"content_type": "application/json",
"body": {
"id": "chatcmpl-$REQUEST_ID",
"object": "chat.completion",
"model": "grok-4.7",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Hello."
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 1840,
"completion_tokens": 412,
"total_tokens": 2252
}
}
},
"expected": {
"spend": 0.006152,
"input_cost": 0.00368,
"output_cost": 0.002472,
"prompt_tokens": 1840,
"completion_tokens": 412
}
},
{
"name": "us.anthropic.claude-opus-5-5-input_text",
"covers": "quota_management.spend_tracking.cost_matrix.logs_cost",
"model": "us.anthropic.claude-opus-5-5",
"request": {
"model": "$MODEL",
"messages": [
{
"role": "user",
"content": "Say hello."
}
],
"stream": false,
"allowed_openai_params": []
},
"response": {
"content_type": "application/json",
"body": {
"output": {
"message": {
"role": "assistant",
"content": [
{
"text": "Hello."
}
]
}
},
"stopReason": "end_turn",
"usage": {
"inputTokens": 1840,
"outputTokens": 412,
"totalTokens": 2252
},
"metrics": {
"latencyMs": 42
}
}
},
"expected": {
"spend": 0.01716,
"input_cost": 0.008096,
"output_cost": 0.009064,
"prompt_tokens": 1840,
"completion_tokens": 412
}
},
{
"name": "claude-opus-5-5-input_text",
"covers": "quota_management.spend_tracking.cost_matrix.logs_cost",
"model": "claude-opus-5-5",
"request": {
"model": "$MODEL",
"messages": [
{
"role": "user",
"content": "Say hello."
}
],
"stream": false,
"allowed_openai_params": []
},
"response": {
"content_type": "application/json",
"body": {
"id": "msg_$REQUEST_ID",
"type": "message",
"role": "assistant",
"model": "claude-opus-5-5",
"content": [
{
"type": "text",
"text": "Hello."
}
],
"stop_reason": "end_turn",
"stop_sequence": null,
"usage": {
"input_tokens": 1840,
"output_tokens": 412
}
}
},
"expected": {
"spend": 0.0156,
"input_cost": 0.00736,
"output_cost": 0.00824,
"prompt_tokens": 1840,
"completion_tokens": 412
}
},
{
"name": "gpt-6.1-sol-input_text",
"covers": "quota_management.spend_tracking.cost_matrix.logs_cost",
"model": "gpt-6.1-sol",
"request": {
"model": "$MODEL",
"messages": [
{
"role": "user",
"content": "Say hello."
}
],
"stream": false,
"allowed_openai_params": []
},
"response": {
"content_type": "application/json",
"body": {
"id": "chatcmpl-$REQUEST_ID",
"object": "chat.completion",
"created": 1,
"model": "gpt-6.1-sol",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Hello."
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 1840,
"completion_tokens": 412,
"total_tokens": 2252
}
}
},
"expected": {
"spend": 0.0078,
"input_cost": 0.00368,
"output_cost": 0.00412,
"prompt_tokens": 1840,
"completion_tokens": 412
}
}
],
"batch_cases": [
@ -32149,8 +32385,8 @@
],
"expected": {
"spend": 0.000126,
"input_cost": 0.000056,
"output_cost": 0.00007,
"input_cost": 5.6e-05,
"output_cost": 7e-05,
"prompt_tokens": 100,
"completion_tokens": 10,
"cost_header": false

0
tests/integration/cost_calculation/stream_parity/__init__.py Normal file
View file

0
tests/integration/cost_calculation/stream_parity/bases/__init__.py Normal file
View file

123
tests/integration/cost_calculation/stream_parity/bases/anthropic.py Normal file
View file

@ -0,0 +1,123 @@
from typing import Final
from integration.cost_calculation.cost_tracking_case import (
CostTrackingTestCase,
Deployment,
ExactExpected,
JsonResponse,
SseResponse,
)
from integration.cost_calculation.stream_parity.case import COVERS, MODEL, REQUEST_ID, StreamParityTestCase, sse_frames
CLAUDE_OPUS_5_5_MESSAGES: Final = CostTrackingTestCase(
name="claude-opus-5-5-messages-parity",
covers=COVERS,
model="claude-opus-5-5",
endpoint="/v1/messages",
deployment=Deployment(input_cost_per_token=0.001, output_cost_per_token=0.002),
request={
"model": MODEL,
"max_tokens": 64,
"messages": [{"role": "user", "content": "Say hello."}],
"cache": {"no-cache": True},
},
response=JsonResponse(
content_type="application/json",
body={
"id": f"msg_{REQUEST_ID}",
"type": "message",
"role": "assistant",
"model": "claude-opus-5-5",
"content": [{"type": "text", "text": "Hello."}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 30, "output_tokens": 40},
},
),
expected=ExactExpected(spend=0.11, input_cost=0.03, output_cost=0.08, prompt_tokens=30, completion_tokens=40),
)
CLAUDE_OPUS_5_5_STREAM_RESPONSE: Final = SseResponse(
content_type="text/event-stream",
frames=sse_frames(
{
"type": "message_start",
"message": {
"id": f"msg_{REQUEST_ID}",
"type": "message",
"role": "assistant",
"model": "claude-opus-5-5",
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 30, "output_tokens": 1},
},
},
{"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
{"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello."}},
{"type": "content_block_stop", "index": 0},
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 40},
},
{"type": "message_stop"},
),
)
CLAUDE_OPUS_5_5_MESSAGES_PARITY: Final = StreamParityTestCase(
plain=CLAUDE_OPUS_5_5_MESSAGES,
streamed=CLAUDE_OPUS_5_5_MESSAGES.model_copy(
update={
"name": "claude-opus-5-5-messages-parity-stream",
"request": {**CLAUDE_OPUS_5_5_MESSAGES.request, "stream": True},
"response": CLAUDE_OPUS_5_5_STREAM_RESPONSE,
}
),
)
CLAUDE_OPUS_5_5_CHAT_COMPLETIONS: Final = CLAUDE_OPUS_5_5_MESSAGES.model_copy(
update={
"name": "claude-opus-5-5-chat-parity",
"endpoint": "/v1/chat/completions",
"request": {
"model": MODEL,
"messages": [{"role": "user", "content": "Say hello."}],
"cache": {"no-cache": True},
},
}
)
CLAUDE_OPUS_5_5_CHAT_COMPLETIONS_PARITY: Final = StreamParityTestCase(
plain=CLAUDE_OPUS_5_5_CHAT_COMPLETIONS,
streamed=CLAUDE_OPUS_5_5_CHAT_COMPLETIONS.model_copy(
update={
"name": "claude-opus-5-5-chat-parity-stream",
"request": {
**CLAUDE_OPUS_5_5_CHAT_COMPLETIONS.request,
"stream": True,
"stream_options": {"include_usage": True},
},
"response": CLAUDE_OPUS_5_5_STREAM_RESPONSE,
}
),
)
CLAUDE_OPUS_5_5_RESPONSES: Final = CLAUDE_OPUS_5_5_MESSAGES.model_copy(
update={
"name": "claude-opus-5-5-responses-parity",
"endpoint": "/v1/responses",
"request": {"model": MODEL, "input": "Say hello.", "cache": {"no-cache": True}},
}
)
CLAUDE_OPUS_5_5_RESPONSES_PARITY: Final = StreamParityTestCase(
plain=CLAUDE_OPUS_5_5_RESPONSES,
streamed=CLAUDE_OPUS_5_5_RESPONSES.model_copy(
update={
"name": "claude-opus-5-5-responses-parity-stream",
"request": {**CLAUDE_OPUS_5_5_RESPONSES.request, "stream": True},
"response": CLAUDE_OPUS_5_5_STREAM_RESPONSE,
}
),
)

62
tests/integration/cost_calculation/stream_parity/bases/bedrock.py Normal file
View file

@ -0,0 +1,62 @@
"""/v1/messages on a Bedrock Converse deployment: the messages -> converse bridge."""
from typing import Final
from integration.cost_calculation.cost_tracking_case import (
CostTrackingTestCase,
Deployment,
EventStreamEvent,
EventStreamResponse,
ExactExpected,
JsonResponse,
)
from integration.cost_calculation.stream_parity.case import COVERS, MODEL, StreamParityTestCase
_USAGE: Final = {"inputTokens": 30, "outputTokens": 40, "totalTokens": 70}
CLAUDE_OPUS_5_5_CONVERSE_MESSAGES: Final = CostTrackingTestCase(
name="bedrock-converse-claude-opus-5-5-messages-parity",
covers=COVERS,
model="us.anthropic.claude-opus-5-5",
endpoint="/v1/messages",
deployment=Deployment(input_cost_per_token=0.001, output_cost_per_token=0.002),
request={
"model": MODEL,
"max_tokens": 64,
"messages": [{"role": "user", "content": "Say hello."}],
"cache": {"no-cache": True},
},
response=JsonResponse(
content_type="application/json",
body={
"output": {"message": {"role": "assistant", "content": [{"text": "Hello."}]}},
"stopReason": "end_turn",
"usage": _USAGE,
"metrics": {"latencyMs": 1},
},
),
expected=ExactExpected(spend=0.11, input_cost=0.03, output_cost=0.08, prompt_tokens=30, completion_tokens=40),
)
CLAUDE_OPUS_5_5_CONVERSE_MESSAGES_PARITY: Final = StreamParityTestCase(
plain=CLAUDE_OPUS_5_5_CONVERSE_MESSAGES,
streamed=CLAUDE_OPUS_5_5_CONVERSE_MESSAGES.model_copy(
update={
"name": "bedrock-converse-claude-opus-5-5-messages-parity-stream",
"request": {**CLAUDE_OPUS_5_5_CONVERSE_MESSAGES.request, "stream": True},
"response": EventStreamResponse(
content_type="application/vnd.amazon.eventstream",
framing="converse",
events=(
EventStreamEvent(event_type="messageStart", payload={"role": "assistant"}),
EventStreamEvent(
event_type="contentBlockDelta", payload={"delta": {"text": "Hello."}, "contentBlockIndex": 0}
),
EventStreamEvent(event_type="contentBlockStop", payload={"contentBlockIndex": 0}),
EventStreamEvent(event_type="messageStop", payload={"stopReason": "end_turn"}),
EventStreamEvent(event_type="metadata", payload={"usage": _USAGE, "metrics": {"latencyMs": 1}}),
),
),
}
),
)

91
tests/integration/cost_calculation/stream_parity/bases/openai.py Normal file
View file

@ -0,0 +1,91 @@
from typing import Final
from integration.cost_calculation.cost_tracking_case import (
CostTrackingTestCase,
Deployment,
ExactExpected,
JsonResponse,
SseResponse,
)
from integration.cost_calculation.stream_parity.bases.openai_responses import (
GPT_6_1_SOL_RESPONSES,
GPT_6_1_SOL_RESPONSES_PARITY_STREAM_RESPONSE,
)
from integration.cost_calculation.stream_parity.case import COVERS, MODEL, REQUEST_ID, StreamParityTestCase, sse_frames
_USAGE: Final = {"prompt_tokens": 30, "completion_tokens": 40, "total_tokens": 70}
_CHUNK: Final = {
"id": f"chatcmpl-{REQUEST_ID}",
"object": "chat.completion.chunk",
"created": 1,
"model": "gpt-6.1-sol",
}
GPT_6_1_SOL_CHAT_COMPLETIONS: Final = CostTrackingTestCase(
name="gpt-6.1-sol-chat-parity",
covers=COVERS,
model="gpt-6.1-sol",
endpoint="/v1/chat/completions",
deployment=Deployment(input_cost_per_token=0.001, output_cost_per_token=0.002),
request={"model": MODEL, "messages": [{"role": "user", "content": "Say hello."}], "cache": {"no-cache": True}},
response=JsonResponse(
content_type="application/json",
body={
"id": f"chatcmpl-{REQUEST_ID}",
"object": "chat.completion",
"created": 1,
"model": "gpt-6.1-sol",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "Hello."}, "finish_reason": "stop"}],
"usage": _USAGE,
},
),
expected=ExactExpected(spend=0.11, input_cost=0.03, output_cost=0.08, prompt_tokens=30, completion_tokens=40),
)
GPT_6_1_SOL_CHAT_COMPLETIONS_PARITY: Final = StreamParityTestCase(
plain=GPT_6_1_SOL_CHAT_COMPLETIONS,
streamed=GPT_6_1_SOL_CHAT_COMPLETIONS.model_copy(
update={
"name": "gpt-6.1-sol-chat-parity-stream",
"request": {
**GPT_6_1_SOL_CHAT_COMPLETIONS.request,
"stream": True,
"stream_options": {"include_usage": True},
},
"response": SseResponse(
content_type="text/event-stream",
frames=sse_frames(
{**_CHUNK, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hello."}}]},
{**_CHUNK, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
{**_CHUNK, "choices": [], "usage": _USAGE},
done=True,
),
),
}
),
)
GPT_6_1_SOL_MESSAGES: Final = GPT_6_1_SOL_RESPONSES.model_copy(
update={
"name": "gpt-6.1-sol-messages-parity",
"model": "gpt-6.1-sol",
"endpoint": "/v1/messages",
"request": {
"model": MODEL,
"max_tokens": 64,
"messages": [{"role": "user", "content": "Say hello."}],
"cache": {"no-cache": True},
},
}
)
GPT_6_1_SOL_MESSAGES_PARITY: Final = StreamParityTestCase(
plain=GPT_6_1_SOL_MESSAGES,
streamed=GPT_6_1_SOL_MESSAGES.model_copy(
update={
"name": "gpt-6.1-sol-messages-parity-stream",
"request": {**GPT_6_1_SOL_MESSAGES.request, "stream": True},
"response": GPT_6_1_SOL_RESPONSES_PARITY_STREAM_RESPONSE,
}
),
)

115
tests/integration/cost_calculation/stream_parity/bases/openai_responses.py Normal file
View file

@ -0,0 +1,115 @@
from typing import Final
from integration.cost_calculation.cost_tracking_case import (
CostTrackingTestCase,
Deployment,
ExactExpected,
JsonResponse,
SseResponse,
)
from integration.cost_calculation.stream_parity.case import COVERS, MODEL, REQUEST_ID, StreamParityTestCase, sse_frames
_ITEM: Final = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": "Hello.", "annotations": []}],
}
_RESPONSE: Final = {
"id": f"resp_{REQUEST_ID}",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-6.1-sol",
"output": [_ITEM],
"usage": {"input_tokens": 30, "output_tokens": 40, "total_tokens": 70},
}
GPT_6_1_SOL_RESPONSES: Final = CostTrackingTestCase(
name="gpt-6.1-sol-responses-parity",
covers=COVERS,
model="gpt-6.1-sol",
endpoint="/v1/responses",
deployment=Deployment(input_cost_per_token=0.001, output_cost_per_token=0.002),
request={"model": MODEL, "input": "Say hello.", "cache": {"no-cache": True}},
response=JsonResponse(content_type="application/json", body=_RESPONSE),
expected=ExactExpected(spend=0.11, input_cost=0.03, output_cost=0.08, prompt_tokens=30, completion_tokens=40),
)
GPT_6_1_SOL_RESPONSES_PARITY_STREAM_RESPONSE: Final = SseResponse(
content_type="text/event-stream",
frames=sse_frames(
{
"type": "response.created",
"sequence_number": 0,
"response": {**_RESPONSE, "status": "in_progress", "output": []},
},
{
"type": "response.output_item.added",
"sequence_number": 1,
"output_index": 0,
"item": {**_ITEM, "status": "in_progress", "content": []},
},
{
"type": "response.output_text.delta",
"sequence_number": 2,
"item_id": "msg_1",
"output_index": 0,
"content_index": 0,
"delta": "Hello.",
},
{"type": "response.output_item.done", "sequence_number": 3, "output_index": 0, "item": _ITEM},
{"type": "response.completed", "sequence_number": 4, "response": _RESPONSE},
),
)
GPT_6_1_SOL_RESPONSES_PARITY: Final = StreamParityTestCase(
plain=GPT_6_1_SOL_RESPONSES,
streamed=GPT_6_1_SOL_RESPONSES.model_copy(
update={
"name": "gpt-6.1-sol-responses-parity-stream",
"request": {**GPT_6_1_SOL_RESPONSES.request, "stream": True},
"response": GPT_6_1_SOL_RESPONSES_PARITY_STREAM_RESPONSE,
}
),
)
_PRO_RESPONSE: Final = {**_RESPONSE, "model": "gpt-5.5-pro"}
GPT_5_5_PRO_STREAM_RESPONSE: Final = GPT_6_1_SOL_RESPONSES_PARITY_STREAM_RESPONSE.model_copy(
update={
"frames": tuple(
frame.replace('"model": "gpt-6.1-sol"', '"model": "gpt-5.5-pro"')
for frame in GPT_6_1_SOL_RESPONSES_PARITY_STREAM_RESPONSE.frames
)
}
)
GPT_5_5_PRO_CHAT_COMPLETIONS: Final = GPT_6_1_SOL_RESPONSES.model_copy(
update={
"name": "gpt-5.5-pro-chat-parity",
"model": "gpt-5.5-pro",
"response": JsonResponse(content_type="application/json", body=_PRO_RESPONSE),
"endpoint": "/v1/chat/completions",
"request": {
"model": MODEL,
"messages": [{"role": "user", "content": "Say hello."}],
"cache": {"no-cache": True},
},
}
)
GPT_5_5_PRO_CHAT_COMPLETIONS_PARITY: Final = StreamParityTestCase(
plain=GPT_5_5_PRO_CHAT_COMPLETIONS,
streamed=GPT_5_5_PRO_CHAT_COMPLETIONS.model_copy(
update={
"name": "gpt-5.5-pro-chat-parity-stream",
"request": {
**GPT_5_5_PRO_CHAT_COMPLETIONS.request,
"stream": True,
"stream_options": {"include_usage": True},
},
"response": GPT_5_5_PRO_STREAM_RESPONSE,
}
),
)

65
tests/integration/cost_calculation/stream_parity/bases/xai.py Normal file
View file

@ -0,0 +1,65 @@
"""/v1/messages on a chat-completions provider: the messages -> chat/completions bridge."""
from typing import Final
from integration.cost_calculation.cost_tracking_case import (
CostTrackingTestCase,
Deployment,
ExactExpected,
JsonResponse,
SseResponse,
)
from integration.cost_calculation.stream_parity.case import COVERS, MODEL, REQUEST_ID, StreamParityTestCase, sse_frames
_USAGE: Final = {"prompt_tokens": 30, "completion_tokens": 40, "total_tokens": 70}
_CHUNK: Final = {
"id": f"chatcmpl-{REQUEST_ID}",
"object": "chat.completion.chunk",
"created": 1,
"model": "grok-4.7",
}
GROK_4_7_MESSAGES: Final = CostTrackingTestCase(
name="grok-4.7-messages-parity",
covers=COVERS,
model="xai/grok-4.7",
endpoint="/v1/messages",
deployment=Deployment(input_cost_per_token=0.001, output_cost_per_token=0.002),
request={
"model": MODEL,
"max_tokens": 64,
"messages": [{"role": "user", "content": "Say hello."}],
"cache": {"no-cache": True},
},
response=JsonResponse(
content_type="application/json",
body={
"id": f"chatcmpl-{REQUEST_ID}",
"object": "chat.completion",
"created": 1,
"model": "grok-4.7",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "Hello."}, "finish_reason": "stop"}],
"usage": _USAGE,
},
),
expected=ExactExpected(spend=0.11, input_cost=0.03, output_cost=0.08, prompt_tokens=30, completion_tokens=40),
)
GROK_4_7_MESSAGES_PARITY: Final = StreamParityTestCase(
plain=GROK_4_7_MESSAGES,
streamed=GROK_4_7_MESSAGES.model_copy(
update={
"name": "grok-4.7-messages-parity-stream",
"request": {**GROK_4_7_MESSAGES.request, "stream": True},
"response": SseResponse(
content_type="text/event-stream",
frames=sse_frames(
{**_CHUNK, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hello."}}]},
{**_CHUNK, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
{**_CHUNK, "choices": [], "usage": _USAGE},
done=True,
),
),
}
),
)

44
tests/integration/cost_calculation/stream_parity/case.py Normal file
View file

@ -0,0 +1,44 @@
import json
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Final
from integration.cost_calculation.cost_tracking_case import CostTrackingTestCase
from pydantic import JsonValue
COVERS: Final = "quota_management.spend_tracking.scripted_wire.logs_cost"
REQUEST_ID: Final = "$REQUEST_ID"
MODEL: Final = "$MODEL"
def sse_frames(*events: Mapping[str, JsonValue], done: bool = False) -> tuple[str, ...]:
frames: Final = tuple(
f"event: {event['type']}\ndata: {json.dumps(event)}" if "type" in event else f"data: {json.dumps(event)}"
for event in events
)
return (*frames, "data: [DONE]") if done else frames
@dataclass(frozen=True, slots=True, kw_only=True)
class StreamParityTestCase:
"""The same request as two CostTrackingTestCases, `plain` answered with JSON and `streamed` answered with SSE,
that must both bill the one `expected` row."""
plain: CostTrackingTestCase
streamed: CostTrackingTestCase
def __post_init__(self) -> None:
if self.plain.expected != self.streamed.expected:
raise ValueError(f"{self.id}: plain and streamed cases expect different rows")
if self.streamed.request.get("stream") is not True or self.plain.request.get("stream") is True:
raise ValueError(f"{self.id}: streamed case must set stream: true and the plain case must not")
if (self.plain.model, self.plain.endpoint, self.plain.deployment) != (
self.streamed.model,
self.streamed.endpoint,
self.streamed.deployment,
):
raise ValueError(f"{self.id}: plain and streamed cases must share model, endpoint and deployment")
@property
def id(self) -> str:
return self.plain.name

70
tests/integration/cost_calculation/stream_parity/runner.py Normal file
View file

@ -0,0 +1,70 @@
from collections import Counter
from hashlib import sha256
from typing import Final
from integration._support.client import Gateway, Scenario
from integration.cost_calculation.assertions import assert_exact
from integration.cost_calculation.conftest import (
poll_rows,
poll_rows_where,
read_rows_now,
register_scenario_deployment,
)
from integration.cost_calculation.cost_tracking_case import ExactExpected
from integration.cost_calculation.stream_parity.bases import openai_responses
from integration.cost_calculation.stream_parity.case import StreamParityTestCase
FLUSH_CASE: Final = openai_responses.GPT_6_1_SOL_RESPONSES
def flush_spend_logs(scenario: Scenario, gateway: Gateway, key: str) -> str:
"""A /v1/responses row wakes the spend-log writer at once, and the queue is FIFO, so every row queued before it lands."""
flush: Final = register_scenario_deployment(scenario, FLUSH_CASE, "flush", key)
response: Final = gateway.request(
"POST", FLUSH_CASE.endpoint, {**FLUSH_CASE.request, "model": flush.model_name}, key=key
)
assert response.status_code == 200, ("flush", response.text[:400])
poll_rows_where(key, 1, lambda row: row.model_id == flush.identity)
return flush.identity
def assert_stream_parity(case: StreamParityTestCase, gateway: Gateway) -> None:
expected: Final = case.plain.expected
assert isinstance(expected, ExactExpected), case.id
with gateway.scenario() as scenario:
key: Final = scenario.key()
deployments: Final = {
leg.name: register_scenario_deployment(scenario, leg, sha256(leg.name.encode()).hexdigest()[:12], key)
for leg in (case.plain, case.streamed)
}
responses: Final = {
leg.name: gateway.request(
"POST", leg.endpoint, {**leg.request, "model": deployments[leg.name].model_name}, key=key
)
for leg in (case.plain, case.streamed)
}
for leg in (case.plain, case.streamed):
assert responses[leg.name].status_code == 200, (leg.name, responses[leg.name].text[:400])
poll_rows(key, 2)
flush_identity: Final = flush_spend_logs(scenario, gateway, key)
rows: Final = read_rows_now(key)
rows_per_deployment: Final = Counter(row.model_id for row in rows)
expected_rows: Final = {deployments[leg.name].identity: 1 for leg in (case.plain, case.streamed)}
assert rows_per_deployment == {**expected_rows, flush_identity: 1}, (
f"{case.id}: expected exactly one SpendLogs row per leg plus the flush row, got {dict(rows_per_deployment)}",
rows,
)
billed: Final = {
leg.name: next(row for row in rows if row.model_id == deployments[leg.name].identity)
for leg in (case.plain, case.streamed)
}
for leg in (case.plain, case.streamed):
assert_exact(leg.name, leg.response.content_type, expected, billed[leg.name], responses[leg.name])
plain: Final = billed[case.plain.name]
streamed: Final = billed[case.streamed.name]
assert (streamed.spend, streamed.prompt_tokens, streamed.completion_tokens) == (
plain.spend,
plain.prompt_tokens,
plain.completion_tokens,
), (plain, streamed)

40
tests/integration/cost_calculation/stream_parity/test_basic.py Normal file
View file

@ -0,0 +1,40 @@
"""A streamed request bills exactly what the same request bills without streaming.
Streaming reconstructs the response, its usage and its model name from chunks after the fact, in a separate code
path from the non-streamed one (LIT-6241, LIT-6872, LIT-7632, LIT-7729, LIT-9044, LIT-9065). Each case is a plain and
a streamed CostTrackingTestCase for one request, sent through the same key, and both LiteLLM_SpendLogs rows must
bill the case's expected row.
Three cells are the native routes (chat, /v1/messages, /v1/responses); the rest are the bridges that translate one
API into another before the provider call, each with its own stream reassembly
"""
from typing import Final
import pytest
from integration._support.client import Gateway
from integration.cost_calculation.stream_parity.bases import anthropic, bedrock, openai, openai_responses, xai
from integration.cost_calculation.stream_parity.case import StreamParityTestCase
from integration.cost_calculation.stream_parity.runner import assert_stream_parity
NATIVE: Final = (
openai.GPT_6_1_SOL_CHAT_COMPLETIONS_PARITY,
anthropic.CLAUDE_OPUS_5_5_MESSAGES_PARITY,
openai_responses.GPT_6_1_SOL_RESPONSES_PARITY,
)
BRIDGES: Final = (
xai.GROK_4_7_MESSAGES_PARITY,
openai.GPT_6_1_SOL_MESSAGES_PARITY,
bedrock.CLAUDE_OPUS_5_5_CONVERSE_MESSAGES_PARITY,
openai_responses.GPT_5_5_PRO_CHAT_COMPLETIONS_PARITY,
anthropic.CLAUDE_OPUS_5_5_RESPONSES_PARITY,
)
CASES: Final = (*NATIVE, *BRIDGES)
@pytest.mark.parametrize("case", CASES, ids=[case.id for case in CASES])
@pytest.mark.timeout(180)
def test_streamed_request_bills_the_same_row_as_the_non_streamed_request(
case: StreamParityTestCase, gateway: Gateway
) -> None:
assert_stream_parity(case, gateway)

12
tests/integration/providers/test_databricks_decisions_wire.py
View file

@ -179,7 +179,7 @@ def _assert_refused_before_any_upstream_call(gateway: Gateway, model: str, messa
assert message in response.text, response.text
assert _upstream_calls(gateway, handle) == []
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (row["status"], row["call_type"], _number(row["spend"])) == ("failure", "adecisions", 0.0), row
assert (row["status"], row["call_type"], _number(row["spend"])) == ("failure", "asystemone", 0.0), row
@pytest.mark.parametrize("name", [name for _, name in _NON_BARE_NAMES], ids=[label for label, _ in _NON_BARE_NAMES])
@ -251,7 +251,7 @@ def test_an_openai_format_request_at_v1_decisions_reaches_the_serving_endpoint_a
row["model_group"],
row["api_base"],
_number(row["spend"]),
) == ("success", "adecisions", "databricks", deployment, f"{handle.api_base()}/{_ENDPOINT}/invocations", 0.0), (
) == ("success", "asystemone", "databricks", deployment, f"{handle.api_base()}/{_ENDPOINT}/invocations", 0.0), (
row
)
@ -284,7 +284,7 @@ def test_a_wildcard_deployment_sends_each_request_to_the_named_endpoint_and_bill
row["model_group"],
row["api_base"],
_number(row["spend"]),
) == ("success", "adecisions", "databricks", model, f"{handle.api_base()}/{_ENDPOINT}/invocations", 0.0), (
) == ("success", "asystemone", "databricks", model, f"{handle.api_base()}/{_ENDPOINT}/invocations", 0.0), (
row
)
@ -322,7 +322,7 @@ def test_a_five_kilobyte_endpoint_name_reaches_the_upstream_and_its_not_found_an
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (row["status"], row["call_type"], row["model_group"], _number(row["spend"])) == (
"failure",
"adecisions",
"asystemone",
f"{prefix}/{name}",
0.0,
), row
@ -343,7 +343,7 @@ def test_a_databricks_error_keeps_its_status_and_message_and_bills_nothing(
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (row["status"], row["call_type"], row["model_group"], _number(row["spend"])) == (
"failure",
"adecisions",
"asystemone",
model,
0.0,
), row
@ -515,4 +515,4 @@ def test_a_megabyte_endpoint_name_is_refused_in_linear_time_while_liveliness_sta
assert _BARE_NAME_RULE in response.text, response.text[-400:]
assert _upstream_calls(gateway, handle) == []
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (row["status"], row["call_type"], _number(row["spend"])) == ("failure", "adecisions", 0.0), row
assert (row["status"], row["call_type"], _number(row["spend"])) == ("failure", "asystemone", 0.0), row

8
tests/integration/providers/test_decisions_wire.py
View file

@ -278,7 +278,7 @@ def test_each_provider_gets_its_own_path_key_and_body_and_is_billed(gateway: Gat
row["api_base"],
row["prompt_tokens"],
row["completion_tokens"],
) == ("success", "adecisions", provider.name, model, f"{handle.api_base()}{provider.path}", 367, 3)
) == ("success", "asystemone", provider.name, model, f"{handle.api_base()}{provider.path}", 367, 3)
assert math.isclose(_number(row["spend"]), expected_spend, rel_tol=1e-9), row
@ -339,7 +339,7 @@ async def test_sdk_sync_and_async_clients_send_the_same_request(gateway: Gateway
synchronous: Final = litellm.decisions(
model=provider.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=handle.api_base(), api_key=_API_KEY
)
asynchronous: Final = await litellm.adecisions(
asynchronous: Final = await litellm.asystemone(
model=provider.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=handle.api_base(), api_key=_API_KEY
)
for response in (synchronous, asynchronous):
@ -507,7 +507,7 @@ def test_upstream_errors_keep_their_status_and_log_an_unbilled_failure(gateway:
row: Final = _spend_row(response.headers["x-litellm-call-id"])
assert (row["status"], row["call_type"], row["model_group"], _number(row["spend"])) == (
"failure",
"adecisions",
"asystemone",
model,
0.0,
)
@ -579,7 +579,7 @@ async def test_sdk_openrouter_connection_failures_raise_a_connection_error(gatew
model=_OPENROUTER.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=dead_api_base, api_key=_API_KEY
)
with pytest.raises(litellm.APIConnectionError):
await litellm.adecisions(
await litellm.asystemone(
model=_OPENROUTER.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=dead_api_base, api_key=_API_KEY
)

6
tests/integration/providers/test_strands_decider_agentcore_wire.py
View file

@ -415,7 +415,7 @@ def test_an_arn_deployment_answers_on_every_decisions_route_signed_for_bedrock_a
row = _spend_row(response.headers["x-litellm-call-id"])
assert row == {
"status": "success",
"call_type": "adecisions",
"call_type": "asystemone",
"custom_llm_provider": "strands_decider",
"model_group": _SIGNED,
"api_base": _invocations_url("audit_signed"),
@ -694,7 +694,7 @@ async def test_sdk_sync_and_async_calls_reach_the_runtime_signed(
aws_access_key_id=_ACCESS_KEY,
aws_secret_access_key=_SECRET_KEY,
)
asynchronous: Final = await litellm.adecisions(
asynchronous: Final = await litellm.asystemone(
model=_MODEL,
state=_STATE,
questions=_SDK_QUESTIONS,
@ -703,7 +703,7 @@ async def test_sdk_sync_and_async_calls_reach_the_runtime_signed(
aws_secret_access_key=_SECRET_KEY,
)
monkeypatch.setenv("STRANDS_DECIDER_API_KEY", _ENV_JWT)
bearer: Final = await litellm.adecisions(
bearer: Final = await litellm.asystemone(
model=_MODEL, state=_STATE, questions=_SDK_QUESTIONS, api_base=_arn("audit_sdk_env_key")
)
finally:

333
tests/integration/providers/test_vertex_timestamps_utc_wire.py Normal file
View file

@ -0,0 +1,333 @@
import json
import socket
import threading
import uuid
from collections.abc import Generator, Iterator
from contextlib import contextmanager, suppress
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Final
from urllib.parse import parse_qs, urlsplit
import pytest
import yaml
from integration._support.client import JSON_OBJECT, Gateway, Scenario, gateway_from_environment, string_value
from integration._support.process import owned_proxy
from integration._support.tls import server_context, write_self_signed_cert
from integration._support.vertex import service_account_json
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue
pytestmark: Final = pytest.mark.timeout(240)
PROJECT: Final = "scripted-utc-project"
LOCATION: Final = "us-central1"
BUCKET: Final = "scripted-utc-bucket"
GCS_AUTHORITY: Final = "storage.googleapis.com:443"
JOBS_PATH: Final = f"projects/{PROJECT}/locations/{LOCATION}/batchPredictionJobs"
OUTPUT_PREFIX: Final = f"gs://{BUCKET}/litellm-vertex-files/publishers/google/models/gemini-2.5-flash"
VEO_MODEL: Final = "veo-3.1-generate-001"
VEO_OPERATION: Final = f"projects/{PROJECT}/locations/{LOCATION}/publishers/google/models/{VEO_MODEL}/operations/op-1"
UPLOAD_TIME_CREATED: Final = "2026-10-08T21:15:30.250Z"
UPLOAD_EPOCH: Final = 1791494130
BATCH_CREATE_TIME: Final = "2026-10-08T21:16:05.123456Z"
WHOLE_SECOND_BATCH_CREATE_TIME: Final = "2026-10-08T21:16:05Z"
WHOLE_SECOND_MODEL: Final = "gemini-2.5-flash-lite"
WHOLE_SECOND_JOB_SUFFIX: Final = "1"
BATCH_EPOCH: Final = 1791494165
OUTPUT_TIME_CREATED: Final = "2026-10-08T23:41:07.902Z"
OUTPUT_EPOCH: Final = 1791502867
OUTPUT_BYTES: Final = 2068
VEO_CREATE_TIME: Final = "2026-10-08T22:00:00.500000Z"
VEO_EPOCH: Final = 1791496800
HOST_ZONE: Final = "America/Los_Angeles"
def _gcs_object(name: str, size: int, time_created: str) -> bytes:
return json.dumps(
{
"kind": "storage#object",
"id": f"{BUCKET}/{name}/1791494130250000",
"name": name,
"bucket": BUCKET,
"size": str(size),
"contentType": "application/jsonl",
"timeCreated": time_created,
"updated": time_created,
}
).encode()
def _batch_job(job_id: str, state: str, output_directory: str | None) -> bytes:
create_time: Final = (
WHOLE_SECOND_BATCH_CREATE_TIME if job_id.endswith(WHOLE_SECOND_JOB_SUFFIX) else BATCH_CREATE_TIME
)
return json.dumps(
{
"name": f"{JOBS_PATH}/{job_id}",
"displayName": "litellm-vertex-batch-scripted",
"model": "publishers/google/models/gemini-2.5-flash",
"state": state,
"outputInfo": {"gcsOutputDirectory": output_directory} if output_directory else None,
"createTime": create_time,
"updateTime": create_time,
}
).encode()
def vertex_peer(request: Request) -> Reply:
path: Final = urlsplit(request.target).path
if request.method == "POST" and path == "/_oauth/token":
return Reply(body=b'{"access_token":"scripted-token","expires_in":3600,"token_type":"Bearer"}')
if request.method == "POST" and path == f"/upload/storage/v1/b/{BUCKET}/o":
name: Final = parse_qs(urlsplit(request.target).query)["name"][0]
return Reply(body=_gcs_object(name, len(request.body), UPLOAD_TIME_CREATED))
if request.method == "POST" and path.endswith("/batchPredictionJobs"):
suffix: Final = WHOLE_SECOND_JOB_SUFFIX if WHOLE_SECOND_MODEL.encode() in request.body else "0"
return Reply(body=_batch_job(str(uuid.uuid4().int)[:18] + suffix, "JOB_STATE_PENDING", None))
if request.method == "GET" and f"/{JOBS_PATH}/" in path:
job_id: Final = path.rsplit("/", 1)[-1]
return Reply(body=_batch_job(job_id, "JOB_STATE_SUCCEEDED", f"{OUTPUT_PREFIX}/prediction-model-{job_id}"))
if request.method == "GET" and path.startswith(f"/storage/v1/b/{BUCKET}/o/"):
object_name: Final = path.removeprefix(f"/storage/v1/b/{BUCKET}/o/").replace("%2F", "/")
return Reply(body=_gcs_object(object_name, OUTPUT_BYTES, OUTPUT_TIME_CREATED))
if request.method == "POST" and path.endswith(f"/{VEO_MODEL}:predictLongRunning"):
return Reply(body=json.dumps({"name": VEO_OPERATION}).encode())
if request.method == "POST" and path.endswith(f"/{VEO_MODEL}:fetchPredictOperation"):
return Reply(
body=json.dumps(
{"name": VEO_OPERATION, "done": False, "metadata": {"createTime": VEO_CREATE_TIME}}
).encode()
)
return Reply(status=404, body=b'{"error":"not scripted"}')
def _pipe(source: socket.socket, sink: socket.socket) -> None:
with suppress(OSError):
for chunk in iter(lambda: source.recv(65536), b""):
sink.sendall(chunk)
with suppress(OSError):
sink.shutdown(socket.SHUT_WR)
@contextmanager
def gcs_tunnel(destination: Wire) -> Generator[str, None, None]:
"""HTTPS_PROXY that sends CONNECT storage.googleapis.com:443 to the TLS peer and refuses every other host."""
destination_port: Final = urlsplit(destination.url).port
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
timeout = 30
def do_CONNECT(self) -> None:
if self.path != GCS_AUTHORITY:
self.send_error(403)
return
self.send_response(200, "Connection Established")
self.end_headers()
with socket.create_connection(("127.0.0.1", destination_port), timeout=10) as upstream:
outbound: Final = threading.Thread(target=_pipe, args=(self.connection, upstream), daemon=True)
outbound.start()
_pipe(upstream, self.connection)
outbound.join(timeout=30)
def log_message(self, format: str, *args: object) -> None:
pass
class Server(ThreadingHTTPServer):
daemon_threads = True
with Server(("127.0.0.1", 0), Handler) as server:
thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
thread.start()
try:
yield f"http://127.0.0.1:{server.server_port}"
finally:
server.shutdown()
thread.join(timeout=6)
server.server_close()
@dataclass(frozen=True, slots=True)
class Rig:
gateway: Gateway
api: Wire
gcs: Wire
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
directory: Final = tmp_path_factory.mktemp("vertex-utc")
certificate, key = write_self_signed_cert(directory, names=("storage.googleapis.com",))
with (
gateway_from_environment() as environment,
wire_server(vertex_peer) as api,
wire_server(vertex_peer, tls=server_context(certificate, key)) as gcs,
gcs_tunnel(gcs) as tunnel,
owned_proxy(
environment,
directory,
{
"TZ": HOST_ZONE,
"HTTPS_PROXY": tunnel,
"https_proxy": tunnel,
"SSL_CERT_FILE": str(certificate),
"NO_PROXY": "127.0.0.1,localhost",
"no_proxy": "127.0.0.1,localhost",
},
config=_config_allowing(directory, api),
) as candidate,
):
yield Rig(candidate, api, gcs)
def _config_allowing(directory: Path, api: Wire) -> Path:
shipped: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
configuration: Final = {
**shipped,
"litellm_settings": {**shipped["litellm_settings"], "user_url_allowed_hosts": [urlsplit(api.url).netloc]},
}
path: Final = directory / "vertex-utc.yaml"
path.write_text(yaml.safe_dump(configuration))
return path
def _vertex_model(rig: Rig, scenario: Scenario, model: str) -> str:
return scenario.model(
model=model,
api_key=None,
api_base=rig.api.url,
vertex_project=PROJECT,
vertex_location=LOCATION,
vertex_credentials=service_account_json(PROJECT, rig.api.url),
gcs_bucket_name=BUCKET,
)
def _object(response_content: bytes) -> dict[str, JsonValue]:
return JSON_OBJECT.validate_json(response_content)
def test_managed_vertex_batch_and_output_file_created_at_are_utc_epochs(rig: Rig) -> None:
with rig.gateway.scenario() as scenario:
model: Final = _vertex_model(rig, scenario, "vertex_ai/gemini-2.5-flash")
owner_key: Final = scenario.key(user_id=scenario.user(user_role="internal_user"), models=[model])
line: Final = {
"custom_id": "req-1",
"method": "POST",
"url": "/v1/chat/completions",
"body": {"model": model, "messages": [{"role": "user", "content": "ping"}], "max_tokens": 8},
}
uploaded: Final = rig.gateway.request_multipart(
"/v1/files",
{"purpose": "batch", "target_model_names": model},
{"file": ("input.jsonl", (json.dumps(line) + "\n").encode(), "application/jsonl")},
key=owner_key,
)
assert uploaded.status_code == 200, uploaded.text
upload_object: Final = _object(uploaded.content)
created: Final = rig.gateway.request(
"POST",
"/v1/batches",
{
"input_file_id": string_value(upload_object["id"]),
"endpoint": "/v1/chat/completions",
"completion_window": "24h",
},
key=owner_key,
)
assert created.status_code == 200, created.text
batch: Final = _object(created.content)
retrieved: Final = rig.gateway.request("GET", f"/v1/batches/{string_value(batch['id'])}", key=owner_key)
assert retrieved.status_code == 200, retrieved.text
completed: Final = _object(retrieved.content)
assert completed["status"] == "completed", completed
output_id: Final = string_value(completed["output_file_id"])
details: Final = rig.gateway.request("GET", f"/v1/files/{output_id}", key=owner_key)
assert details.status_code == 200, details.text
output: Final = _object(details.content)
assert (output["id"], output["bytes"]) == (output_id, OUTPUT_BYTES), output
gcs_reads: Final = tuple(
request for request in rig.gcs.drain() if request.method == "GET" and "/storage/v1/b/" in request.target
)
assert gcs_reads, "The output file details never came from the GCS object metadata"
listed: Final = rig.gateway.request("GET", "/v1/files", key=owner_key)
assert listed.status_code == 200, listed.text
listed_files: Final = _object(listed.content)["data"]
assert isinstance(listed_files, list), listed.text
listed_output: Final = next(
(
JSON_OBJECT.validate_python(file)
for file in listed_files
if isinstance(file, dict) and file.get("id") == output_id
),
None,
)
assert listed_output is not None, f"Output file {output_id} is absent from GET /v1/files: {listed.text}"
assert {
"POST /v1/files": upload_object["created_at"],
"POST /v1/batches": batch["created_at"],
"GET /v1/batches/{id}": completed["created_at"],
"GET /v1/files/{output}": output["created_at"],
"GET /v1/files": listed_output["created_at"],
} == {
"POST /v1/files": UPLOAD_EPOCH,
"POST /v1/batches": BATCH_EPOCH,
"GET /v1/batches/{id}": BATCH_EPOCH,
"GET /v1/files/{output}": OUTPUT_EPOCH,
"GET /v1/files": OUTPUT_EPOCH,
}
def test_managed_vertex_batch_with_whole_second_create_time_is_created_and_retrieved(rig: Rig) -> None:
with rig.gateway.scenario() as scenario:
model: Final = _vertex_model(rig, scenario, f"vertex_ai/{WHOLE_SECOND_MODEL}")
owner_key: Final = scenario.key(user_id=scenario.user(user_role="internal_user"), models=[model])
line: Final = {
"custom_id": "req-1",
"method": "POST",
"url": "/v1/chat/completions",
"body": {"model": model, "messages": [{"role": "user", "content": "ping"}], "max_tokens": 8},
}
uploaded: Final = rig.gateway.request_multipart(
"/v1/files",
{"purpose": "batch", "target_model_names": model},
{"file": ("input.jsonl", (json.dumps(line) + "\n").encode(), "application/jsonl")},
key=owner_key,
)
assert uploaded.status_code == 200, uploaded.text
created: Final = rig.gateway.request(
"POST",
"/v1/batches",
{
"input_file_id": string_value(_object(uploaded.content)["id"]),
"endpoint": "/v1/chat/completions",
"completion_window": "24h",
},
key=owner_key,
)
assert created.status_code == 200, created.text
batch: Final = _object(created.content)
retrieved: Final = rig.gateway.request("GET", f"/v1/batches/{string_value(batch['id'])}", key=owner_key)
assert retrieved.status_code == 200, retrieved.text
assert (batch["created_at"], _object(retrieved.content)["created_at"]) == (BATCH_EPOCH, BATCH_EPOCH)
def test_veo_video_status_created_at_is_a_utc_epoch(rig: Rig) -> None:
with rig.gateway.scenario() as scenario:
model: Final = _vertex_model(rig, scenario, f"vertex_ai/{VEO_MODEL}")
created: Final = rig.gateway.request("POST", "/v1/videos", {"model": model, "prompt": "a scripted cat"})
assert created.status_code == 200, created.text
video_id: Final = string_value(_object(created.content)["id"])
status: Final = rig.gateway.request("GET", f"/v1/videos/{video_id}")
assert status.status_code == 200, status.text
video: Final = _object(status.content)
assert (video["status"], video["created_at"]) == ("processing", VEO_EPOCH), video

135
tests/unit/decisions/test_main.py
View file

@ -163,7 +163,7 @@ async def test_adecisions_sends_the_provider_wire_contract(
) -> None:
route: Final = respx_mock.post(url).respond(json=_RESPONSE)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model=model,
state={"source": "unit-test"},
questions=_QUESTIONS,
@ -225,7 +225,7 @@ async def test_router_dispatches_typesafe_decisions_without_api_base(
)
upstream: Final = respx_mock.post("https://api.typesafe.ai/v1/systemone").respond(json=_RESPONSE)
response: Final = await router.adecisions(
response: Final = await router.asystemone(
model="jev",
state="router-test",
questions={
@ -256,7 +256,7 @@ async def test_router_dispatches_typesafe_decisions_without_api_base(
def test_decisions_uses_the_same_wire_contract_for_sync_calls(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE)
response: Final = litellm.decisions(
response: Final = litellm.systemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -276,7 +276,7 @@ def test_openrouter_response_keeps_provider_fields(respx_mock: respx.MockRouter)
}
respx_mock.post("https://openrouter.ai/api/alpha/decisions").respond(json=payload)
response: Final = litellm.decisions(
response: Final = litellm.systemone(
model="openrouter/typesafe/jev-1.13",
state="review",
questions=_QUESTIONS,
@ -364,7 +364,7 @@ async def test_decisions_cost_is_in_standard_logging_object(respx_mock: respx.Mo
litellm.callbacks = [recording_logger]
try:
await litellm.adecisions(
await litellm.asystemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -389,7 +389,7 @@ async def test_decisions_cost_is_in_standard_logging_object(respx_mock: respx.Mo
@pytest.mark.asyncio
async def test_unknown_provider_is_rejected_before_http(respx_mock: respx.MockRouter) -> None:
with pytest.raises(litellm.BadRequestError, match="LLM Provider NOT provided"):
await litellm.adecisions(
await litellm.asystemone(
model="unknown/jev-1.13",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -402,7 +402,7 @@ async def test_unknown_provider_is_rejected_before_http(respx_mock: respx.MockRo
@pytest.mark.asyncio
async def test_provider_without_decisions_support_is_rejected_before_http(respx_mock: respx.MockRouter) -> None:
with pytest.raises(litellm.BadRequestError, match=r"Unknown Decisions provider 'anthropic'\. Supported providers"):
await litellm.adecisions(
await litellm.asystemone(
model="anthropic/claude-sonnet-4-5",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -416,7 +416,7 @@ async def test_provider_without_decisions_support_is_rejected_before_http(respx_
async def test_empty_custom_provider_falls_back_to_the_model_prefix(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -436,7 +436,7 @@ async def test_upstream_reply_without_answers_is_a_server_error(respx_mock: resp
)
with pytest.raises(litellm.InternalServerError, match="unexpected response"):
await litellm.adecisions(
await litellm.asystemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -464,7 +464,7 @@ async def test_router_sends_the_openrouter_deployment_key_when_the_provider_is_a
)
upstream: Final = respx_mock.post("https://egress.example/openrouter/alpha/decisions").respond(json=_RESPONSE)
await router.adecisions(
await router.asystemone(
model="jev",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -478,7 +478,7 @@ async def test_router_sends_the_openrouter_deployment_key_when_the_provider_is_a
@pytest.mark.asyncio
async def test_invalid_question_is_rejected_before_http(respx_mock: respx.MockRouter) -> None:
with pytest.raises(litellm.BadRequestError, match="Invalid Decisions request"):
await litellm.adecisions(
await litellm.asystemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"sentiment": {"type": "choice"}},
@ -495,7 +495,7 @@ def test_upstream_bad_request_maps_to_litellm_error(respx_mock: respx.MockRouter
)
with pytest.raises(litellm.BadRequestError):
litellm.decisions(
litellm.systemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -509,7 +509,7 @@ def test_unreachable_upstream_maps_to_a_connection_error(respx_mock: respx.MockR
)
with pytest.raises(litellm.APIConnectionError, match="PerplexityException - Cannot connect to host"):
litellm.decisions(
litellm.systemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -525,7 +525,7 @@ def test_server_key_is_sent_to_an_explicit_api_base(
monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False)
route: Final = respx_mock.post("https://egress.example/perplexity/v1/decisions").respond(json=_RESPONSE)
litellm.decisions(
litellm.systemone(
model="perplexity/pplx-decider-v1-27b",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -555,7 +555,7 @@ async def test_cloudflare_clef_resolves_model_and_response_envelope(
"https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef"
).respond(json=response_body)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model=model,
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -585,7 +585,7 @@ async def test_cloudflare_clef_flash_uses_flash_endpoint_and_request_model(
"https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef-flash"
).respond(json=_RESPONSE)
await litellm.adecisions(
await litellm.asystemone(
model="cloudflare/clef-flash",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -607,7 +607,7 @@ async def test_cloudflare_api_base_from_env_uses_workers_ai_run_path(
"https://api.cloudflare.com/client/v4/accounts/acct/ai/run/@cf/cloudflare/clef"
).respond(json=_RESPONSE)
await litellm.adecisions(
await litellm.asystemone(
model="cloudflare/clef",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -626,7 +626,7 @@ async def test_cloudflare_requires_account_id_or_api_base_before_http(
monkeypatch.setenv("CLOUDFLARE_API_KEY", "cloudflare-key")
with pytest.raises(litellm.BadRequestError, match="Missing CLOUDFLARE_ACCOUNT_ID - set CLOUDFLARE_ACCOUNT_ID"):
await litellm.adecisions(
await litellm.asystemone(
model="cloudflare/clef",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -647,7 +647,7 @@ async def test_cloudflare_clef_cost_uses_the_model_cost_map(
json=_RESPONSE
)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model="cloudflare/clef",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -672,7 +672,7 @@ async def test_strands_decider_requires_api_base_before_http(
monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False)
with pytest.raises(litellm.BadRequestError, match="api_base is required"):
await litellm.adecisions(
await litellm.asystemone(
model="strands_decider/strands-decider-2B-hobson-v19",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -690,7 +690,7 @@ async def test_strands_decider_without_key_preserves_response_extras(
monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False)
route: Final = respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model="strands_decider/strands-decider-2B-hobson-v19",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -717,7 +717,7 @@ async def test_strands_decider_plain_api_base_answers_while_every_aws_signing_th
try:
response: Final = await asyncio.wait_for(
litellm.adecisions(
litellm.asystemone(
model="strands_decider/strands-decider-2B-hobson-v19",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -741,7 +741,7 @@ async def test_strands_decider_uses_key_from_matching_environment_base(
monkeypatch.setenv("STRANDS_DECIDER_API_KEY", "strands-key")
route: Final = respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE)
await litellm.adecisions(
await litellm.asystemone(
model="strands_decider/strands-decider-2B-hobson-v19",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -773,7 +773,7 @@ async def test_strands_decider_provider_resolution_and_router_dispatch(
)
route: Final = respx_mock.post("https://strands.example/v1/systemone").respond(json=_STRANDS_RESPONSE)
response: Final = await router.adecisions(
response: Final = await router.asystemone(
model="strands",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -827,7 +827,7 @@ async def test_strands_decider_runtime_arn_sends_a_sigv4_signed_invoke_agent_run
"aws_secret_access_key": _RUNTIME_CREDENTIALS.secret_key,
}
response: Final = await litellm.adecisions(**call_args) if use_async else litellm.decisions(**call_args)
response: Final = await litellm.asystemone(**call_args) if use_async else litellm.systemone(**call_args)
assert route.call_count == 1
sent: Final = route.calls[0].request
@ -888,7 +888,7 @@ async def test_strands_decider_runtime_arn_with_api_key_sends_the_bearer_token_u
monkeypatch.delenv("STRANDS_DECIDER_API_KEY", raising=False)
route: Final = respx_mock.post(_RUNTIME_URL).respond(json=_STRANDS_RESPONSE)
await litellm.adecisions(
await litellm.asystemone(
model="strands_decider/strands-decider-2B-hobson-v19",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -921,7 +921,7 @@ async def test_strands_decider_runtime_error_envelope_raises_instead_of_answerin
respx_mock.post(_RUNTIME_URL).respond(json={"error": {"code": code, "message": "at most 16 questions"}})
with pytest.raises(error_class, match="at most 16 questions"):
await litellm.adecisions(
await litellm.asystemone(
model="strands_decider/strands-decider-2B-hobson-v19",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -942,7 +942,7 @@ async def test_strands_decider_plain_api_base_error_envelope_stays_an_unexpected
)
with pytest.raises(litellm.InternalServerError, match="returned an unexpected response"):
await litellm.adecisions(
await litellm.asystemone(
model="strands_decider/strands-decider-2B-hobson-v19",
state="review",
questions={"severity": {"type": "score", "criteria": ["none", "low", "high"]}},
@ -977,7 +977,7 @@ async def test_hosted_vllm_requires_api_base_before_http(
monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False)
with pytest.raises(litellm.BadRequestError, match="api_base is required"):
await litellm.adecisions(
await litellm.asystemone(
model="hosted_vllm/Qwen/Qwen3-0.6B",
state="review",
questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}},
@ -995,7 +995,7 @@ async def test_hosted_vllm_without_key_sends_no_authorization_and_preserves_resp
monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False)
route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model="hosted_vllm/Qwen/Qwen3-0.6B",
state="The package arrived broken.",
questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}},
@ -1018,7 +1018,7 @@ async def test_hosted_vllm_uses_key_and_base_from_environment(
monkeypatch.setenv("HOSTED_VLLM_API_KEY", "vllm-key")
route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE)
await litellm.adecisions(
await litellm.asystemone(
model="hosted_vllm/Qwen/Qwen3-0.6B",
state="The package arrived broken.",
questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}},
@ -1037,7 +1037,7 @@ async def test_hosted_vllm_api_base_with_v1_suffix_posts_to_v1_systemone(
monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False)
route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE)
await litellm.adecisions(
await litellm.asystemone(
model="hosted_vllm/Qwen/Qwen3-0.6B",
state="The package arrived broken.",
questions={"is_defect": {"type": "choice", "criteria": {"yes": None, "no": None}}},
@ -1056,7 +1056,7 @@ async def test_hosted_vllm_choice_question_goes_out_as_a_jev_choice_with_the_ser
monkeypatch.delenv("HOSTED_VLLM_API_KEY", raising=False)
route: Final = respx_mock.post("http://vllm.local:8000/v1/systemone").respond(json=_VLLM_RESPONSE)
await litellm.adecisions(
await litellm.asystemone(
model="hosted_vllm/Qwen/Qwen3-0.6B",
state="The package arrived broken.",
questions={
@ -1092,7 +1092,7 @@ async def test_databricks_requires_api_base_before_http(
monkeypatch.setenv("DATABRICKS_API_KEY", "dapi-key")
with pytest.raises(litellm.BadRequestError, match="DATABRICKS_API_BASE"):
await litellm.adecisions(
await litellm.asystemone(
model=f"databricks/{_DATABRICKS_ENDPOINT}",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -1114,7 +1114,7 @@ async def test_databricks_rejects_a_serving_endpoint_name_that_would_rewrite_the
monkeypatch.setenv("DATABRICKS_API_KEY", "dapi-key")
with pytest.raises(litellm.BadRequestError, match="bare endpoint name"):
await litellm.adecisions(
await litellm.asystemone(
model=f"databricks/{endpoint_name}",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -1133,7 +1133,7 @@ async def test_databricks_requires_a_key_before_http(
monkeypatch.delenv("DATABRICKS_TOKEN", raising=False)
with pytest.raises(litellm.AuthenticationError, match="Missing API key for Decisions provider 'databricks'"):
await litellm.adecisions(
await litellm.asystemone(
model=f"databricks/{_DATABRICKS_ENDPOINT}",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -1162,7 +1162,7 @@ async def test_databricks_posts_the_jev_body_to_the_serving_endpoint_invocations
"sentiment": {"type": "choice", "criteria": {"positive": None, "negative": "unhappy"}},
}
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model=f"databricks/{_DATABRICKS_ENDPOINT}",
state={"ticket": "export hangs"},
questions=questions,
@ -1206,7 +1206,7 @@ async def test_databricks_router_deployment_sends_its_own_key_to_its_own_base(
f"https://workspace.example/serving-endpoints/{_DATABRICKS_ENDPOINT}/invocations"
).respond(json=_DATABRICKS_RESPONSE)
response: Final = await router.adecisions(
response: Final = await router.asystemone(
model="decider",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -1260,7 +1260,7 @@ async def test_openai_decisions_translate_systemone_to_the_openai_wire_contract_
monkeypatch.delenv("OPENAI_API_BASE", raising=False)
route: Final = respx_mock.post(url).respond(json=_OPENAI_RESPONSE)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model="openai/gpt-6-luna",
state="The package arrived broken.",
questions={
@ -1345,7 +1345,7 @@ def test_openai_decisions_use_the_same_settings_as_other_openai_calls(
monkeypatch.setenv(name, value)
route: Final = respx_mock.post(url).respond(json=_OPENAI_RESPONSE)
litellm.decisions(
litellm.systemone(
model="openai/gpt-6-luna",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -1408,17 +1408,8 @@ async def test_openai_format_calls_to_a_systemone_provider_get_openai_format_ans
},
"cannot serve this request",
),
(
"openai/gpt-6-luna",
{
"state": "review",
"input": "review",
"questions": [{"type": "predicate", "instructions": "Is this a defect?"}],
},
"not both",
),
),
ids=("image_to_systemone_provider", "state_and_input"),
ids=("image_to_systemone_provider",),
)
async def test_requests_a_provider_cannot_serve_are_rejected_before_http(
respx_mock: respx.MockRouter,
@ -1451,7 +1442,7 @@ async def test_openrouter_decisions_uses_provider_reported_cost_without_cost_map
}
)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model="openrouter/typesafe/jev-1.13",
state={"source": "unit-test"},
questions=_QUESTIONS,
@ -1466,24 +1457,27 @@ _SAFETY_IDENTIFIER: Final = "end-user-7"
_PREDICATE_QUESTIONS: Final[tuple[Mapping[str, object], ...]] = (
{"type": "predicate", "name": "is_defect", "instructions": "Is this a defect?"},
)
_PREDICATE_REQUESTS: Final[tuple[Mapping[str, object], ...]] = (
MappingProxyType({"input": "review", "questions": _PREDICATE_QUESTIONS}),
MappingProxyType(
{"state": "review", "questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}}
_PREDICATE_REQUESTS: Final[tuple[tuple[str, Mapping[str, object]], ...]] = (
("decisions", MappingProxyType({"input": "review", "questions": _PREDICATE_QUESTIONS})),
(
"systemone",
MappingProxyType(
{"state": "review", "questions": {"is_defect": {"type": "noul", "instructions": "Is this a defect?"}}}
),
),
)
_PREDICATE_REQUEST_IDS: Final = ("input_format", "state_format")
_PREDICATE_REQUEST_IDS: Final = ("decisions", "systemone")
@pytest.mark.parametrize("request_kwargs", _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
@pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
def test_safety_identifier_is_refused_before_http_when_the_provider_cannot_take_it(
request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
api: str, request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
route: Final = respx_mock.post("https://api.perplexity.ai/v1/decisions").respond(json=_RESPONSE)
with pytest.raises(litellm.UnsupportedParamsError, match=r"safety_identifier.*drop_params") as caught:
litellm.decisions(
getattr(litellm, api)(
model="perplexity/pplx-decider-v1-27b",
safety_identifier=_SAFETY_IDENTIFIER,
api_key="caller-key",
@ -1524,10 +1518,11 @@ async def test_safety_identifier_is_dropped_from_the_wire_under_drop_params(
@pytest.mark.asyncio
@pytest.mark.parametrize("request_kwargs", _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
@pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
@pytest.mark.parametrize("drop_params", (False, True), ids=("strict", "drop_params"))
async def test_safety_identifier_reaches_openai_whether_or_not_params_are_dropped(
drop_params: bool,
api: str,
request_kwargs: Mapping[str, object],
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
@ -1538,7 +1533,7 @@ async def test_safety_identifier_reaches_openai_whether_or_not_params_are_droppe
monkeypatch.delenv("OPENAI_API_BASE", raising=False)
route: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(json=_OPENAI_RESPONSE)
await litellm.adecisions(
await getattr(litellm, f"a{api}")(
model="openai/gpt-6-luna",
safety_identifier=_SAFETY_IDENTIFIER,
api_key="caller-key",
@ -1555,16 +1550,16 @@ async def test_safety_identifier_reaches_openai_whether_or_not_params_are_droppe
@pytest.mark.asyncio
@pytest.mark.parametrize("request_kwargs", _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
@pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
async def test_non_string_safety_identifier_is_rejected_before_http(
request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
api: str, request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
route: Final = respx_mock.post("https://api.openai.com/v1/decisions").respond(json=_OPENAI_RESPONSE)
invalid_safety_identifier: Final[Mapping[str, object]] = MappingProxyType({"safety_identifier": 7})
with pytest.raises(litellm.BadRequestError, match=r"(?s)Invalid Decisions request.*safety_identifier"):
await litellm.adecisions(
await getattr(litellm, f"a{api}")(
model="openai/gpt-6-luna", api_key="caller-key", **request_kwargs, **invalid_safety_identifier
)
@ -1572,9 +1567,9 @@ async def test_non_string_safety_identifier_is_rejected_before_http(
@pytest.mark.asyncio
@pytest.mark.parametrize("request_kwargs", _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
@pytest.mark.parametrize(("api", "request_kwargs"), _PREDICATE_REQUESTS, ids=_PREDICATE_REQUEST_IDS)
async def test_non_string_safety_identifier_is_dropped_from_the_wire_under_drop_params(
request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
api: str, request_kwargs: Mapping[str, object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
monkeypatch.setattr(litellm, "api_base", None)
@ -1585,7 +1580,7 @@ async def test_non_string_safety_identifier_is_dropped_from_the_wire_under_drop_
{"safety_identifier": 7, "drop_params": True}
)
await litellm.adecisions(
await getattr(litellm, f"a{api}")(
model="openai/gpt-6-luna", api_key="caller-key", **request_kwargs, **dropped_safety_identifier
)
@ -1622,7 +1617,7 @@ async def test_azure_ai_decision_posts_deployment_to_foundry_systemone_route(
json={**_RESPONSE, "model": "microsoft-decision-1"}
)
response: Final = await litellm.adecisions(
response: Final = await litellm.asystemone(
model="azure_ai/decision-1",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -1654,7 +1649,7 @@ async def test_azure_ai_decision_reads_foundry_env_and_keeps_gateway_path_prefix
json=_RESPONSE
)
await litellm.adecisions(
await litellm.asystemone(
model="azure_ai/decision-1",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},
@ -1673,7 +1668,7 @@ async def test_azure_ai_decision_requires_api_base_before_http(
monkeypatch.setenv("AZURE_AI_API_KEY", "foundry-key")
with pytest.raises(litellm.BadRequestError, match="Missing AZURE_AI_API_BASE"):
await litellm.adecisions(
await litellm.asystemone(
model="azure_ai/decision-1",
state="review",
questions={"is_defect": {"type": "noul", "instructions": "Is this a defect?"}},

1
tests/unit/llms/azure/test_azure_common_utils.py
View file

@ -471,6 +471,7 @@ def test_default_max_retries_env_var_reaches_azure_sdk_client():
"llm_passthrough_route",
"asearch",
"adecisions",
"asystemone",
"avector_store_create",
"avector_store_search",
"acreate_skill",

10
tests/unit/llms/vertex_ai/batches/test_transformation.py
View file

@ -22,10 +22,7 @@ import pytest
from litellm.llms.vertex_ai.batches.transformation import ( # noqa: E402
VertexAIBatchTransformation,
)
from litellm.llms.vertex_ai.common_utils import ( # noqa: E402
VertexAIError,
convert_vertex_datetime_to_openai_datetime,
)
from litellm.llms.vertex_ai.common_utils import VertexAIError # noqa: E402
from litellm.types.utils import LiteLLMBatch # noqa: E402
T = VertexAIBatchTransformation
@ -142,10 +139,7 @@ def test_transform_vertex_response_full_mapping():
assert isinstance(batch, LiteLLMBatch)
assert batch.id == "3814889423749775360"
assert batch.completion_window == "24h"
# created_at is parsed via the shared helper (uses local tz); assert the
# transform forwards createTime through that helper rather than a hardcoded
# epoch that would be tz-dependent
assert batch.created_at == convert_vertex_datetime_to_openai_datetime("2024-12-04T21:53:12.120184Z")
assert batch.created_at == 1733349192
assert batch.endpoint == ""
assert batch.object == "batch"
assert batch.input_file_id == "gs://bucket/in.jsonl"

31
tests/unit/llms/vertex_ai/test_vertex_ai_common_utils.py
View file

@ -1,3 +1,5 @@
import time
from collections.abc import Iterator
from unittest.mock import patch
import pytest
@ -5,6 +7,7 @@ import pytest
from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
from litellm.llms.vertex_ai.common_utils import (
convert_anyof_null_to_nullable,
convert_vertex_datetime_to_openai_datetime,
get_vertex_location_from_url,
get_vertex_project_id_from_url,
get_vertex_url,
@ -1882,3 +1885,31 @@ def test_get_vertex_ai_lyria_model_info_is_none_for_non_lyria_speech_models(mode
assert get_vertex_ai_lyria_model_info(model=model) is None
@pytest.fixture
def host_zone(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setenv("TZ", request.param)
time.tzset()
yield
monkeypatch.undo()
time.tzset()
@pytest.mark.parametrize(
"host_zone", ["UTC", "America/Los_Angeles", "Asia/Kolkata", "Pacific/Kiritimati"], indirect=True
)
@pytest.mark.parametrize(
("vertex_datetime", "utc_epoch"),
[
("2024-12-04T21:53:12.120184Z", 1733349192),
("2026-10-08T23:41:07.902Z", 1791502867),
("2026-01-15T08:00:00.000000Z", 1768464000),
("2026-09-09T21:39:16Z", 1788989956),
("2026-10-09T08:19:53.576944123Z", 1791533993),
],
)
@pytest.mark.usefixtures("host_zone")
@pytest.mark.skipif(not hasattr(time, "tzset"), reason="switching the process timezone needs time.tzset()")
def test_convert_vertex_datetime_to_openai_datetime_is_the_utc_epoch_on_any_host_zone(
vertex_datetime: str, utc_epoch: int
) -> None:
assert convert_vertex_datetime_to_openai_datetime(vertex_datetime) == utc_epoch

2
tests/unit/proxy/guardrails/guardrail_hooks/test_decision_model.py
View file

@ -414,7 +414,7 @@ def test_initialize_rejects_invalid_streaming_scan_settings(mock_mgr, overrides,
def test_add_guardrail_settings_returns_decisions_providers():
import asyncio
from litellm.decisions.main import supported_decisions_providers
from litellm.decisions.call import supported_decisions_providers
from litellm.proxy.guardrails.guardrail_endpoints import get_guardrail_ui_settings
settings: Final = asyncio.run(get_guardrail_ui_settings())

0
tests/unit/systemone/__init__.py Normal file
View file

138
tests/unit/systemone/test_main.py Normal file
View file

@ -0,0 +1,138 @@
from __future__ import annotations
import asyncio
import json
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
import pytest
import respx
from respx.models import Call
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.types.decisions import DecisionsResponse, OpenAIDecisionResponse
_STATE: Final = "Customer asked for a refund"
_SYSTEMONE_QUESTIONS: Final[Mapping[str, object]] = MappingProxyType(
{"is_refund": {"type": "noul", "instructions": "Is this a refund request?"}}
)
_OPENAI_QUESTIONS: Final[tuple[Mapping[str, object], ...]] = (
{"type": "predicate", "name": "is_refund", "instructions": "Is this a refund request?"},
)
_SYSTEMONE_RESPONSE: Final[Mapping[str, object]] = {
"model": "jev-1.13",
"answers": {"is_refund": {"type": "noul", "noul": 0.93}},
"usage": {"input_tokens": 12, "output_tokens": 1},
}
_TYPESAFE_URL: Final = "https://api.typesafe.ai/v1/systemone"
@pytest.fixture(autouse=True)
def _httpx_transport(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
litellm.in_memory_llm_clients_cache.flush_cache()
class _CallTypeRecorder(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.call_types: list[str] = [] # mutable-ok: test recorder collects logger callbacks
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:
self.call_types.append(kwargs["call_type"])
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", (True, False), ids=("asystemone", "systemone"))
async def test_systemone_sends_the_systemone_wire_format_and_returns_a_systemone_response(
use_async: bool, respx_mock: respx.MockRouter
) -> None:
route: Final = respx_mock.post(_TYPESAFE_URL).respond(json=_SYSTEMONE_RESPONSE)
kwargs: Final[Mapping[str, object]] = MappingProxyType(
{"model": "typesafe/jev-1.13", "state": _STATE, "questions": _SYSTEMONE_QUESTIONS, "api_key": "caller-key"}
)
response: Final = await litellm.asystemone(**kwargs) if use_async else litellm.systemone(**kwargs)
assert route.called
call: Final[Call] = respx_mock.calls.last
assert json.loads(call.request.content) == {
"model": "jev-1.13",
"state": _STATE,
"questions": {"is_refund": {"type": "noul", "instructions": "Is this a refund request?"}},
}
assert isinstance(response, DecisionsResponse)
assert response.answers["is_refund"].model_dump(mode="json") == {"type": "noul", "noul": 0.93}
@pytest.mark.asyncio
async def test_systemone_rejects_openai_input_and_points_at_decisions(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post(_TYPESAFE_URL).respond(json=_SYSTEMONE_RESPONSE)
with pytest.raises(litellm.BadRequestError, match=r"litellm\.decisions") as caught:
await litellm.asystemone(
model="typesafe/jev-1.13", input=_STATE, questions=_SYSTEMONE_QUESTIONS, api_key="caller-key"
)
assert caught.value.status_code == 400
assert not route.called
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", (True, False), ids=("adecisions", "decisions"))
async def test_decisions_with_state_still_sends_the_systemone_format_and_warns(
use_async: bool, respx_mock: respx.MockRouter
) -> None:
route: Final = respx_mock.post(_TYPESAFE_URL).respond(json=_SYSTEMONE_RESPONSE)
kwargs: Final[Mapping[str, object]] = MappingProxyType(
{"model": "typesafe/jev-1.13", "state": _STATE, "questions": _SYSTEMONE_QUESTIONS, "api_key": "caller-key"}
)
with pytest.warns(UserWarning, match=r"litellm\.systemone"):
response: Final = await litellm.adecisions(**kwargs) if use_async else litellm.decisions(**kwargs)
assert route.called
call: Final[Call] = respx_mock.calls.last
assert json.loads(call.request.content) == {
"model": "jev-1.13",
"state": _STATE,
"questions": {"is_refund": {"type": "noul", "instructions": "Is this a refund request?"}},
}
assert isinstance(response, DecisionsResponse)
assert response.answers["is_refund"].model_dump(mode="json") == {"type": "noul", "noul": 0.93}
@pytest.mark.asyncio
async def test_systemone_rejects_openai_question_lists_before_http(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post(_TYPESAFE_URL).respond(json=_SYSTEMONE_RESPONSE)
with pytest.raises(litellm.BadRequestError, match="Invalid Decisions request"):
await litellm.asystemone(
model="typesafe/jev-1.13", state=_STATE, questions=_OPENAI_QUESTIONS, api_key="caller-key"
)
assert not route.called
@pytest.mark.asyncio
async def test_router_asystemone_logs_the_systemone_call_type(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
respx_mock.post(_TYPESAFE_URL).respond(json=_SYSTEMONE_RESPONSE)
recorder: Final = _CallTypeRecorder()
monkeypatch.setattr(litellm, "callbacks", [recorder])
router: Final = litellm.Router(
model_list=[{"model_name": "jev", "litellm_params": {"model": "typesafe/jev-1.13", "api_key": "caller-key"}}]
)
response: Final = await router.asystemone(model="jev", state=_STATE, questions=_SYSTEMONE_QUESTIONS)
await asyncio.sleep(0)
GLOBAL_LOGGING_WORKER.start()
await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10.0)
assert isinstance(response, DecisionsResponse)
assert recorder.call_types == ["asystemone"]
assert not isinstance(response, OpenAIDecisionResponse)

189
ui/litellm-dashboard/src/app/(dashboard)/layout.integration.test.tsx Normal file
View file

@ -0,0 +1,189 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import userEvent from "@testing-library/user-event";
import { renderWithProviders, screen, testQueryClient, waitFor, within } from "../../../tests/test-utils";
import { AuthProvider } from "@/contexts/AuthContext";
import { Dialog, DialogClose, DialogContent, DialogTitle, DialogTrigger } from "@/components/ui/dialog";
import {
AlertDialog,
AlertDialogCancel,
AlertDialogContent,
AlertDialogTitle,
AlertDialogTrigger,
} from "@/components/ui/alert-dialog";
import { Sheet, SheetClose, SheetContent, SheetTitle, SheetTrigger } from "@/components/ui/sheet";
import Layout from "./layout";
vi.unmock("@/app/(dashboard)/hooks/useAuthorized");
vi.unmock("@/lib/toast");
vi.mock("next/navigation", () => ({
useRouter: () => ({ push: vi.fn(), replace: vi.fn() }),
useSearchParams: () => new URLSearchParams(),
usePathname: () => "/ui/guardrails",
}));
beforeEach(() => {
testQueryClient.clear();
localStorage.clear();
const claims = {
key: "sk-layout-test",
user_id: "layout-admin",
user_role: "proxy_admin",
login_method: "sso",
exp: Date.now() / 1000 + 3600,
};
const token = `${btoa(JSON.stringify({ alg: "none" }))}.${btoa(JSON.stringify(claims))}.test`;
document.cookie = `token=${token}; Path=/`;
vi.stubGlobal(
"fetch",
vi.fn(async (input: RequestInfo | URL) => {
const url = input instanceof Request ? input.url : String(input);
const path = new URL(url, "http://localhost").pathname;
switch (path) {
case "/litellm/.well-known/litellm-ui-config":
return Response.json({ server_root_path: "", proxy_base_url: "", admin_ui_disabled: false });
case "/get/ui_settings":
case "/get/ui_theme_settings":
return Response.json({ values: {} });
case "/api/plugins":
case "/team/list":
case "/organization/list":
return Response.json([]);
case "/health/readiness/details":
return Response.json({ status: "healthy" });
case "/health/license": {
const license = {
has_license: false,
license_type: null,
expiration_date: null,
allowed_features: [],
limits: { max_users: null, max_teams: null },
};
return Response.json(license);
}
case "/user/available_users": {
const usage = {
total_users: null,
total_users_used: 1,
total_users_remaining: null,
total_teams: null,
total_teams_used: 0,
total_teams_remaining: null,
};
return Response.json(usage);
}
case "/sso/get/ui_settings":
return Response.json({});
case "/get/user_banner":
return Response.json({ enabled: false, message: "", severity: "info" });
case "/get/latest_release_info": {
const release = { version: "", release_url: "", bug_fixes: 0, new_features: 0, other_updates: 0 };
return Response.json(release);
}
case "/public/litellm_blog_posts":
return Response.json({ posts: [] });
default:
throw new Error(`Unexpected request: ${url}`);
}
}),
);
});
afterEach(() => {
testQueryClient.clear();
vi.unstubAllGlobals();
document.cookie = "token=; Max-Age=0; Path=/";
localStorage.clear();
});
describe("dashboard overlay boundaries", () => {
it("should render the page with mobile navigation closed", async () => {
renderWithProviders(
<AuthProvider>
<Layout>
<h1>Gateway page</h1>
</Layout>
</AuthProvider>,
);
expect(await screen.findByRole("heading", { name: "Gateway page" })).toBeVisible();
expect(screen.getByRole("button", { name: "Open navigation" })).toBeVisible();
expect(screen.queryByRole("dialog", { name: "Navigation" })).not.toBeInTheDocument();
});
it.each([
{
name: "dialog",
role: "dialog",
slot: "dialog-overlay",
page: (
<Dialog>
<DialogTrigger>Open page overlay</DialogTrigger>
<DialogContent showCloseButton={false}>
<DialogTitle>Page overlay</DialogTitle>
<DialogClose>Close page overlay</DialogClose>
</DialogContent>
</Dialog>
),
},
{
name: "alert dialog",
role: "alertdialog",
slot: "alert-dialog-overlay",
page: (
<AlertDialog>
<AlertDialogTrigger>Open page overlay</AlertDialogTrigger>
<AlertDialogContent>
<AlertDialogTitle>Page overlay</AlertDialogTitle>
<AlertDialogCancel>Close page overlay</AlertDialogCancel>
</AlertDialogContent>
</AlertDialog>
),
},
{
name: "sheet",
role: "dialog",
slot: "sheet-overlay",
page: (
<Sheet>
<SheetTrigger>Open page overlay</SheetTrigger>
<SheetContent showCloseButton={false}>
<SheetTitle>Page overlay</SheetTitle>
<SheetClose>Close page overlay</SheetClose>
</SheetContent>
</Sheet>
),
},
])("should display the page $name backdrop while navigation is closed", async ({ role, slot, page }) => {
const user = userEvent.setup();
renderWithProviders(
<AuthProvider>
<Layout>{page}</Layout>
</AuthProvider>,
);
await user.click(await screen.findByRole("button", { name: "Open page overlay" }));
await screen.findByRole(role, { name: "Page overlay" });
expect(
screen.getAllByRole("presentation", { hidden: true }).find((element) => element.dataset.slot === slot),
).toBeVisible();
await user.click(screen.getByRole("button", { name: "Close page overlay" }));
});
it("should return focus to the navigation trigger after closing the drawer with Escape", async () => {
const user = userEvent.setup();
renderWithProviders(
<AuthProvider>
<Layout>
<h1>Gateway page</h1>
</Layout>
</AuthProvider>,
);
const trigger = await screen.findByRole("button", { name: "Open navigation" });
await user.click(trigger);
const navigation = await screen.findByRole("dialog", { name: "Navigation" });
expect(within(navigation).getByRole("link", { name: "Virtual Keys" })).toBeVisible();
await user.keyboard("{Escape}");
await waitFor(() => expect(trigger).toHaveFocus());
});
});

2
ui/litellm-dashboard/src/app/(dashboard)/layout.test.tsx
View file

@ -1,5 +1,5 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { fireEvent, render, screen, waitFor, within } from "@testing-library/react";
import { fireEvent, renderWithProviders as render, screen, waitFor, within } from "../../../tests/test-utils";
import { usePathname } from "next/navigation";
import { AuthProvider } from "@/contexts/AuthContext";
import Layout from "./layout";

69
ui/litellm-dashboard/src/app/(dashboard)/layout.tsx
View file

@ -1,6 +1,7 @@
"use client";
import React, { Suspense, useState, useRef, useEffect } from "react";
import { Dialog as DialogPrimitive } from "@base-ui/react/dialog";
import { DashboardHeader } from "@/components/DashboardHeader";
import Navbar from "@/components/navbar";
import LoadingScreen from "@/components/common_components/LoadingScreen";
@ -115,6 +116,7 @@ function DashboardShell({ children }: { children: React.ReactNode }) {
const navigationKey = `${pathname}?${searchParams.toString()}`;
const isDesktop = useMediaQuery("(min-width: 768px)", { initializeWithValue: false });
const [mobileNavigationKey, setMobileNavigationKey] = useState<string | null>(null);
const [mobileNavigationHandle] = useState(() => DialogPrimitive.createHandle());
if (mobileNavigationKey !== null && (isDesktop || mobileNavigationKey !== navigationKey)) {
setMobileNavigationKey(null);
}
@ -152,11 +154,15 @@ function DashboardShell({ children }: { children: React.ReactNode }) {
// sidebar owns its own scroll and the content column scrolls independently,
// so the page can't be dragged past the end of the nav.
return (
<Sheet open={mobileNavigationOpen} onOpenChange={(open) => setMobileNavigationKey(open ? navigationKey : null)}>
<div className="flex h-screen overflow-hidden bg-background max-md:h-dvh">
<div className="hidden h-full md:flex">
<SidebarProvider sidebarCollapsed={sidebarCollapsed} onToggleCollapsed={toggleSidebar} />
</div>
<div className="flex h-screen overflow-hidden bg-background max-md:h-dvh">
<div className="hidden h-full md:flex">
<SidebarProvider sidebarCollapsed={sidebarCollapsed} onToggleCollapsed={toggleSidebar} />
</div>
<Sheet
handle={mobileNavigationHandle}
open={mobileNavigationOpen}
onOpenChange={(open) => setMobileNavigationKey(open ? navigationKey : null)}
>
<SheetContent
side="left"
showCloseButton={false}
@ -168,32 +174,33 @@ function DashboardShell({ children }: { children: React.ReactNode }) {
<SheetTitle className="sr-only">Navigation</SheetTitle>
<SidebarProvider sidebarCollapsed={false} onToggleCollapsed={() => setMobileNavigationKey(null)} />
</SheetContent>
<LiteAdminFrame>
<CommandPaletteProvider>
<div className="flex min-w-0 flex-1 flex-col overflow-hidden">
<DashboardHeader
navigationTrigger={
<SheetTrigger
render={
<Button variant="ghost" size="icon" className="size-11 md:hidden" aria-label="Open navigation" />
}
>
<Menu />
</SheetTrigger>
}
/>
<DebugWarningBanner accessToken={accessToken} />
<NoRedisWarningBanner accessToken={accessToken} />
<EnvCredentialLoginWarningBanner accessToken={accessToken} />
<LicenseExpiryBanner accessToken={accessToken} />
<UserBanner accessToken={accessToken} />
<UpgradeBanner accessToken={accessToken} />
<main className="min-h-0 min-w-0 flex-1 overflow-y-auto">{children}</main>
</div>
</CommandPaletteProvider>
</LiteAdminFrame>
</div>
</Sheet>
</Sheet>
<LiteAdminFrame>
<CommandPaletteProvider>
<div className="flex min-w-0 flex-1 flex-col overflow-hidden">
<DashboardHeader
navigationTrigger={
<SheetTrigger
handle={mobileNavigationHandle}
render={
<Button variant="ghost" size="icon" className="size-11 md:hidden" aria-label="Open navigation" />
}
>
<Menu />
</SheetTrigger>
}
/>
<DebugWarningBanner accessToken={accessToken} />
<NoRedisWarningBanner accessToken={accessToken} />
<EnvCredentialLoginWarningBanner accessToken={accessToken} />
<LicenseExpiryBanner accessToken={accessToken} />
<UserBanner accessToken={accessToken} />
<UpgradeBanner accessToken={accessToken} />
<main className="min-h-0 min-w-0 flex-1 overflow-y-auto">{children}</main>
</div>
</CommandPaletteProvider>
</LiteAdminFrame>
</div>
);
}

2
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -28075,7 +28075,7 @@ export interface components {
* CallTypes
* @enum {string}
*/
CallTypes: "embedding" | "aembedding" | "completion" | "acompletion" | "atext_completion" | "text_completion" | "image_generation" | "aimage_generation" | "image_edit" | "aimage_edit" | "moderation" | "amoderation" | "atranscription" | "transcription" | "aspeech" | "speech" | "rerank" | "arerank" | "search" | "asearch" | "decisions" | "adecisions" | "_arealtime" | "_aresponses_websocket" | "create_batch" | "acreate_batch" | "aretrieve_batch" | "retrieve_batch" | "acancel_batch" | "cancel_batch" | "pass_through_endpoint" | "anthropic_messages" | "aanthropic_messages" | "get_assistants" | "aget_assistants" | "create_assistants" | "acreate_assistants" | "delete_assistant" | "adelete_assistant" | "acreate_thread" | "create_thread" | "aget_thread" | "get_thread" | "a_add_message" | "add_message" | "aget_messages" | "get_messages" | "arun_thread" | "run_thread" | "arun_thread_stream" | "run_thread_stream" | "afile_retrieve" | "file_retrieve" | "afile_delete" | "file_delete" | "afile_list" | "file_list" | "acreate_file" | "create_file" | "afile_content" | "file_content" | "create_fine_tuning_job" | "acreate_fine_tuning_job" | "create_video" | "acreate_video" | "video_generation" | "avideo_generation" | "avideo_retrieve" | "video_retrieve" | "avideo_content" | "video_content" | "video_remix" | "avideo_remix" | "video_list" | "avideo_list" | "video_retrieve_job" | "avideo_retrieve_job" | "video_delete" | "avideo_delete" | "video_create_character" | "avideo_create_character" | "video_get_character" | "avideo_get_character" | "video_edit" | "avideo_edit" | "video_extension" | "avideo_extension" | "vector_store_file_create" | "avector_store_file_create" | "vector_store_file_list" | "avector_store_file_list" | "vector_store_file_retrieve" | "avector_store_file_retrieve" | "vector_store_file_content" | "avector_store_file_content" | "vector_store_file_update" | "avector_store_file_update" | "vector_store_file_delete" | "avector_store_file_delete" | "vector_store_create" | "avector_store_create" | "vector_store_search" | "avector_store_search" | "ingest" | "aingest" | "query" | "aquery" | "create_interaction" | "acreate_interaction" | "create_container" | "acreate_container" | "list_containers" | "alist_containers" | "retrieve_container" | "aretrieve_container" | "delete_container" | "adelete_container" | "list_container_files" | "alist_container_files" | "upload_container_file" | "aupload_container_file" | "create_sandbox" | "acreate_sandbox" | "delete_sandbox" | "adelete_sandbox" | "run_code" | "arun_code" | "code_interpreter_tool" | "acode_interpreter_tool" | "acancel_fine_tuning_job" | "cancel_fine_tuning_job" | "alist_fine_tuning_jobs" | "list_fine_tuning_jobs" | "aretrieve_fine_tuning_job" | "retrieve_fine_tuning_job" | "responses" | "aresponses" | "alist_input_items" | "llm_passthrough_route" | "allm_passthrough_route" | "generate_content" | "agenerate_content" | "generate_content_stream" | "agenerate_content_stream" | "ocr" | "aocr" | "call_mcp_tool" | "list_mcp_tools" | "asend_message" | "send_message" | "acreate_skill";
CallTypes: "embedding" | "aembedding" | "completion" | "acompletion" | "atext_completion" | "text_completion" | "image_generation" | "aimage_generation" | "image_edit" | "aimage_edit" | "moderation" | "amoderation" | "atranscription" | "transcription" | "aspeech" | "speech" | "rerank" | "arerank" | "search" | "asearch" | "decisions" | "adecisions" | "systemone" | "asystemone" | "_arealtime" | "_aresponses_websocket" | "create_batch" | "acreate_batch" | "aretrieve_batch" | "retrieve_batch" | "acancel_batch" | "cancel_batch" | "pass_through_endpoint" | "anthropic_messages" | "aanthropic_messages" | "get_assistants" | "aget_assistants" | "create_assistants" | "acreate_assistants" | "delete_assistant" | "adelete_assistant" | "acreate_thread" | "create_thread" | "aget_thread" | "get_thread" | "a_add_message" | "add_message" | "aget_messages" | "get_messages" | "arun_thread" | "run_thread" | "arun_thread_stream" | "run_thread_stream" | "afile_retrieve" | "file_retrieve" | "afile_delete" | "file_delete" | "afile_list" | "file_list" | "acreate_file" | "create_file" | "afile_content" | "file_content" | "create_fine_tuning_job" | "acreate_fine_tuning_job" | "create_video" | "acreate_video" | "video_generation" | "avideo_generation" | "avideo_retrieve" | "video_retrieve" | "avideo_content" | "video_content" | "video_remix" | "avideo_remix" | "video_list" | "avideo_list" | "video_retrieve_job" | "avideo_retrieve_job" | "video_delete" | "avideo_delete" | "video_create_character" | "avideo_create_character" | "video_get_character" | "avideo_get_character" | "video_edit" | "avideo_edit" | "video_extension" | "avideo_extension" | "vector_store_file_create" | "avector_store_file_create" | "vector_store_file_list" | "avector_store_file_list" | "vector_store_file_retrieve" | "avector_store_file_retrieve" | "vector_store_file_content" | "avector_store_file_content" | "vector_store_file_update" | "avector_store_file_update" | "vector_store_file_delete" | "avector_store_file_delete" | "vector_store_create" | "avector_store_create" | "vector_store_search" | "avector_store_search" | "ingest" | "aingest" | "query" | "aquery" | "create_interaction" | "acreate_interaction" | "create_container" | "acreate_container" | "list_containers" | "alist_containers" | "retrieve_container" | "aretrieve_container" | "delete_container" | "adelete_container" | "list_container_files" | "alist_container_files" | "upload_container_file" | "aupload_container_file" | "create_sandbox" | "acreate_sandbox" | "delete_sandbox" | "adelete_sandbox" | "run_code" | "arun_code" | "code_interpreter_tool" | "acode_interpreter_tool" | "acancel_fine_tuning_job" | "cancel_fine_tuning_job" | "alist_fine_tuning_jobs" | "list_fine_tuning_jobs" | "aretrieve_fine_tuning_job" | "retrieve_fine_tuning_job" | "responses" | "aresponses" | "alist_input_items" | "llm_passthrough_route" | "allm_passthrough_route" | "generate_content" | "agenerate_content" | "generate_content_stream" | "agenerate_content_stream" | "ocr" | "aocr" | "call_mcp_tool" | "list_mcp_tools" | "asend_message" | "send_message" | "acreate_skill";
/** CallbackDelete */
CallbackDelete: {
/** Callback Name */