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

Merge pull request #35916 from BerriAI/litellm_passthrough_live_credentials

Browse source 
fix(proxy): resolve pass-through credentials live from router deployments
This commit is contained in:
Mateo Wang 2026-08-05 12:57:46 -07:00 • committed by GitHub
commit c6dbf48944
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 367 additions and 141 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
basedpyright-code-budget.json
View file

@ -111,7 +111,7 @@
"limit": 20309
},
"reportUnknownVariableType": {
"limit": 31880
"limit": 31879
},
"reportUnnecessaryCast": {
"limit": 124

137
litellm/proxy/pass_through_endpoints/passthrough_endpoint_router.py
View file

@ -1,68 +1,124 @@
from typing import Final
from collections.abc import Callable
from typing import TYPE_CHECKING, Final
import litellm
from litellm._logging import verbose_router_logger
from litellm.integrations.vector_store_integrations.vector_store_pre_call_hook import (
LiteLLM_ManagedVectorStore,
)
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.vertex_ai import VERTEX_CREDENTIALS_TYPES
from litellm.types.passthrough_endpoints.vertex_ai import VertexPassThroughCredentials
from litellm.types.router import LiteLLMParamsTypedDict
if TYPE_CHECKING:
from litellm.router import Router
def _get_proxy_llm_router() -> "Router | None":
from litellm.proxy.proxy_server import llm_router
return llm_router
def _get_str_value(values: dict[str, object] | None, key: str) -> str | None:
value: Final = values.get(key) if values is not None else None
return value if isinstance(value, str) else None
class PassthroughEndpointRouter:
"""
Use this class to Set/Get credentials for pass-through endpoints
Use this class to Get credentials for pass-through endpoints
"""
def __init__(self):
self.credentials: dict[str, str] = {}
def __init__(
self,
llm_router_getter: "Callable[[], Router | None]" = _get_proxy_llm_router,
):
self.llm_router_getter: Final = llm_router_getter
self.deployment_key_to_vertex_credentials: dict[str, VertexPassThroughCredentials] = {}
self.default_vertex_config: VertexPassThroughCredentials | None = None
def set_pass_through_credentials(
self,
custom_llm_provider: str,
api_base: str | None,
api_key: str | None,
):
"""
Set credentials for a pass-through endpoint. Used when a user adds a pass-through LLM endpoint on the UI.
Args:
custom_llm_provider: The provider of the pass-through endpoint
api_base: The base URL of the pass-through endpoint
api_key: The API key for the pass-through endpoint
"""
credential_name: Final = self._get_credential_name_for_provider(
custom_llm_provider=custom_llm_provider,
region_name=self._get_region_name_from_api_base(api_base=api_base, custom_llm_provider=custom_llm_provider),
)
if api_key is None:
raise ValueError("api_key is required for setting pass-through credentials")
self.credentials[credential_name] = api_key
def get_credentials(
self,
custom_llm_provider: str,
region_name: str | None,
) -> str | None:
credential_name: Final = self._get_credential_name_for_provider(
deployment_api_key: Final = self._get_deployment_api_key(
custom_llm_provider=custom_llm_provider,
region_name=region_name,
)
if deployment_api_key is not None:
return deployment_api_key
verbose_router_logger.debug(
"Pass-through llm endpoints router, looking for credentials for %s", credential_name
"No pass-through deployment credentials found for %s, looking for env variable", custom_llm_provider
)
if credential_name in self.credentials:
verbose_router_logger.debug("Found credentials for %s", credential_name)
return self.credentials[credential_name]
else:
verbose_router_logger.debug("No credentials found for %s, looking for env variable", credential_name)
_env_variable_name: Final = self._get_default_env_variable_name_passthrough_endpoint(
custom_llm_provider=custom_llm_provider,
_env_variable_name: Final = self._get_default_env_variable_name_passthrough_endpoint(
custom_llm_provider=custom_llm_provider,
)
return get_secret_str(_env_variable_name)
def _get_deployment_api_key(
self,
custom_llm_provider: str,
region_name: str | None,
) -> str | None:
llm_router: Final = self.llm_router_getter()
if llm_router is None:
return None
deployments: Final = llm_router.get_model_list() or ()
return next(
(
api_key
for deployment in deployments
if (
api_key := self._resolve_matching_deployment_api_key(
litellm_params=deployment["litellm_params"],
custom_llm_provider=custom_llm_provider,
region_name=region_name,
)
)
is not None
),
None,
)
def _resolve_matching_deployment_api_key(
self,
litellm_params: LiteLLMParamsTypedDict,
custom_llm_provider: str,
region_name: str | None,
) -> str | None:
if litellm_params.get("use_in_pass_through") is not True:
return None
if self._get_deployment_provider(litellm_params) != custom_llm_provider:
return None
credential_name: Final = litellm_params.get("litellm_credential_name")
credential_values: Final = (
CredentialAccessor.get_credential_values(credential_name) if credential_name is not None else None
)
api_base: Final = _get_str_value(credential_values, "api_base") or litellm_params.get("api_base")
deployment_region: Final = self._get_region_name_from_api_base(
custom_llm_provider=custom_llm_provider,
api_base=api_base,
)
if deployment_region != region_name:
return None
return _get_str_value(credential_values, "api_key") or litellm_params.get("api_key")
def _get_deployment_provider(self, litellm_params: LiteLLMParamsTypedDict) -> str | None:
model: Final = litellm_params.get("model")
if model is None:
return None
try:
_, provider, _, _ = litellm.get_llm_provider(
model=model,
custom_llm_provider=litellm_params.get("custom_llm_provider"),
)
return get_secret_str(_env_variable_name)
except litellm.exceptions.BadRequestError:
return None
return provider
def _get_vertex_env_vars(self) -> VertexPassThroughCredentials:
"""
@ -165,15 +221,6 @@ class PassthroughEndpointRouter:
else:
return self.default_vertex_config
def _get_credential_name_for_provider(
self,
custom_llm_provider: str,
region_name: str | None,
) -> str:
if region_name is None:
return f"{custom_llm_provider.upper()}_API_KEY"
return f"{custom_llm_provider.upper()}_{region_name.upper()}_API_KEY"
def _get_region_name_from_api_base(
self,
custom_llm_provider: str,

26
litellm/proxy/proxy_server.py
View file

@ -8289,6 +8289,19 @@ class ProxyStartupEvent:
)
if store_model_in_db is True:
### GET STORED CREDENTIALS ###
scheduler.add_job(
proxy_config.get_credentials,
"interval",
seconds=config_reload_interval_seconds,
# REMOVED jitter parameter - major cause of memory leak
args=[prisma_client],
id="get_credentials_job",
replace_existing=True,
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
)
await proxy_config.get_credentials(prisma_client=prisma_client)
# MEMORY LEAK FIX: Increase interval from 10s to 30s minimum
# Frequent polling was causing excessive memory allocations
scheduler.add_job(
@ -8305,19 +8318,6 @@ class ProxyStartupEvent:
# this will load all existing models on proxy startup
await proxy_config.add_deployment(prisma_client=prisma_client, proxy_logging_obj=proxy_logging_obj)
### GET STORED CREDENTIALS ###
scheduler.add_job(
proxy_config.get_credentials,
"interval",
seconds=config_reload_interval_seconds,
# REMOVED jitter parameter - major cause of memory leak
args=[prisma_client],
id="get_credentials_job",
replace_existing=True,
misfire_grace_time=APSCHEDULER_MISFIRE_GRACE_TIME,
)
await proxy_config.get_credentials(prisma_client=prisma_client)
proxy_config.start_config_sync_subscriber(
prisma_client=prisma_client,
proxy_logging_obj=proxy_logging_obj,

75
litellm/router.py
View file

@ -8115,7 +8115,6 @@ class Router:
self._initialize_deployment_for_pass_through(
deployment=deployment,
custom_llm_provider=custom_llm_provider,
model=deployment.litellm_params.model,
)
#########################################################
@ -8142,55 +8141,39 @@ class Router:
return deployment
def _initialize_deployment_for_pass_through(self, deployment: Deployment, custom_llm_provider: str, model: str):
def _initialize_deployment_for_pass_through(self, deployment: Deployment, custom_llm_provider: str):
"""
Optional: Initialize deployment for pass-through endpoints if `deployment.litellm_params.use_in_pass_through` is True
Optional: Register vertex credentials for pass-through endpoints if `deployment.litellm_params.use_in_pass_through` is True
Each provider uses diff .env vars for pass-through endpoints, this helper uses the deployment credentials to set the .env vars for pass-through endpoints
Other providers need no registration here: PassthroughEndpointRouter.get_credentials resolves their credentials per-request from the live router deployments
"""
if deployment.litellm_params.use_in_pass_through is True:
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
passthrough_endpoint_router,
if deployment.litellm_params.use_in_pass_through is not True:
return
if custom_llm_provider != "vertex_ai":
return
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
passthrough_endpoint_router,
)
credential_name: Final = deployment.litellm_params.litellm_credential_name
credential_values: Final = (
CredentialAccessor.get_credential_values(credential_name) if credential_name is not None else {}
)
vertex_project: Final = credential_values.get("vertex_project") or deployment.litellm_params.vertex_project
vertex_location: Final = credential_values.get("vertex_location") or deployment.litellm_params.vertex_location
vertex_credentials: Final = (
credential_values.get("vertex_credentials") or deployment.litellm_params.vertex_credentials
)
if vertex_project is None or vertex_location is None:
raise ValueError(
"vertex_project, and vertex_location must be set in litellm_params for pass-through endpoints."
)
if deployment.litellm_params.litellm_credential_name is not None:
credential_values = CredentialAccessor.get_credential_values(
deployment.litellm_params.litellm_credential_name
)
else:
credential_values = {}
if custom_llm_provider == "vertex_ai":
vertex_project = credential_values.get("vertex_project") or deployment.litellm_params.vertex_project
vertex_location = credential_values.get("vertex_location") or deployment.litellm_params.vertex_location
vertex_credentials: Final = (
credential_values.get("vertex_credentials") or deployment.litellm_params.vertex_credentials
)
if vertex_project is None or vertex_location is None:
raise ValueError(
"vertex_project, and vertex_location must be set in litellm_params for pass-through endpoints."
)
passthrough_endpoint_router.add_vertex_credentials(
project_id=vertex_project,
location=vertex_location,
vertex_credentials=vertex_credentials,
)
else:
api_base: Final = credential_values.get("api_base") or deployment.litellm_params.api_base
api_key: Final = credential_values.get("api_key") or deployment.litellm_params.api_key
if api_key is None:
verbose_router_logger.debug(
"Skipping pass-through credential setup for deployment model=%s, custom_llm_provider=%s; no api_key set. Providers like bedrock resolve credentials at request time.",
model,
custom_llm_provider,
)
return
passthrough_endpoint_router.set_pass_through_credentials(
custom_llm_provider=custom_llm_provider,
api_base=api_base,
api_key=api_key,
)
passthrough_endpoint_router.add_vertex_credentials(
project_id=vertex_project,
location=vertex_location,
vertex_credentials=vertex_credentials,
)
def add_deployment(self, deployment: Deployment) -> Deployment | None:
"""

3
litellm/types/router.py
View file

@ -379,6 +379,9 @@ class LiteLLMParamsTypedDict(TypedDict, total=False):
drop_params: bool | None
## RESPONSES API → CHAT COMPLETIONS BRIDGE ##
use_chat_completions_api: bool | None
## PASS-THROUGH ENDPOINTS ##
use_in_pass_through: bool | None
litellm_credential_name: str | None
## UNIFIED PROJECT/REGION ##
region_name: str | None
## VERTEX AI ##

2
ruff-strict-budget.json
View file

@ -9,7 +9,7 @@
"limit": 840
},
"ANN201": {
"limit": 2039
"limit": 2038
},
"ANN202": {
"limit": 871

72
tests/pass_through_unit_tests/test_unit_test_passthrough_router.py
View file

@ -31,38 +31,60 @@ passthrough_endpoint_router = PassthroughEndpointRouter()
class TestPassthroughEndpointRouter(unittest.TestCase):
def setUp(self):
self.router = PassthroughEndpointRouter()
self.router = PassthroughEndpointRouter(llm_router_getter=lambda: None)
def test_set_and_get_credentials(self):
def test_deployment_and_get_credentials(self):
"""
1. Basic Usage:
- Set credentials for OpenAI, AssemblyAI, Anthropic, Cohere
- GET credentials from passthrough_endpoint_router (from the memory store when available)
- Flag deployments for OpenAI, AssemblyAI, Anthropic, Cohere with use_in_pass_through
- GET credentials from passthrough_endpoint_router (resolved live from the llm router)
"""
import litellm
# OpenAI: standard (no region-specific logic)
self.router.set_pass_through_credentials("openai", None, "openai_key")
self.assertEqual(self.router.get_credentials("openai", None), "openai_key")
# AssemblyAI: using an API base that contains 'eu' should trigger regional logic.
api_base_eu = "https://api.eu.assemblyai.com"
self.router.set_pass_through_credentials(
"assemblyai", api_base_eu, "assemblyai_key"
)
# When calling get_credentials, pass the region "eu" (extracted from the API base)
self.assertEqual(
self.router.get_credentials("assemblyai", "eu"), "assemblyai_key"
llm_router = litellm.Router(
model_list=[
{
"model_name": "gpt-4o",
"litellm_params": {
"model": "openai/gpt-4o",
"api_key": "openai_key",
"use_in_pass_through": True,
},
},
{
"model_name": "best",
"litellm_params": {
"model": "assemblyai/best",
"api_key": "assemblyai_key",
"api_base": "https://api.eu.assemblyai.com",
"use_in_pass_through": True,
},
},
{
"model_name": "claude-sonnet-4-5",
"litellm_params": {
"model": "anthropic/claude-sonnet-4-5",
"api_key": "anthropic_key",
"use_in_pass_through": True,
},
},
{
"model_name": "embed-english-v3.0",
"litellm_params": {
"model": "cohere/embed-english-v3.0",
"api_key": "cohere_key",
"use_in_pass_through": True,
},
},
]
)
router = PassthroughEndpointRouter(llm_router_getter=lambda: llm_router)
# Anthropic: no region set
self.router.set_pass_through_credentials("anthropic", None, "anthropic_key")
self.assertEqual(
self.router.get_credentials("anthropic", None), "anthropic_key"
)
# Cohere: no region set
self.router.set_pass_through_credentials("cohere", None, "cohere_key")
self.assertEqual(self.router.get_credentials("cohere", None), "cohere_key")
self.assertEqual(router.get_credentials("openai", None), "openai_key")
# AssemblyAI: an API base that contains 'eu' triggers regional matching
self.assertEqual(router.get_credentials("assemblyai", "eu"), "assemblyai_key")
self.assertEqual(router.get_credentials("anthropic", None), "anthropic_key")
self.assertEqual(router.get_credentials("cohere", None), "cohere_key")
def test_get_credentials_from_env(self):
"""

3
tests/router_unit_tests/test_router_adding_deployments.py
View file

@ -60,7 +60,6 @@ def test_initialize_deployment_for_pass_through_success(reusable_credentials):
router._initialize_deployment_for_pass_through(
deployment=deployment,
custom_llm_provider="vertex_ai",
model="vertex_ai/test-model",
)
# Verify the credentials were properly set
@ -100,7 +99,6 @@ def test_initialize_deployment_for_pass_through_missing_params():
router._initialize_deployment_for_pass_through(
deployment=deployment,
custom_llm_provider="vertex_ai",
model="vertex_ai/test-model",
)
@ -120,7 +118,6 @@ def test_initialize_deployment_when_pass_through_disabled():
router._initialize_deployment_for_pass_through(
deployment=deployment,
custom_llm_provider="vertex_ai",
model="vertex_ai/test-model",
)
# If we reach this point, the test passes as the method exited without raising any errors

174
tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_endpoint_router.py Normal file
View file

@ -0,0 +1,174 @@
import pytest
import litellm
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
from litellm.proxy.pass_through_endpoints.passthrough_endpoint_router import (
PassthroughEndpointRouter,
)
from litellm.types.utils import CredentialItem
@pytest.fixture(autouse=True)
def isolated_credential_list(monkeypatch):
monkeypatch.setattr(litellm, "credential_list", [])
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("ASSEMBLYAI_API_KEY", raising=False)
def _credential(name: str, api_key: str) -> CredentialItem:
return CredentialItem(
credential_name=name,
credential_values={"api_key": api_key},
credential_info={},
)
def _flagged_deployment(model: str, **litellm_params) -> dict:
return {
"model_name": model.split("/", 1)[-1],
"litellm_params": {"model": model, "use_in_pass_through": True, **litellm_params},
}
def _passthrough_router(llm_router: litellm.Router | None) -> PassthroughEndpointRouter:
return PassthroughEndpointRouter(llm_router_getter=lambda: llm_router)
def test_credential_loaded_after_deployment_registration_still_resolves():
llm_router = litellm.Router(
model_list=[_flagged_deployment("openai/gpt-4o", litellm_credential_name="cred_openai")]
)
passthrough_router = _passthrough_router(llm_router)
assert passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) is None
CredentialAccessor.upsert_credentials([_credential("cred_openai", "sk-loaded-after-boot")])
assert (
passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None)
== "sk-loaded-after-boot"
)
def test_credential_rotation_is_reflected_without_deployment_update():
CredentialAccessor.upsert_credentials([_credential("cred_openai", "sk-before-rotation")])
llm_router = litellm.Router(
model_list=[_flagged_deployment("openai/gpt-4o", litellm_credential_name="cred_openai")]
)
passthrough_router = _passthrough_router(llm_router)
assert (
passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None)
== "sk-before-rotation"
)
CredentialAccessor.upsert_credentials([_credential("cred_openai", "sk-after-rotation")])
assert (
passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None)
== "sk-after-rotation"
)
def test_deleted_deployment_stops_serving_its_key(monkeypatch):
llm_router = litellm.Router(model_list=[_flagged_deployment("openai/gpt-4o", api_key="sk-inline")])
passthrough_router = _passthrough_router(llm_router)
assert passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) == "sk-inline"
llm_router.set_model_list([])
monkeypatch.setenv("OPENAI_API_KEY", "sk-from-env")
assert (
passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) == "sk-from-env"
)
def test_inline_api_key_resolves_without_credential_name():
llm_router = litellm.Router(
model_list=[_flagged_deployment("anthropic/claude-sonnet-4-5", api_key="sk-ant-inline")]
)
passthrough_router = _passthrough_router(llm_router)
assert (
passthrough_router.get_credentials(custom_llm_provider="anthropic", region_name=None)
== "sk-ant-inline"
)
def test_missing_credential_and_no_inline_key_falls_back_to_env(monkeypatch):
llm_router = litellm.Router(
model_list=[_flagged_deployment("openai/gpt-4o", litellm_credential_name="cred_deleted")]
)
passthrough_router = _passthrough_router(llm_router)
monkeypatch.setenv("OPENAI_API_KEY", "sk-from-env")
assert (
passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) == "sk-from-env"
)
def test_deployment_for_other_provider_does_not_match():
llm_router = litellm.Router(
model_list=[_flagged_deployment("anthropic/claude-sonnet-4-5", api_key="sk-ant-inline")]
)
passthrough_router = _passthrough_router(llm_router)
assert passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) is None
def test_unflagged_deployment_does_not_match():
llm_router = litellm.Router(
model_list=[
{
"model_name": "gpt-4o",
"litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-not-flagged"},
}
]
)
passthrough_router = _passthrough_router(llm_router)
assert passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) is None
def test_first_matching_deployment_wins():
llm_router = litellm.Router(
model_list=[
_flagged_deployment("openai/gpt-4o", api_key="sk-first"),
_flagged_deployment("openai/gpt-4o-mini", api_key="sk-second"),
]
)
passthrough_router = _passthrough_router(llm_router)
assert passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) == "sk-first"
def test_assemblyai_region_matching():
llm_router = litellm.Router(
model_list=[
_flagged_deployment(
"assemblyai/best", api_key="sk-eu", api_base="https://api.eu.assemblyai.com"
),
_flagged_deployment("assemblyai/best", api_key="sk-us", api_base="https://api.assemblyai.com"),
]
)
passthrough_router = _passthrough_router(llm_router)
assert passthrough_router.get_credentials(custom_llm_provider="assemblyai", region_name="eu") == "sk-eu"
assert passthrough_router.get_credentials(custom_llm_provider="assemblyai", region_name=None) == "sk-us"
def test_env_fallback_when_no_router(monkeypatch):
passthrough_router = _passthrough_router(None)
monkeypatch.setenv("OPENAI_API_KEY", "sk-from-env")
assert (
passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) == "sk-from-env"
)
def test_returns_none_when_no_router_and_no_env():
passthrough_router = _passthrough_router(None)
assert passthrough_router.get_credentials(custom_llm_provider="openai", region_name=None) is None

10
tests/test_litellm/test_router.py
View file

@ -6021,16 +6021,16 @@ def test_initialize_deployment_for_pass_through_keeps_bedrock_iam_deployment():
]
def test_initialize_deployment_for_pass_through_sets_credentials_with_api_key():
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
passthrough_endpoint_router,
def test_pass_through_deployment_api_key_resolves_via_get_credentials():
from litellm.proxy.pass_through_endpoints.passthrough_endpoint_router import (
PassthroughEndpointRouter,
)
passthrough_endpoint_router.credentials.clear()
router = _router_with_two_pass_through_deployments([False, False])
passthrough_router = PassthroughEndpointRouter(llm_router_getter=lambda: router)
assert len(router.get_model_list()) == 2
assert (
passthrough_endpoint_router.get_credentials(
passthrough_router.get_credentials(
custom_llm_provider="openai", region_name=None
)
== "sk-fake-for-tests"

4
type-discipline-budget.json
View file

@ -3,7 +3,7 @@
"limit": 23346
},
"LIT002": {
"limit": 27224
"limit": 27223
},
"LIT003": {
"limit": 269
@ -27,7 +27,7 @@
"limit": 0
},
"LIT010": {
"limit": 16828
"limit": 16824
},
"LIT011": {
"limit": 5603