Merge pull request #41607 from BerriAI/litellm_typesafe_passthrough
Some checks are pending
CI Coverage / assert-ci-coverage (push) Waiting to run
CodeQL / Analyze (actions) (push) Waiting to run
CodeQL / Analyze (javascript-typescript) (push) Waiting to run
CodeQL / Analyze (python) (push) Waiting to run
CodSpeed Benchmarks / benchmarks (push) Waiting to run
Publish basedpyright base counts / publish (push) Waiting to run
Postgres Tests / schema-migration (push) Waiting to run
LiteLLM Rust / rust-test (push) Waiting to run
Unit Tests: Documentation Validation / documentation (push) Waiting to run
Unit Tests: Proxy DB Operations / assert-shard-coverage (push) Waiting to run
Unit Tests: Proxy DB Operations / endpoints-and-responses (push) Blocked by required conditions
Unit Tests / caching-local (push) Waiting to run
Unit Tests / core-utils (push) Waiting to run
Unit Tests / enterprise-package (push) Waiting to run
Unit Tests / enterprise-routing (push) Waiting to run
Unit Tests / integrations (push) Waiting to run
Helm unit test / unit-test (push) Waiting to run
Scorecard supply-chain security / Scorecard analysis (push) Waiting to run
Code Quality Checks / code-quality (push) Waiting to run
Code Quality Checks / python-310-import-smoke (push) Waiting to run
UI Unit Tests / ui-unit-tests (push) Waiting to run
Postgres Tests / proxy-security (push) Waiting to run
Postgres Tests / proxy-behavior (push) Waiting to run
LiteLLM Rust / rust-wheel (push) Waiting to run
LiteLLM Rust / rust-lint (push) Waiting to run
Unit Tests: Proxy DB Operations / auth-checks (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / budgets (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / custom-logging (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / db-and-spend (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / guardrails-hooks (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / jwt-and-keys (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / key-generation (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / logging-misc (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-runtime (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-server-core (push) Blocked by required conditions
Unit Tests: Proxy DB Operations / proxy-utils (push) Blocked by required conditions
Unit Tests / Vertex AI (push) Waiting to run
Unit Tests / All Other Providers (push) Waiting to run
Unit Tests / misc (push) Waiting to run
Unit Tests / proxy-auth (push) Waiting to run
Unit Tests / proxy-endpoints (push) Waiting to run
Unit Tests / proxy-extras (push) Waiting to run
Unit Tests / proxy-infra (push) Waiting to run
Unit Tests / proxy-server (push) Waiting to run
Unit Tests / responses-caching-types (push) Waiting to run
GitHub Actions Security Analysis / zizmor (push) Waiting to run

feat(proxy): add TypeSafe AI Jev evaluate passthrough with registry-priced spend tracking
This commit is contained in:
Mateo Wang 2026-09-17 16:37:33 -07:00 • committed by GitHub
commit deb9d8aedd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 601 additions and 0 deletions

View file

@ -96,6 +96,7 @@ GATEWAY_PATH_PREFIXES: tuple[str, ...] = (
"/langfuse/", "/langfuse/",
"/vllm/", "/vllm/",
"/mistral/", "/mistral/",
"/typesafe/",
"/nvidia_nim/", "/nvidia_nim/",
"/groq/", "/groq/",
"/voyage/", "/voyage/",

View file

@ -69184,6 +69184,27 @@
"supports_reasoning": true, "supports_reasoning": true,
"supports_vision": true "supports_vision": true
}, },
"typesafe/jev-1.13.0": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-latest": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-preview": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"wandb/zai-org/GLM-5.3-Flash": { "wandb/zai-org/GLM-5.3-Flash": {
"cache_read_input_token_cost": 5e-08, "cache_read_input_token_cost": 5e-08,
"input_cost_per_token": 1.5e-07, "input_cost_per_token": 1.5e-07,

View file

@ -208,6 +208,7 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = (
"/nvidia_nim/", "/nvidia_nim/",
"/openai/", "/openai/",
"/openai_passthrough/", "/openai_passthrough/",
"/typesafe/",
"/vertex-ai/", "/vertex-ai/",
"/vertex_ai/", "/vertex_ai/",
"/vllm/", "/vllm/",

View file

@ -20373,6 +20373,96 @@
] ]
} }
}, },
"/typesafe/{endpoint}": {
"get": {
"description": "[Docs](https://docs.litellm.ai/docs/pass_through/typesafe)",
"operationId": "typesafe_proxy_route_typesafe__endpoint__get",
"parameters": [
{
"in": "path",
"name": "endpoint",
"required": true,
"schema": {
"title": "Endpoint",
"type": "string"
}
}
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {}
}
},
"description": "Successful Response"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
"description": "Validation Error"
}
},
"security": [
{
"APIKeyHeader": []
}
],
"summary": "Typesafe Proxy Route",
"tags": [
"llm_passthrough"
]
},
"post": {
"description": "[Docs](https://docs.litellm.ai/docs/pass_through/typesafe)",
"operationId": "typesafe_proxy_route_typesafe__endpoint__post",
"parameters": [
{
"in": "path",
"name": "endpoint",
"required": true,
"schema": {
"title": "Endpoint",
"type": "string"
}
}
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {}
}
},
"description": "Successful Response"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
"description": "Validation Error"
}
},
"security": [
{
"APIKeyHeader": []
}
],
"summary": "Typesafe Proxy Route",
"tags": [
"llm_passthrough"
]
}
},
"/vertex_ai/discovery/{endpoint}": { "/vertex_ai/discovery/{endpoint}": {
"delete": { "delete": {
"description": "Call any vertex discovery endpoint using the proxy.\n\nJust use `{PROXY_BASE_URL}/vertex_ai/discovery/{endpoint:path}`\n\nTarget url: `https://discoveryengine.googleapis.com`", "description": "Call any vertex discovery endpoint using the proxy.\n\nJust use `{PROXY_BASE_URL}/vertex_ai/discovery/{endpoint:path}`\n\nTarget url: `https://discoveryengine.googleapis.com`",

View file

@ -483,6 +483,7 @@ class LiteLLMRoutes(enum.Enum):
"/eu.assemblyai", "/eu.assemblyai",
"/vllm", "/vllm",
"/mistral", "/mistral",
"/typesafe",
"/milvus", "/milvus",
"/gigachat", "/gigachat",
"/watsonx", "/watsonx",

View file

@ -525,6 +525,42 @@ async def mistral_proxy_route(
return received_value return received_value
@router.api_route(
"/typesafe/{endpoint:path}",
methods=["GET", "POST"], # mutable-ok: FastAPI route metadata requires a list
tags=["TypeSafe AI Pass-through", "pass-through"], # mutable-ok: FastAPI route metadata requires a list
)
async def typesafe_proxy_route(
endpoint: str,
request: Request,
fastapi_response: Response,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
):
"""[Docs](https://docs.litellm.ai/docs/pass_through/typesafe)"""
base_target_url: Final = get_secret_str("TYPESAFE_API_BASE") or "https://api.typesafe.ai"
encoded_endpoint: Final = httpx.URL(endpoint).path
normalized_endpoint: Final = encoded_endpoint if encoded_endpoint.startswith("/") else f"/{encoded_endpoint}"
base_url: Final = httpx.URL(base_target_url)
updated_url: Final = base_url.copy_with(
path=HttpPassThroughEndpointHelpers.join_base_and_endpoint_path(base_url, normalized_endpoint),
)
typesafe_api_key: Final = passthrough_endpoint_router.get_credentials(
custom_llm_provider="typesafe",
region_name=None,
)
endpoint_func: Final = create_pass_through_route(
endpoint=endpoint,
target=str(updated_url),
custom_headers={ # mutable-ok: pass-through request headers require a mutable mapping
"Authorization": f"Bearer {typesafe_api_key}",
"Content-Type": "application/json",
},
custom_llm_provider="typesafe",
is_streaming_request=False,
)
return await endpoint_func(request, fastapi_response, user_api_key_dict)
@router.api_route( @router.api_route(
"/milvus/{endpoint:path}", "/milvus/{endpoint:path}",
methods=["GET", "POST", "PUT", "DELETE", "PATCH"], methods=["GET", "POST", "PUT", "DELETE", "PATCH"],

View file

@ -0,0 +1,117 @@
from collections.abc import Mapping
from datetime import datetime
from typing import Final
import httpx
from pydantic import BaseModel, TypeAdapter, ValidationError
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import (
get_standard_logging_object_payload, # pyright: ignore[reportUnknownVariableType] # legacy helper has an untyped signature
)
from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
from litellm.types.utils import ModelResponse, StandardPassThroughResponseObject, Usage
class _TypeSafeUsage(BaseModel):
input_tokens: int = 0
output_tokens: int = 0
class _TypeSafeResponse(BaseModel):
model: str | None = None
usage: _TypeSafeUsage | None = None
class _RegistryPricing(BaseModel):
input_cost_per_token: float = 0.0
output_cost_per_token: float = 0.0
_TYPESAFE_RESPONSE_ADAPTER: Final = TypeAdapter(_TypeSafeResponse)
_REGISTRY_PRICING_ADAPTER: Final = TypeAdapter(_RegistryPricing)
def _parse_typesafe_response(response_body: Mapping[str, object]) -> _TypeSafeResponse:
try:
return _TYPESAFE_RESPONSE_ADAPTER.validate_python(response_body)
except ValidationError:
return _TypeSafeResponse()
def _pricing_for(model_keys: tuple[str, ...]) -> _RegistryPricing:
for model_key in model_keys:
if model_key not in litellm.model_cost: # pyright: ignore[reportUnknownMemberType] # registry is dynamically typed
continue
try:
return _REGISTRY_PRICING_ADAPTER.validate_python(
litellm.model_cost[model_key] # pyright: ignore[reportUnknownMemberType] # registry is dynamically typed
)
except ValidationError:
continue
return _RegistryPricing()
class TypeSafePassthroughLoggingHandler:
@staticmethod
def typesafe_passthrough_handler(
httpx_response: httpx.Response,
response_body: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
url_route: str,
result: str,
start_time: datetime,
end_time: datetime,
cache_hit: bool,
request_body: Mapping[str, object],
**kwargs: object,
) -> PassThroughEndpointLoggingTypedDict:
response: Final = _parse_typesafe_response(response_body)
response_model: Final = response.model
request_model_value: Final = request_body.get("model")
request_model: Final = request_model_value if isinstance(request_model_value, str) else None
logged_model: Final = response_model or request_model or "unknown"
model_name: Final = f"typesafe/{logged_model}"
usage: Final = response.usage or _TypeSafeUsage()
input_tokens: Final = usage.input_tokens
output_tokens: Final = usage.output_tokens
candidate_model_keys: Final = tuple(
f"typesafe/{model}" for model in (response_model, request_model) if model is not None
)
pricing: Final = _pricing_for(candidate_model_keys)
response_cost: Final = (
input_tokens * pricing.input_cost_per_token + output_tokens * pricing.output_cost_per_token
)
usage_object: Final = Usage(
prompt_tokens=input_tokens,
completion_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
)
updated_kwargs: Final = { # mutable-ok: pass-through logging contract requires mutable kwargs
**kwargs,
"model": model_name,
"custom_llm_provider": "typesafe",
"response_cost": response_cost,
"combined_usage_object": usage_object,
}
logging_obj.model_call_details.update(
model=model_name,
custom_llm_provider="typesafe",
response_cost=response_cost,
)
standard_logging_object: Final = get_standard_logging_object_payload(
kwargs=updated_kwargs,
init_response_obj=ModelResponse(model=model_name, usage=usage_object),
start_time=start_time,
end_time=end_time,
logging_obj=logging_obj,
status="success",
)
return { # mutable-ok: pass-through logging contract requires mutable result
"result": StandardPassThroughResponseObject(response=result),
"kwargs": { # mutable-ok: pass-through logging contract requires mutable kwargs
**updated_kwargs,
"standard_logging_object": standard_logging_object,
},
}

View file

@ -1,5 +1,6 @@
import json import json
from datetime import datetime from datetime import datetime
from types import MappingProxyType
from typing import Any, Final from typing import Any, Final
from urllib.parse import urlparse from urllib.parse import urlparse
@ -256,6 +257,25 @@ class PassThroughEndpointLogging:
) )
standard_logging_response_object = comprehend_medical_handler_result["result"] # rebind-ok: elif-chain standard_logging_response_object = comprehend_medical_handler_result["result"] # rebind-ok: elif-chain
kwargs = comprehend_medical_handler_result["kwargs"] # rebind-ok: elif-chain contract kwargs = comprehend_medical_handler_result["kwargs"] # rebind-ok: elif-chain contract
elif self.is_typesafe_route(custom_llm_provider):
from .llm_provider_handlers.typesafe_passthrough_logging_handler import (
TypeSafePassthroughLoggingHandler,
)
typesafe_handler_result: Final = TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=httpx_response,
response_body=response_body if isinstance(response_body, dict) else MappingProxyType({}),
logging_obj=logging_obj,
url_route=url_route,
result=result,
start_time=start_time,
end_time=end_time,
cache_hit=cache_hit,
request_body=request_body,
**kwargs,
)
standard_logging_response_object = typesafe_handler_result["result"]
kwargs = typesafe_handler_result["kwargs"]
elif self.is_vertex_ai_live_route(url_route): elif self.is_vertex_ai_live_route(url_route):
from .llm_provider_handlers.vertex_ai_live_passthrough_logging_handler import ( from .llm_provider_handlers.vertex_ai_live_passthrough_logging_handler import (
VertexAILivePassthroughLoggingHandler, VertexAILivePassthroughLoggingHandler,
@ -389,6 +409,9 @@ class PassThroughEndpointLogging:
def is_comprehend_medical_route(self, custom_llm_provider: str | None) -> bool: def is_comprehend_medical_route(self, custom_llm_provider: str | None) -> bool:
return custom_llm_provider == "comprehendmedical" return custom_llm_provider == "comprehendmedical"
def is_typesafe_route(self, custom_llm_provider: str | None) -> bool:
return custom_llm_provider == "typesafe"
def is_langfuse_route(self, url_route: str): def is_langfuse_route(self, url_route: str):
parsed_url: Final = urlparse(url_route) parsed_url: Final = urlparse(url_route)
for route in self.TRACKED_LANGFUSE_ROUTES: for route in self.TRACKED_LANGFUSE_ROUTES:

View file

@ -348,6 +348,7 @@ class ModelInfoBase(ProviderSpecificModelInfo, total=False):
"audio_transcription", "audio_transcription",
"audio_speech", "audio_speech",
"responses", "responses",
"evaluation",
"ocr", "ocr",
"realtime", "realtime",
] ]

View file

@ -69184,6 +69184,27 @@
"supports_reasoning": true, "supports_reasoning": true,
"supports_vision": true "supports_vision": true
}, },
"typesafe/jev-1.13.0": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-latest": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"typesafe/jev-preview": {
"input_cost_per_token": 4.2e-08,
"litellm_provider": "typesafe",
"mode": "evaluation",
"output_cost_per_token": 0.0,
"source": "https://docs.typesafe.ai/models"
},
"wandb/zai-org/GLM-5.3-Flash": { "wandb/zai-org/GLM-5.3-Flash": {
"cache_read_input_token_cost": 5e-08, "cache_read_input_token_cost": 5e-08,
"input_cost_per_token": 1.5e-07, "input_cost_per_token": 1.5e-07,

View file

@ -427,6 +427,7 @@
"chat", "chat",
"completion", "completion",
"embedding", "embedding",
"evaluation",
"guardrail", "guardrail",
"image_edit", "image_edit",
"image_generation", "image_generation",

View file

@ -0,0 +1,134 @@
from datetime import datetime
from unittest.mock import MagicMock
import httpx
import pytest
import litellm
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.typesafe_passthrough_logging_handler import (
TypeSafePassthroughLoggingHandler,
)
from litellm.proxy.pass_through_endpoints.success_handler import PassThroughEndpointLogging
@pytest.fixture(autouse=True)
def local_model_cost_map(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
def _response() -> httpx.Response:
return httpx.Response(
200,
request=httpx.Request("POST", "https://api.typesafe.ai/v1/systemone"),
json={"model": "jev-1.13.0"},
)
def _logging_obj() -> MagicMock:
logging_obj = MagicMock()
logging_obj.model_call_details = {}
return logging_obj
def _handler_result(response_body: dict, request_body: dict) -> dict:
return TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=_response(),
response_body=response_body,
logging_obj=_logging_obj(),
url_route="https://api.typesafe.ai/v1/systemone",
result='{"answers": {}}',
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body=request_body,
)
def test_uses_registry_pricing_and_standard_usage():
logging_obj = _logging_obj()
model_key = "typesafe/jev-1.13.0"
model_cost = litellm.model_cost[model_key]
response = TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=_response(),
response_body={"model": "jev-1.13.0", "usage": {"input_tokens": 312, "output_tokens": 48}},
logging_obj=logging_obj,
url_route="https://api.typesafe.ai/v1/systemone",
result='{"answers": {}}',
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body={"model": "jev-latest"},
)
expected_cost = 312 * model_cost["input_cost_per_token"] + 48 * model_cost["output_cost_per_token"]
assert response["kwargs"]["response_cost"] == pytest.approx(expected_cost)
assert response["kwargs"]["combined_usage_object"].prompt_tokens == 312
assert response["kwargs"]["combined_usage_object"].completion_tokens == 48
assert response["kwargs"]["combined_usage_object"].total_tokens == 360
def test_falls_back_to_request_model_when_response_model_is_missing():
result = _handler_result(
{"usage": {"input_tokens": 10, "output_tokens": 2}},
{"model": "jev-latest"},
)
model_cost = litellm.model_cost["typesafe/jev-latest"]
expected_cost = 10 * model_cost["input_cost_per_token"] + 2 * model_cost["output_cost_per_token"]
assert result["kwargs"]["model"] == "typesafe/jev-latest"
assert result["kwargs"]["response_cost"] == pytest.approx(expected_cost)
def test_call_naming_no_model_is_logged_as_unknown_and_never_priced_as_a_registry_model():
result = _handler_result({"usage": {"input_tokens": 10, "output_tokens": 2}}, {})
assert result["kwargs"]["model"] == "typesafe/unknown"
assert result["kwargs"]["response_cost"] == 0.0
def test_missing_usage_is_zero_cost():
result = _handler_result({"model": "jev-1.13.0"}, {"model": "jev-latest"})
assert result["kwargs"]["response_cost"] == 0.0
def test_records_model_provider_and_cost_on_logging_details():
logging_obj = _logging_obj()
result = TypeSafePassthroughLoggingHandler.typesafe_passthrough_handler(
httpx_response=_response(),
response_body={"model": "jev-1.13.0", "usage": {"input_tokens": 1, "output_tokens": 0}},
logging_obj=logging_obj,
url_route="https://api.typesafe.ai/v1/systemone",
result="{}",
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body={"model": "jev-latest"},
)
assert result["kwargs"]["model"] == "typesafe/jev-1.13.0"
assert result["kwargs"]["custom_llm_provider"] == "typesafe"
assert result["kwargs"]["response_cost"] > 0
assert logging_obj.model_call_details["model"] == "typesafe/jev-1.13.0"
assert logging_obj.model_call_details["custom_llm_provider"] == "typesafe"
assert logging_obj.model_call_details["response_cost"] == result["kwargs"]["response_cost"]
def test_success_handler_dispatches_to_typesafe_handler():
logging_obj = _logging_obj()
normalized = PassThroughEndpointLogging().normalize_llm_passthrough_logging_payload(
httpx_response=_response(),
response_body={"model": "jev-1.13.0", "usage": {"input_tokens": 1, "output_tokens": 0}},
request_body={"model": "jev-latest"},
logging_obj=logging_obj,
url_route="https://api.typesafe.ai/v1/systemone",
result="{}",
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
custom_llm_provider="typesafe",
)
assert normalized["kwargs"]["custom_llm_provider"] == "typesafe"
assert normalized["kwargs"]["model"] == "typesafe/jev-1.13.0"

View file

@ -9,6 +9,7 @@ from types import MappingProxyType, SimpleNamespace
from typing import Final from typing import Final
from unittest import mock from unittest import mock
from unittest.mock import AsyncMock, MagicMock, Mock, patch from unittest.mock import AsyncMock, MagicMock, Mock, patch
from urllib.parse import parse_qs
import httpx import httpx
import pytest import pytest
@ -43,6 +44,7 @@ from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
mistral_proxy_route, mistral_proxy_route,
relay_nvidia_nim_request, relay_nvidia_nim_request,
openai_proxy_route, openai_proxy_route,
typesafe_proxy_route,
vertex_discovery_proxy_route, vertex_discovery_proxy_route,
vertex_proxy_route, vertex_proxy_route,
vllm_proxy_route, vllm_proxy_route,
@ -6136,3 +6138,51 @@ class TestAzureRelayDeploymentSegment:
) )
assert [call["model"] for call in captured] == ["gpt", "gpt"] assert [call["model"] for call in captured] == ["gpt", "gpt"]
class TestTypeSafePassthroughRoute:
@staticmethod
def _request(body: object, query_params: Mapping[str, str] | None = None) -> MagicMock:
request = MagicMock(spec=Request)
request.method = "POST"
request.query_params = query_params or {}
request.json = AsyncMock(return_value=body)
return request
@pytest.mark.asyncio
async def test_forwards_target_auth_headers_provider_and_query(self, monkeypatch):
monkeypatch.setenv("TYPESAFE_API_KEY", "typesafe-test-key")
monkeypatch.setenv("TYPESAFE_API_BASE", "https://typesafe.example/base")
async def fake_upstream(request, *_args):
target: Final = create_route.call_args.kwargs["target"]
upstream_url: Final = httpx.URL(target).copy_merge_params(request.query_params)
return {"upstream_query": parse_qs(upstream_url.query.decode())}
endpoint_func = AsyncMock(side_effect=fake_upstream)
create_route = Mock(return_value=endpoint_func)
monkeypatch.setattr(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route",
create_route,
)
request = self._request({"state": "x"}, {"trace": "yes"})
result = await typesafe_proxy_route(
endpoint="v1/systemone",
request=request,
fastapi_response=MagicMock(spec=Response),
user_api_key_dict=UserAPIKeyAuth(api_key="virtual-key"),
)
assert result == {"upstream_query": {"trace": ["yes"]}}
endpoint_func.assert_awaited_once()
create_route.assert_called_once_with(
endpoint="v1/systemone",
target="https://typesafe.example/base/v1/systemone",
custom_headers={
"Authorization": "Bearer typesafe-test-key",
"Content-Type": "application/json",
},
custom_llm_provider="typesafe",
is_streaming_request=False,
)

