scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-30 06:43:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 249

feat(providers): add Kyma API as an OpenAI-compatible provider (#48756)

Browse source 
Adds Kyma API (kyma/<model>) as a JSON-configured OpenAI-compatible provider. Spend reads the usage.cost Kyma reports, with 83 catalog rows as the fallback price, and image and audio routes refuse before calling Kyma. Carries over #39799 with its author's commits kept
This commit is contained in:
m6atlas 2026-10-29 05:19:21 -07:00 • committed by GitHub
parent 3e3362d461
commit cd42694b48
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 3151 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
litellm/__init__.py
View file

@ -685,6 +685,7 @@ moonshot_models: Set = set()
publicai_models: Set = set()
darkbloom_models: Set = set()
openinfer_models: Final[Set[str]] = set()
kyma_models: Final[set[str]] = set() # mutable-ok: populated when the shared model catalog loads
streamlake_models: Final[Set[str]] = set()
zerogpu_models: Final[Set[str]] = set()
v0_models: Set = set()
@ -981,6 +982,8 @@ def _populate_provider_model_sets(model_cost_map: Mapping[str, Mapping[str, obje
darkbloom_models.add(key)
elif value.get("litellm_provider") == "openinfer":
openinfer_models.add(key)
elif value.get("litellm_provider") == "kyma":
kyma_models.add(key)
elif value.get("litellm_provider") == "streamlake":
streamlake_models.add(key)
elif value.get("litellm_provider") == "zerogpu":
@ -1159,6 +1162,7 @@ model_list = list(
| publicai_models
| darkbloom_models
| openinfer_models
| kyma_models
| streamlake_models
| zerogpu_models
| v0_models
@ -1290,6 +1294,7 @@ def _build_models_by_provider() -> dict:
"publicai": publicai_models,
"darkbloom": darkbloom_models,
"openinfer": openinfer_models,
"kyma": kyma_models,
"streamlake": streamlake_models,
"zerogpu": zerogpu_models,
"v0": v0_models,

58
litellm/cost_calculator.py
View file

@ -22,7 +22,7 @@ from litellm.constants import (
OPENAI_LIVE_TERMINAL_RESPONSE_EVENT_TYPES,
REALTIME_METERED_USAGE_KEY,
)
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.core_helpers import as_str_mapping, billed_usage_cost
from litellm.litellm_core_utils.cost_discount import resolve_cost_discount
from litellm.litellm_core_utils.decisions_pricing import decisions_pricing_fields, is_decisions_call_type
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR
@ -1396,9 +1396,9 @@ def _apply_cost_margin(
def _store_cost_breakdown_in_logging_obj(
litellm_logging_obj: LitellmLoggingObject | None,
prompt_tokens_cost_usd_dollar: float,
completion_tokens_cost_usd_dollar: float,
cost_for_built_in_tools_cost_usd_dollar: float,
prompt_tokens_cost_usd_dollar: float | None,
completion_tokens_cost_usd_dollar: float | None,
cost_for_built_in_tools_cost_usd_dollar: float | None,
total_cost_usd_dollar: float,
additional_costs: dict | None = None,
original_cost: float | None = None,
@ -1505,6 +1505,39 @@ def _response_time_ms_for_cost(
return 0.0
def _adjusted_provider_billed_cost(
reported_cost: float,
custom_llm_provider: str | None,
model: str | None,
region_name: str | None,
litellm_logging_obj: LitellmLoggingObject | None,
service_tier: str | None,
data_residency: str | None,
) -> float:
discounted_cost, discount_percent, discount_amount = _apply_cost_discount(
base_cost=reported_cost, custom_llm_provider=custom_llm_provider, model=model, region_name=region_name
)
adjusted_cost, margin_percent, margin_fixed_amount, margin_total_amount = _apply_cost_margin(
base_cost=discounted_cost, custom_llm_provider=custom_llm_provider
)
_store_cost_breakdown_in_logging_obj(
litellm_logging_obj=litellm_logging_obj,
prompt_tokens_cost_usd_dollar=None,
completion_tokens_cost_usd_dollar=None,
cost_for_built_in_tools_cost_usd_dollar=None,
total_cost_usd_dollar=adjusted_cost,
original_cost=reported_cost,
discount_percent=discount_percent,
discount_amount=discount_amount,
margin_percent=margin_percent,
margin_fixed_amount=margin_fixed_amount,
margin_total_amount=margin_total_amount,
service_tier=service_tier,
data_residency=data_residency,
)
return adjusted_cost
def completion_cost(
completion_response: object | None = None,
model: str | None = None,
@ -1616,6 +1649,9 @@ def completion_cost(
)
explicit_pricing: Final = custom_pricing is True or base_model is not None
caller_set_prices: Final = (
explicit_pricing or custom_cost_per_token is not None or custom_cost_per_second is not None
)
custom_priced_request_model: Final = model if custom_pricing is True else None
configured_cost_per_query: Final = (
_configured_search_cost_per_query(litellm_logging_obj) if custom_pricing else None
@ -1666,6 +1702,7 @@ def completion_cost(
raw_candidates if custom_pricing is True else _priced_candidates_before_unpriced_rules(raw_candidates)
)
priced_response: Final[object] = completion_response
for idx, (candidate_model, candidate_pricing_provider) in enumerate(potential_model_names):
model = candidate_model
try:
@ -1984,6 +2021,19 @@ def completion_cost(
total_time=total_time,
)
model_for_discount = custom_priced_request_model or model
if (
not caller_set_prices
and (reported_cost := billed_usage_cost(priced_response, custom_llm_provider)) is not None
):
return _adjusted_provider_billed_cost(
reported_cost=reported_cost,
custom_llm_provider=custom_llm_provider,
model=model_for_discount,
region_name=region_name if isinstance(region_name, str) else None,
litellm_logging_obj=litellm_logging_obj,
service_tier=service_tier,
data_residency=data_residency,
)
# Calculate cost based on prompt_tokens, completion_tokens
if (
"togethercomputer" in model or "together_ai" in model or custom_llm_provider == "together_ai"

2
litellm/images/main.py
View file

@ -641,6 +641,8 @@ def image_generation(
timeout=timeout,
client=custom_client,
)
elif custom_llm_provider == LlmProviders.KYMA.value:
raise LiteLLMUnknownProvider(model=model, custom_llm_provider=custom_llm_provider)
return model_response
except Exception as e:

2
litellm/interactions/litellm_responses_transformation/handler.py
View file

@ -103,6 +103,7 @@ class LiteLLMResponsesInteractionsHandler:
return LiteLLMResponsesInteractionsConfig.transform_responses_response_to_interactions_response(
responses_response=responses_api_response,
model=model,
custom_llm_provider=custom_llm_provider,
)
async def async_interactions_api_handler(
@ -138,4 +139,5 @@ class LiteLLMResponsesInteractionsHandler:
return LiteLLMResponsesInteractionsConfig.transform_responses_response_to_interactions_response(
responses_response=responses_api_response,
model=model,
custom_llm_provider=responses_request.get("custom_llm_provider"),
)

4
litellm/interactions/litellm_responses_transformation/transformation.py
View file

@ -12,6 +12,7 @@ from typing import Final, cast
from pydantic import BaseModel
from litellm.litellm_core_utils.core_helpers import billed_usage_cost
from litellm.types.interactions import (
InteractionInput,
InteractionsAPIOptionalRequestParams,
@ -190,6 +191,7 @@ class LiteLLMResponsesInteractionsConfig:
def transform_responses_response_to_interactions_response(
responses_response: ResponsesAPIResponse,
model: str | None = None,
custom_llm_provider: str | None = None,
) -> InteractionsAPIResponse:
"""
Transform a Responses API response to an Interactions API response.
@ -266,9 +268,11 @@ class LiteLLMResponsesInteractionsConfig:
# (total_input_tokens, total_output_tokens)
usage: Final[object] = getattr(responses_response, "usage", None)
if usage:
reported_cost: Final = billed_usage_cost(responses_response, custom_llm_provider)
interactions_response_dict["usage"] = {
"total_input_tokens": getattr(usage, "input_tokens", 0),
"total_output_tokens": getattr(usage, "output_tokens", 0),
**({} if reported_cost is None else {"cost": reported_cost}),
}
# Add updated (same as created for now)

11
litellm/litellm_core_utils/core_helpers.py
View file

@ -2,6 +2,7 @@
## Helper utilities
import copy
import logging
import math
import re
from collections.abc import Collection, Iterable, Iterator, Mapping
from types import MappingProxyType
@ -1016,6 +1017,16 @@ def chunk_usage(chunk: object) -> object:
return getattr(chunk, "usage", None)
USAGE_COST_BILLED_PROVIDERS: Final = frozenset({"kyma"})
def billed_usage_cost(response: object, custom_llm_provider: str | None) -> float | None:
if custom_llm_provider not in USAGE_COST_BILLED_PROVIDERS:
return None
cost: Final = provider_reported_usage_cost(_usage_field(response, "usage"), custom_llm_provider)
return cost if cost is not None and math.isfinite(cost) and cost >= 0 else None
_MESSAGES_REPORTED_COST_PROVIDERS: Final = frozenset({"opencode_go"})

15
litellm/litellm_core_utils/litellm_logging.py
View file

@ -1942,10 +1942,10 @@ class Logging(LiteLLMLoggingBaseClass):
def set_cost_breakdown(
self,
input_cost: float,
output_cost: float,
input_cost: float | None,
output_cost: float | None,
total_cost: float,
cost_for_built_in_tools_cost_usd_dollar: float,
cost_for_built_in_tools_cost_usd_dollar: float | None,
additional_costs: dict | None = None,
original_cost: float | None = None,
discount_percent: float | None = None,
@ -1984,14 +1984,17 @@ class Logging(LiteLLMLoggingBaseClass):
self.billed_token_rates = billed_token_rates
self.cost_breakdown = CostBreakdown(
input_cost=input_cost,
output_cost=output_cost,
total_cost=total_cost,
tool_usage_cost=cost_for_built_in_tools_cost_usd_dollar,
service_tier=service_tier,
data_residency=data_residency,
vertex_location=vertex_location,
)
if input_cost is not None:
self.cost_breakdown["input_cost"] = input_cost
if output_cost is not None:
self.cost_breakdown["output_cost"] = output_cost
if cost_for_built_in_tools_cost_usd_dollar is not None:
self.cost_breakdown["tool_usage_cost"] = cost_for_built_in_tools_cost_usd_dollar
if cache_read_cost is not None and cache_read_cost > 0:
self.cost_breakdown["cache_read_cost"] = cache_read_cost
if cache_creation_cost is not None and cache_creation_cost > 0:

8
litellm/litellm_core_utils/streaming_chunk_builder_utils.py
View file

@ -12,7 +12,11 @@ from typing_extensions import ReadOnly, Required
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.core_helpers import chunk_usage, provider_reported_usage_cost
from litellm.litellm_core_utils.core_helpers import (
USAGE_COST_BILLED_PROVIDERS,
chunk_usage,
provider_reported_usage_cost,
)
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hidden_params
from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import BEDROCK_TRACE_FIELDS, object_items, object_mapping
from litellm.litellm_core_utils.prompt_templates.common_utils import (
@ -1352,7 +1356,7 @@ def stream_chunk_builder_text_completion(
return TextCompletionResponse(**response)
_CALCULATOR_PRICED_REPORTED_COST_PROVIDERS: Final = frozenset({LlmProviders.XAI.value})
_CALCULATOR_PRICED_REPORTED_COST_PROVIDERS: Final = frozenset({LlmProviders.XAI.value}) | USAGE_COST_BILLED_PROVIDERS
def _logged_provider(logging_obj: Optional["Logging"]) -> object:

12
litellm/llms/anthropic/messages/adapters/transformation.py
View file

@ -203,7 +203,7 @@ from litellm.types.llms.openai import (
ChatCompletionUserMessage,
ToolMessageContentPart,
)
from litellm.types.utils import Choices, ModelResponse, StreamingChoices, Usage
from litellm.types.utils import Choices, LlmProviders, ModelResponse, StreamingChoices, Usage
from litellm.utils import supports_mid_conversation_system
from .streaming_iterator import AnthropicSSEStream, AnthropicStreamWrapper
@ -362,6 +362,9 @@ class AnthropicAdapter:
_BlockT: Final = TypeVar("_BlockT", bound=Mapping[str, object])
_CLAUDE_OVER_REASONING_EFFORT_PROVIDERS: Final = frozenset({LlmProviders.KYMA.value})
class LiteLLMAnthropicMessagesAdapter:
def __init__(self):
pass
@ -1128,7 +1131,8 @@ class LiteLLMAnthropicMessagesAdapter:
"""Translate Anthropic thinking to either thinking or reasoning_effort.
A Claude-family target keeps ``thinking`` verbatim, since every bridged provider serving one
speaks that param. Carrying its adaptive effort tier alongside takes two different params,
speaks that param, except Kyma, which serves Claude over the OpenAI dialect and takes only
``reasoning_effort``, so its Claude models get the non-Claude mapping. Carrying its adaptive effort tier alongside takes two different params,
because the two are not interchangeable at the provider mapping below.
Bedrock takes ``output_config`` directly, which attaches the tier and leaves ``thinking``
@ -1159,7 +1163,9 @@ class LiteLLMAnthropicMessagesAdapter:
is_bedrock_target: Final = model.startswith(("bedrock/", "converse/", "invoke/")) or self.is_bedrock_arn_model(
model
)
is_claude_target: Final = self.is_anthropic_claude_model(model) or self.is_bedrock_arn_model(model)
is_claude_target: Final = (
self.is_anthropic_claude_model(model) or self.is_bedrock_arn_model(model)
) and custom_llm_provider not in _CLAUDE_OVER_REASONING_EFFORT_PROVIDERS
output_config: Final = anthropic_message_request.get("output_config")
if is_claude_target:

7
litellm/llms/openai_like/providers.json
View file

@ -270,6 +270,13 @@
"/v1/responses"
]
},
"kyma": {
"base_url": "https://kymaapi.com/v1",
"api_key_env": "KYMA_API_KEY",
"provider_info": {
"supports_function_calling": true
}
},
"streamlake": {
"base_url": "https://vanchin.streamlake.ai/api/gateway/v1/endpoints",
"api_key_env": "STREAMLAKE_API_KEY",

1248
litellm/model_prices_and_context_window_backup.json
View file

File diff suppressed because it is too large Load diff

26
litellm/provider_endpoints_support_backup.json
View file

@ -5327,6 +5327,32 @@
"generateContent": false
}
},
"kyma": {
"display_name": "Kyma API (`kyma`)",
"url": "https://docs.litellm.ai/docs/providers/kyma",
"endpoints": {
"chat_completions": true,
"responses": true,
"text_completion": true,
"embeddings": false,
"image_generations": false,
"image_edits": false,
"audio_transcriptions": false,
"audio_speech": false,
"rerank": false,
"ocr": false,
"search": false,
"moderations": false,
"realtime": false,
"video_generations": false,
"batches": false,
"messages": true,
"interactions": true,
"systemone": false,
"a2a": true,
"generateContent": true
}
},
"bigmodel": {
"display_name": "BigModel (Zhipu AI) (`bigmodel`)",
"url": "https://docs.bigmodel.cn/cn/guide/develop/responses/introduction",

28
litellm/proxy/public_endpoints/provider_create_fields.json
View file

@ -2023,6 +2023,34 @@
],
"default_model_placeholder": "jina_ai/"
},
{
"provider": "KYMA",
"provider_display_name": "Kyma API",
"litellm_provider": "kyma",
"credential_fields": [
{
"key": "api_base",
"label": "API Base",
"placeholder": "https://kymaapi.com/v1",
"tooltip": null,
"required": false,
"field_type": "text",
"options": null,
"default_value": null
},
{
"key": "api_key",
"label": "API Key",
"placeholder": null,
"tooltip": null,
"required": true,
"field_type": "password",
"options": null,
"default_value": null
}
],
"default_model_placeholder": "kyma/qwen3.7-flash"
},
{
"provider": "Laya",
"provider_display_name": "Laya",

1
litellm/types/utils.py
View file

@ -4600,6 +4600,7 @@ class LlmProviders(str, Enum):
CURSOR = "cursor"
BEDROCK_MANTLE = "bedrock_mantle"
GDC = "gdc"
KYMA = "kyma"
# Create a set of all provider values for quick lookup

1248
model_prices_and_context_window.json
View file

File diff suppressed because it is too large Load diff

26
provider_endpoints_support.json
View file

@ -5327,6 +5327,32 @@
"generateContent": false
}
},
"kyma": {
"display_name": "Kyma API (`kyma`)",
"url": "https://docs.litellm.ai/docs/providers/kyma",
"endpoints": {
"chat_completions": true,
"responses": true,
"text_completion": true,
"embeddings": false,
"image_generations": false,
"image_edits": false,
"audio_transcriptions": false,
"audio_speech": false,
"rerank": false,
"ocr": false,
"search": false,
"moderations": false,
"realtime": false,
"video_generations": false,
"batches": false,
"messages": true,
"interactions": true,
"systemone": false,
"a2a": true,
"generateContent": true
}
},
"bigmodel": {
"display_name": "BigModel (Zhipu AI) (`bigmodel`)",
"url": "https://docs.bigmodel.cn/cn/guide/develop/responses/introduction",

465
tests/unit/llms/openai_like/test_kyma_provider.py Normal file
View file

@ -0,0 +1,465 @@
import json
import re
from collections.abc import Callable
from datetime import datetime
from itertools import chain
from pathlib import Path
from typing import Final
import pytest
import respx
import litellm
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.proxy.auth.model_checks import get_known_models_from_wildcard
from litellm.proxy.common_request_processing import cost_headers
from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices, Usage
_REPO_ROOT: Final = Path(__file__).resolve().parents[4]
_PRIMARY_PRICES: Final = json.loads((_REPO_ROOT / "model_prices_and_context_window.json").read_text())
_BACKUP_PRICES: Final = json.loads((_REPO_ROOT / "litellm" / "model_prices_and_context_window_backup.json").read_text())
_CATALOG_MODELS: Final = tuple(
key for key, row in _PRIMARY_PRICES.items() if isinstance(row, dict) and row.get("litellm_provider") == "kyma"
)
_TIER_KEY: Final = re.compile(r"^input_cost_per_token_above_(\d+)k_tokens$")
def _long_prompt_tiers(model: str) -> tuple[tuple[str, int], ...]:
matches: Final = map(_TIER_KEY.match, _PRIMARY_PRICES[model])
return tuple((model, int(match.group(1))) for match in matches if match is not None)
_LONG_PROMPT_TIERS: Final = tuple(chain.from_iterable(map(_long_prompt_tiers, _CATALOG_MODELS)))
_CHAT_URL: Final = "https://kymaapi.com/v1/chat/completions"
_UNLISTED_MODEL: Final = "kyma/model-added-after-this-release"
def _kyma_chat_body(usage: dict[str, object]) -> dict[str, object]:
return {
"id": "chatcmpl_kyma",
"object": "chat.completion",
"created": 1_792_000_000,
"model": "sonar",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "pong"}, "finish_reason": "stop"}],
"usage": usage,
}
def test_kyma_provider_resolution(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("KYMA_API_KEY", "kyma-test-key")
model, provider, api_key, api_base = get_llm_provider(
model="kyma/qwen3.7-flash",
custom_llm_provider=None,
api_base=None,
api_key=None,
)
assert model == "qwen3.7-flash"
assert provider == "kyma"
assert api_key == "kyma-test-key"
assert api_base == "https://kymaapi.com/v1"
def test_kyma_provider_keeps_explicit_credentials(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("KYMA_API_KEY", "kyma-env-key")
_, provider, api_key, api_base = get_llm_provider(
model="kyma/qwen3.7-flash",
custom_llm_provider=None,
api_base="https://kyma.internal.example/v1",
api_key="kyma-explicit-key",
)
assert provider == "kyma"
assert api_key == "kyma-explicit-key"
assert api_base == "https://kyma.internal.example/v1"
def test_bare_model_on_kyma_base_url_keeps_openai_routing(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("OPENAI_API_KEY", "openai-env-key")
model, provider, api_key, api_base = get_llm_provider(
model="gpt-6-luna",
custom_llm_provider=None,
api_base="https://kymaapi.com/v1",
api_key=None,
)
assert (model, provider, api_base) == ("gpt-6-luna", "openai", "https://kymaapi.com/v1")
assert api_key is None
def test_kyma_is_available_in_add_model_form():
fields_path: Final = Path(litellm.__file__).parent / "proxy" / "public_endpoints" / "provider_create_fields.json"
providers: Final = json.loads(fields_path.read_text())
kyma: Final = next(provider for provider in providers if provider["litellm_provider"] == "kyma")
assert kyma["default_model_placeholder"] in _CATALOG_MODELS
assert {field["key"]: field["required"] for field in kyma["credential_fields"]} == {
"api_base": False,
"api_key": True,
}
def test_kyma_chat_completion_request():
with respx.mock() as upstream:
route: Final = upstream.post(_CHAT_URL).respond(
200,
json={
"id": "chatcmpl_kyma",
"object": "chat.completion",
"created": 1_792_000_000,
"model": "qwen3.7-flash",
"choices": [
{"index": 0, "message": {"role": "assistant", "content": "Hello from Kyma"}, "finish_reason": "stop"}
],
"usage": {"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7},
},
)
response: Final = litellm.completion(
model="kyma/qwen3.7-flash",
messages=[{"role": "user", "content": "Say hello"}],
api_key="kyma-test-key",
)
request: Final = route.calls.last.request
body: Final = json.loads(request.content)
assert route.call_count == 1
assert str(request.url) == _CHAT_URL
assert request.headers["authorization"] == "Bearer kyma-test-key"
assert body["model"] == "qwen3.7-flash"
assert body["messages"] == [{"role": "user", "content": "Say hello"}]
assert response.choices[0].message.content == "Hello from Kyma"
@pytest.mark.parametrize("model", _CATALOG_MODELS)
def test_catalog_row_matches_its_backup_and_fits_its_context(model: str):
row: Final = _PRIMARY_PRICES[model]
assert row == _BACKUP_PRICES[model]
assert row["mode"] == "chat"
assert 0 < row["max_output_tokens"] <= row["max_input_tokens"]
assert row["input_cost_per_token"] > 0
assert row["output_cost_per_token"] > 0
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize("model", _CATALOG_MODELS)
def test_chat_response_is_priced_at_its_catalog_row(model: str):
row: Final = _PRIMARY_PRICES[model]
response: Final = litellm.ModelResponse(
model=model.removeprefix("kyma/"),
usage=litellm.Usage(prompt_tokens=1000, completion_tokens=200, total_tokens=1200),
)
cost: Final = litellm.completion_cost(completion_response=response, model=model)
assert cost == pytest.approx(1000 * row["input_cost_per_token"] + 200 * row["output_cost_per_token"])
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(("model", "threshold_k"), _LONG_PROMPT_TIERS)
def test_long_prompt_is_priced_at_its_tier(model: str, threshold_k: int):
row: Final = _PRIMARY_PRICES[model]
prompt_tokens: Final = threshold_k * 1000 + 1000
response: Final = litellm.ModelResponse(
model=model.removeprefix("kyma/"),
usage=litellm.Usage(prompt_tokens=prompt_tokens, completion_tokens=200, total_tokens=prompt_tokens + 200),
)
cost: Final = litellm.completion_cost(completion_response=response, model=model)
assert cost == pytest.approx(
prompt_tokens * row[f"input_cost_per_token_above_{threshold_k}k_tokens"]
+ 200 * row[f"output_cost_per_token_above_{threshold_k}k_tokens"]
)
@pytest.mark.parametrize("model", ("kyma/sonar", _UNLISTED_MODEL))
def test_kyma_response_is_billed_at_the_cost_kyma_reports(model: str):
reported_cost: Final = 0.00721
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": reported_cost})
)
response: Final = litellm.completion(model=model, messages=[{"role": "user", "content": "ping"}], api_key="k")
assert response._hidden_params["response_cost"] == pytest.approx(reported_cost)
def test_deployment_pricing_overrides_the_cost_kyma_reports():
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": 0.00721})
)
response: Final = litellm.completion(
model="kyma/sonar",
messages=[{"role": "user", "content": "ping"}],
api_key="k",
input_cost_per_token=0.001,
output_cost_per_token=0.002,
)
assert response._hidden_params["response_cost"] == pytest.approx(4 * 0.001 + 3 * 0.002)
def _catalog_price(model: str, prompt_tokens: int, completion_tokens: int) -> float:
row: Final = _PRIMARY_PRICES[model]
return prompt_tokens * row["input_cost_per_token"] + completion_tokens * row["output_cost_per_token"]
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize("reported_cost", ("-0.5", "1e999"))
def test_unusable_cost_from_kyma_falls_back_to_the_catalog_price(reported_cost: str):
usage: Final = {"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": "REPORTED_COST"}
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(
200,
content=json.dumps(_kyma_chat_body(usage)).replace('"REPORTED_COST"', reported_cost),
headers={"content-type": "application/json"},
)
response: Final = litellm.completion(model="kyma/sonar", messages=[{"role": "user", "content": "ping"}], api_key="k")
assert response._hidden_params["response_cost"] == pytest.approx(_catalog_price("kyma/sonar", 4, 3))
@pytest.mark.usefixtures("local_model_cost_map")
def test_deployment_base_model_overrides_the_cost_kyma_reports():
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": 0.00721})
)
response: Final = litellm.completion(
model="kyma/sonar",
messages=[{"role": "user", "content": "ping"}],
api_key="k",
base_model="kyma/qwen3.7-flash",
)
assert response._hidden_params["response_cost"] == pytest.approx(_catalog_price("kyma/qwen3.7-flash", 4, 3))
def test_interaction_on_kyma_is_billed_at_the_cost_kyma_reports():
reported_cost: Final = 0.00721
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": reported_cost})
)
interaction: Final = litellm.interactions.create(model="kyma/qwen3.7-flash", input="ping", api_key="k")
assert interaction._hidden_params["response_cost"] == pytest.approx(reported_cost)
@pytest.mark.parametrize("model", ("openrouter/openai/gpt-4o-mini", "deepseek/deepseek-chat"))
def test_interaction_on_another_provider_keeps_its_upstream_cost_out_of_the_body(model: str):
chat_usage: Final = {"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": 0.00721}
responses_body: Final = {
"id": "resp_upstream",
"object": "response",
"created_at": 1_792_000_000,
"status": "completed",
"model": "x",
"output": [
{
"type": "message",
"id": "msg_upstream",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "pong", "annotations": []}],
}
],
"usage": {"input_tokens": 4, "output_tokens": 3, "total_tokens": 7, "cost": 0.00721},
}
with respx.mock(assert_all_called=False) as upstream:
upstream.route(path__regex=r".*/chat/completions$").respond(200, json=_kyma_chat_body(chat_usage))
upstream.route(path__regex=r".*/responses$").respond(200, json=responses_body)
interaction: Final = litellm.interactions.create(model=model, input="ping", api_key="k")
assert "cost" not in interaction.model_dump(exclude_none=True)["usage"]
def test_sdk_completion_cost_prices_kyma_at_the_cost_kyma_reports():
reported_cost: Final = 0.00721
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": reported_cost})
)
response: Final = litellm.completion(model="kyma/sonar", messages=[{"role": "user", "content": "ping"}], api_key="k")
assert litellm.completion_cost(completion_response=response) == pytest.approx(reported_cost)
def _kyma_sdk_response(usage: dict[str, float]) -> litellm.ModelResponse:
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(200, json=_kyma_chat_body(usage))
response: Final = litellm.completion(model="kyma/sonar", messages=[{"role": "user", "content": "ping"}], api_key="k")
assert isinstance(response, litellm.ModelResponse)
return response
_KYMA_TOKEN_USAGE: Final = {"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7}
_KYMA_REPORTED_COST: Final = 0.00721
def test_sdk_completion_cost_uses_the_callers_token_prices_over_the_cost_kyma_reports():
caller_prices: Final = {"input_cost_per_token": 0.5, "output_cost_per_token": 2.0}
with_reported_cost: Final = _kyma_sdk_response({**_KYMA_TOKEN_USAGE, "cost": _KYMA_REPORTED_COST})
without_reported_cost: Final = _kyma_sdk_response(_KYMA_TOKEN_USAGE)
assert litellm.completion_cost(
completion_response=with_reported_cost, custom_cost_per_token=caller_prices
) == pytest.approx(litellm.completion_cost(completion_response=without_reported_cost, custom_cost_per_token=caller_prices))
assert litellm.completion_cost(
completion_response=with_reported_cost, custom_cost_per_token=caller_prices
) != pytest.approx(_KYMA_REPORTED_COST)
def test_sdk_completion_cost_uses_the_callers_per_second_price_over_the_cost_kyma_reports():
response: Final = _kyma_sdk_response({**_KYMA_TOKEN_USAGE, "cost": _KYMA_REPORTED_COST})
assert litellm.completion_cost(completion_response=response, custom_cost_per_second=50.0) == pytest.approx(
50.0 * response._response_ms / 1000
)
def _with_kyma_discount_and_margin(monkeypatch: pytest.MonkeyPatch) -> float:
monkeypatch.setattr(litellm, "cost_discount_config", {"kyma": 0.5})
monkeypatch.setattr(litellm, "cost_margin_config", {"kyma": 0.1})
return 0.5 * 1.1
def test_cost_kyma_reports_takes_the_configured_discount_and_margin(monkeypatch: pytest.MonkeyPatch):
adjustment: Final = _with_kyma_discount_and_margin(monkeypatch)
with respx.mock() as upstream:
upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7, "cost": 0.008})
)
response: Final = litellm.completion(model="kyma/sonar", messages=[{"role": "user", "content": "ping"}], api_key="k")
assert response._hidden_params["response_cost"] == pytest.approx(0.008 * adjustment)
def _kyma_stream_chunk(content: str, finish_reason: str | None = None) -> ModelResponseStream:
return ModelResponseStream(
id="chatcmpl_kyma",
created=1_792_000_000,
model="sonar",
object="chat.completion.chunk",
choices=[StreamingChoices(finish_reason=finish_reason, index=0, delta=Delta(content=content, role="assistant"))],
)
def test_streamed_cost_kyma_reports_takes_the_configured_discount_and_margin(monkeypatch: pytest.MonkeyPatch):
adjustment: Final = _with_kyma_discount_and_margin(monkeypatch)
usage_chunk: Final = _kyma_stream_chunk("")
usage_chunk.usage = Usage(prompt_tokens=4, completion_tokens=3, total_tokens=7, cost=0.008)
logging_obj: Final = LiteLLMLogging(
model="sonar",
messages=[{"role": "user", "content": "ping"}],
stream=True,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="kyma-call",
function_id="kyma-function",
)
logging_obj.update_environment_variables(
model="sonar",
user=None,
optional_params={},
litellm_params={"custom_llm_provider": "kyma"},
custom_llm_provider="kyma",
)
response: Final = litellm.stream_chunk_builder(
chunks=[_kyma_stream_chunk("pong", finish_reason="stop"), usage_chunk],
messages=[{"role": "user", "content": "ping"}],
logging_obj=logging_obj,
)
assert logging_obj.response_cost_calculator(result=response) == pytest.approx(0.008 * adjustment)
def test_cost_headers_show_the_discount_and_margin_taken_from_the_cost_kyma_reports(monkeypatch: pytest.MonkeyPatch):
_with_kyma_discount_and_margin(monkeypatch)
logging_obj: Final = LiteLLMLogging(
model="sonar",
messages=[{"role": "user", "content": "ping"}],
stream=False,
call_type="completion",
start_time=datetime.now(),
litellm_call_id="kyma-call",
function_id="kyma-function",
)
logging_obj.update_environment_variables(
model="sonar",
user=None,
optional_params={},
litellm_params={"custom_llm_provider": "kyma"},
custom_llm_provider="kyma",
)
response: Final = litellm.ModelResponse(
model="sonar", usage=Usage(prompt_tokens=4, completion_tokens=3, total_tokens=7, cost=0.008)
)
headers: Final = cost_headers(logging_obj.response_cost_calculator(result=response), logging_obj)
assert float(headers["x-litellm-response-cost-original"] or "nan") == pytest.approx(0.008)
assert float(headers["x-litellm-response-cost-discount-amount"] or "nan") == pytest.approx(0.004)
assert float(headers["x-litellm-response-cost-margin-amount"] or "nan") == pytest.approx(0.0004)
assert float(headers["x-litellm-response-cost"] or "nan") == pytest.approx(0.0044)
assert headers["x-litellm-response-cost-input"] is None
assert headers["x-litellm-response-cost-output"] is None
@pytest.mark.parametrize("thinking", ({"type": "enabled", "budget_tokens": 1024}, {"type": "adaptive"}))
def test_messages_thinking_on_a_kyma_claude_model_is_sent_as_reasoning_effort(thinking: dict[str, object]):
with respx.mock() as upstream:
route: Final = upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7})
)
litellm.anthropic.messages.create(
model="kyma/claude-haiku-4-5",
max_tokens=2048,
messages=[{"role": "user", "content": "ping"}],
thinking=thinking,
api_key="k",
)
sent: Final = json.loads(route.calls.last.request.content)
assert "thinking" not in sent
assert sent["reasoning_effort"] in ("low", "medium", "high")
def test_kyma_model_missing_from_the_catalog_keeps_its_tools():
tools: Final = [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object"}}}]
with respx.mock() as upstream:
route: Final = upstream.post(_CHAT_URL).respond(
200, json=_kyma_chat_body({"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7})
)
litellm.completion(model=_UNLISTED_MODEL, messages=[{"role": "user", "content": "ping"}], api_key="k", tools=tools)
assert json.loads(route.calls.last.request.content)["tools"] == tools
def test_kyma_wildcard_lists_every_catalog_model():
assert sorted(get_known_models_from_wildcard("kyma/*")) == sorted(_CATALOG_MODELS)
def _generate_image() -> object:
return litellm.image_generation(model="kyma/flux-2", prompt="a cat", api_key="k")
def _transcribe() -> object:
return litellm.transcription(model="kyma/whisper-1", file=("a.wav", b"RIFF0000WAVE", "audio/wav"), api_key="k")
@pytest.mark.parametrize("call", (_generate_image, _transcribe))
def test_unpriced_kyma_media_calls_are_refused_without_calling_kyma(call: Callable[[], object]):
with respx.mock(assert_all_called=False) as upstream:
kyma: Final = upstream.route(host="kymaapi.com").respond(202, json={"id": "job_kyma"})
with pytest.raises(litellm.BadRequestError):
call()
assert kyma.call_count == 0