View file

@ -0,0 +1,17 @@
import pytest
import litellm
@pytest.fixture(autouse=True)
def local_model_cost_map(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
def test_typesafe_models_share_pricing_and_provider_metadata():
entries = [litellm.model_cost[f"typesafe/{model}"] for model in ("jev-1.13.0", "jev-latest", "jev-preview")]
assert {entry["input_cost_per_token"] for entry in entries} == {entries[0]["input_cost_per_token"]}
assert {entry["output_cost_per_token"] for entry in entries} == {entries[0]["output_cost_per_token"]}
assert {entry["litellm_provider"] for entry in entries} == {"typesafe"}

View file

@ -819,6 +819,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
"container", "container",
"image_edit", "image_edit",
"embedding", "embedding",
"evaluation",
"guardrail", "guardrail",
"image_generation", "image_generation",
"video_generation", "video_generation",

View file

@ -16476,6 +16476,30 @@ export interface paths {
patch: operations["toolset_mcp_route_toolset__toolset_name__mcp_patch"]; patch: operations["toolset_mcp_route_toolset__toolset_name__mcp_patch"];
trace?: never; trace?: never;
}; };
"/typesafe/{endpoint}": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
/**
* Typesafe Proxy Route
* @description [Docs](https://docs.litellm.ai/docs/pass_through/typesafe)
*/
get: operations["typesafe_proxy_route_typesafe__endpoint__get"];
put?: never;
/**
* Typesafe Proxy Route
* @description [Docs](https://docs.litellm.ai/docs/pass_through/typesafe)
*/
post: operations["typesafe_proxy_route_typesafe__endpoint__post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/update/default_team_settings": { "/update/default_team_settings": {
parameters: { parameters: {
query?: never; query?: never;
@ -61585,6 +61609,68 @@ export interface operations {
}; };
}; };
}; };
typesafe_proxy_route_typesafe__endpoint__get: {
parameters: {
query?: never;
header?: never;
path: {
endpoint: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
typesafe_proxy_route_typesafe__endpoint__post: {
parameters: {
query?: never;
header?: never;
path: {
endpoint: string;
};
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
update_default_team_settings_update_default_team_settings_patch: { update_default_team_settings_update_default_team_settings_patch: {
parameters: { parameters: {
query?: never; query?: never;