""" Allow proxy admin to add/update/delete models in the db Currently most endpoints are in `proxy_server.py`, but those should be moved here over time. Endpoints here: model/{model_id}/update - PATCH endpoint for model update. """ #### MODEL MANAGEMENT #### import asyncio import datetime import json from collections.abc import Awaitable, Mapping, Sequence from json import JSONDecodeError from types import MappingProxyType from typing import TYPE_CHECKING, Annotated, Final, Literal, Protocol, cast from fastapi import APIRouter, Depends, Header, HTTPException, Request, status from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator from litellm._logging import verbose_proxy_logger from litellm._uuid import uuid from litellm.constants import LITELLM_PROXY_ADMIN_NAME from litellm.litellm_core_utils.ptu_pricing import ( CUSTOM_PRICING_FIELDS, PTU_EMPTIED_PRICING_FIELDS, PTU_MODEL_INFO_FIELDS, PTU_ZEROED_PRICING_FIELDS, PTU_ZEROED_TABLE_FIELDS, SEARCH_CONTEXT_SIZES, ptu_config_error, ) from litellm.proxy._types import ( BlockModelRequest, CommonProxyErrors, LiteLLM_ProxyModelTable, LiteLLM_TeamTable, LitellmTableNames, LitellmUserRoles, ModelInfoDelete, PrismaCompatibleUpdateDBModel, ProxyErrorTypes, ProxyException, ReconcileOutcome, TeamModelAddRequest, TeamModelDeleteRequest, UserAPIKeyAuth, ) from litellm.proxy.auth.user_api_key_auth import user_api_key_auth from litellm.proxy.common_utils.config_sync_pubsub import ( coordination_redis_cache, publish_config_change, ) from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache from litellm.proxy.management_endpoints.common_utils import _is_user_team_admin from litellm.proxy.management_endpoints.team_endpoints import ( _refresh_cached_team, team_model_add, team_model_delete, ) from litellm.proxy.management_endpoints.team_endpoints import ( update_team as _legacy_update_team, ) from litellm.proxy.management_helpers.audit_logs import create_object_audit_log from litellm.proxy.spend_tracking.ptu_feature_flag import ( PTU_COST_ATTRIBUTION_ENV_VAR, is_ptu_cost_attribution_enabled, ) from litellm.proxy.utils import PrismaClient, ProxyLogging from litellm.repositories.model_repository import ModelRepository from litellm.repositories.prisma_protocols import TableActions from litellm.repositories.table_repositories import ModelTableRepository from litellm.repositories.team_repository import TeamRepository from litellm.router import Router from litellm.router_strategy.complexity_router import ( DEFAULT_CLASSIFIER_CONTEXT_WINDOW_SIZE, ClassificationRubric, ComplexityRouterConfig, ComplexityTier, TierDefinition, classification_system_prompt, custom_tier_classification_prompt, normalize_classification_prompt, ) from litellm.router_utils.auto_router_model_naming import ( STRATEGY_ROUTER_PARAM_FIELDS, carries_complexity_router_settings, validate_complexity_router_config_placement, validate_complexity_router_config_write, validate_strategy_router_model_write, ) from litellm.types.proxy.management_endpoints.model_management_endpoints import ( AutoRouterClassifierDefaultPromptResponse, UpdateUsefulLinksRequest, ) from litellm.types.router import ( SPECIAL_MODEL_INFO_PARAMS, Deployment, GenericLiteLLMParams, ModelInfo, updateDeployment, ) from litellm.utils import get_utc_datetime if TYPE_CHECKING: from prisma import models as prisma_models router: Final = APIRouter() async def update_team(*args, **kwargs): """ Backward-compatible shim for tests/legacy call sites that patch this symbol. Team model management now uses team_model_add/team_model_delete directly. """ return await _legacy_update_team(*args, **kwargs) class UpdatePublicModelGroupsRequest(BaseModel): """Request model for updating public model groups""" model_groups: list[str] = Field(description="List of model group names to make public") model_config = ConfigDict(extra="forbid") class _ProxyModelRow(Protocol): @property def model_id(self) -> str: ... @property def model_name(self) -> str: ... @property def model_info(self) -> object: ... def model_dump_json(self, *, exclude_none: bool = False) -> str: ... class _ProxyModelTable(Protocol): def find_unique(self, *, where: Mapping[str, object]) -> Awaitable[BaseModel | None]: ... def find_many(self, *, where: Mapping[str, object]) -> Awaitable[Sequence[_ProxyModelRow]]: ... def update( self, *, where: Mapping[str, object], data: Mapping[str, object] ) -> Awaitable[_ProxyModelRow | None]: ... def delete(self, *, where: Mapping[str, object]) -> Awaitable[_ProxyModelRow | None]: ... def delete_many(self, *, where: Mapping[str, object]) -> Awaitable[int]: ... class _TxModelTables(Protocol): litellm_proxymodeltable: _ProxyModelTable class _ExistingModelRow(Protocol): @property def litellm_params(self) -> Mapping[str, object]: ... def model_dump_json(self, *, exclude_none: bool = False) -> str: ... class _TeamRow(Protocol): @property def models(self) -> Sequence[str]: ... def model_dump(self) -> Mapping[str, object]: ... class _TeamLookupTable(Protocol): def find_unique(self, *, where: Mapping[str, object]) -> Awaitable[_TeamRow | None]: ... class _TeamTable(_TeamLookupTable, Protocol): def update( self, *, where: Mapping[str, object], data: Mapping[str, object], include: Mapping[str, bool] ) -> Awaitable[LiteLLM_TeamTable]: ... def _proxy_model_table(prisma_client: PrismaClient) -> _ProxyModelTable: return ModelRepository(prisma_client).table def _repo_team_table(prisma_client: PrismaClient) -> _TeamLookupTable: return TeamRepository(prisma_client).table def _db_team_table(prisma_client: PrismaClient) -> _TeamTable: return prisma_client.db.litellm_teamtable def _model_alias_table(prisma_client: PrismaClient) -> "TableActions[prisma_models.LiteLLM_ModelTable]": return ModelTableRepository(prisma_client).table async def get_db_model(model_id: str, prisma_client: PrismaClient) -> Deployment | None: db_model: Final = await _proxy_model_table(prisma_client).find_unique(where={"model_id": model_id}) if not db_model: return None deployment_pydantic_obj: Final = Deployment(**db_model.model_dump(exclude_none=True)) return deployment_pydantic_obj def _strategy_router_write_violation( incoming_params: GenericLiteLLMParams | None, existing_params: GenericLiteLLMParams | None, ) -> str | None: """Reject writes that would corrupt a strategy router's pseudo-model. An auto-router deployment's ``litellm_params.model`` (``auto_router/...``) is the discriminator the router loads it by; a write that mangles it makes the router drop the deployment silently under ``ignore_invalid_deployments``. Only writes that supply ``litellm_params.model`` are judged on the naming contract, against the merged (stored + incoming) params, so partial patches and restores of an already-corrupted row stay legal. A config is judged only when the write carries one, for the same reason: a rename must not be held hostage by a stored config it does not touch. Returns the violation, or None. """ if incoming_params is None: return None config_violation: Final = validate_complexity_router_config_write( complexity_router_config=incoming_params.complexity_router_config ) if config_violation is not None: return config_violation present_fields: Final = frozenset( field for field in STRATEGY_ROUTER_PARAM_FIELDS for source in (incoming_params, existing_params) if source is not None and getattr(source, field, None) is not None ) # Scope reads the incoming model because the stored one is encrypted at rest. if carries_complexity_router_settings(incoming_params.model, present_fields): placement_violation: Final = validate_complexity_router_config_placement(incoming_params.model_extra) if placement_violation is not None: return placement_violation if incoming_params.model is None: return None return validate_strategy_router_model_write(model=incoming_params.model, present_fields=present_fields) def _raise_on_strategy_router_write_violation( incoming_params: GenericLiteLLMParams | None, existing_params: GenericLiteLLMParams | None, ) -> None: violation = _strategy_router_write_violation(incoming_params=incoming_params, existing_params=existing_params) if violation is None: return raise ProxyException( message=violation, type=ProxyErrorTypes.validation_error.value, code=status.HTTP_400_BAD_REQUEST, param="litellm_params.model", ) ENFORCE_RPM_TPM_ON_MODEL_ADD_SETTING: Final = "enforce_rpm_tpm_on_model_add" _REQUIRED_RATE_LIMIT_FIELDS: Final = ("rpm", "tpm") def _raise_if_rate_limits_required_but_missing(*, litellm_params: GenericLiteLLMParams, enforced: bool) -> None: """Require both rpm and tpm (each a positive value) when the operator opts in via config.yaml. Off by default, so deployments keep adding models without limits. When ``enforce_rpm_tpm_on_model_add: true`` is set under general_settings, a model added without both rpm and tpm set to a positive value is rejected rather than stored unbounded (or effectively excluded from routing by a zero/negative limit). """ if not enforced: return missing: Final = tuple( field for field in _REQUIRED_RATE_LIMIT_FIELDS if (value := getattr(litellm_params, field)) is None or value <= 0 ) if not missing: return raise ProxyException( message=( f"{' and '.join(missing)} must be set to a positive value when " f"'{ENFORCE_RPM_TPM_ON_MODEL_ADD_SETTING}' is enabled in general_settings" ), type=ProxyErrorTypes.validation_error.value, code=status.HTTP_400_BAD_REQUEST, param=f"litellm_params.{missing[0]}", ) _PTU_PRICED_PAIR: Final = frozenset({"ptu_count", "cost_per_ptu_per_hour"}) def _explicitly_cleared_ptu_fields(model_info: ModelInfo | None) -> frozenset[str]: """The PTU fields a patch sends as an explicit null, which update_db_model drops. Empty while the feature is off, so disabling pauses PTU rather than letting a client that round-trips a model_info blob erase a configuration set up during an earlier opt-in. """ if model_info is None or not is_ptu_cost_attribution_enabled(): return frozenset() return frozenset( field for field in PTU_MODEL_INFO_FIELDS if field in model_info.model_fields_set and getattr(model_info, field) is None ) def _merged_ptu_model_info(*, db_model: Deployment, patch_data: updateDeployment) -> Mapping[str, object]: """The model_info a patch would store, which is the stored blob updated by the patch. A PTU invariant holds over the deployment as it will exist, not over whichever subset of fields a caller happened to send. """ stored: Final = db_model.model_info.model_dump(exclude_none=True) if db_model.model_info else _EMPTY_MODEL_INFO incoming: Final = ( patch_data.model_info.model_dump(exclude_none=True) if patch_data.model_info else _EMPTY_MODEL_INFO ) cleared: Final = _explicitly_cleared_ptu_fields(patch_data.model_info) return MappingProxyType({k: v for k, v in {**stored, **incoming}.items() if k not in cleared}) def _raise_if_ptu_cost_attribution_disabled(incoming_model_info: Mapping[str, object]) -> None: """Reject PTU model_info fields unless the operator opted into PTU cost attribution. Takes the incoming request's model_info rather than the merged deployment, so an unrelated patch of a model that still stores PTU config from an earlier opt-in is left alone. The fields are rejected rather than dropped so a caller never believes a flat cost was configured while the rollup that would price it is not running. Only a value is rejected. An explicit null reaches the clear loop, which is gated on the same flag, so a disabled proxy neither writes PTU config nor erases what an earlier opt-in stored. Disabling pauses the feature rather than discarding its setup. """ if is_ptu_cost_attribution_enabled(): return supplied: Final = tuple(field for field in PTU_MODEL_INFO_FIELDS if incoming_model_info.get(field) is not None) if not supplied: return raise HTTPException( status_code=400, detail=( f"PTU cost attribution is disabled, so {', '.join(supplied)} cannot be set. " f"Set {PTU_COST_ATTRIBUTION_ENV_VAR}=true to enable it." ), ) def _validate_ptu_model_info(model_info: Mapping[str, object]) -> None: """Enforce the PTU cross-field invariant on the effective model_info. The rules live in litellm_core_utils.ptu_pricing so that config.yaml registration refuses the same deployments this endpoint does, for the same reason. Per-field bounds (positive count, non-negative rate) are enforced by ModelInfo itself. Registration additionally requires an operator-declared ``model_info.id``, which this endpoint does not: a stored deployment already holds a stable primary key, where a config-declared one is otherwise keyed by a hash of its own parameters. """ error: Final = ptu_config_error(model_info) if error is not None: raise HTTPException(status_code=400, detail=error) # The mirrored per-token pricing fields plus the remaining rates the public cost map or a # provider default would otherwise supply (the cache back-fills, the Maps grounding rate). An # unset field falls back to those sources, so a field left out here is one a PTU deployment # still bills. # tiered_pricing is the one mirrored field that is a table of ranges, not a rate, so it is stored # empty (see _PTU_EMPTIED_PRICING_FIELDS): its tiers outrank the zeros written beside them, so # dropping it would leave the cost map's tiers billing the traffic the reserved capacity covers. _PTU_ZEROED_PRICING_FIELDS: Final = PTU_ZEROED_PRICING_FIELDS _PTU_EMPTIED_PRICING_FIELDS: Final = PTU_EMPTIED_PRICING_FIELDS _PTU_ZEROED_PRICING: Final[Mapping[str, float | tuple[()]]] = MappingProxyType( { **dict.fromkeys(_PTU_ZEROED_PRICING_FIELDS, 0.0), **dict.fromkeys(_PTU_EMPTIED_PRICING_FIELDS, ()), } ) _NO_PRICING_OVERRIDE: Final[Mapping[str, float | tuple[()]]] = MappingProxyType({}) _EMPTY_MODEL_INFO: Final[Mapping[str, object]] = _NO_PRICING_OVERRIDE # Rate fields only. CustomPricingLiteLLMParams also carries settings that are not charges # (an embedding's output_vector_size, the regional uplift multipliers), and zeroing one of # those would destroy the deployment's configuration rather than stop a charge. _CUSTOM_PRICING_FIELDS: Final = CUSTOM_PRICING_FIELDS # search_context_cost_per_query holds its rates in a table keyed by context size, and an absent # table means the provider's own default rate rather than free (litellm/llms/gemini/cost_calculator # falls back to $0.035), so it is zeroed in place rather than emptied like tiered_pricing, and # written on every PTU deployment rather than only where a table is already stored. _PTU_ZEROED_TABLE_FIELDS: Final = PTU_ZEROED_TABLE_FIELDS _SEARCH_CONTEXT_SIZES: Final = SEARCH_CONTEXT_SIZES def _is_nonzero_rate(value: object) -> bool: return isinstance(value, (int, float)) and not isinstance(value, bool) and value != 0 def _is_nonzero_price(value: object) -> bool: if isinstance(value, dict): # an all-zero table is how a rate is expressed as free return any(_is_nonzero_rate(rate) for rate in value.values()) return _is_nonzero_rate(value) def _is_zero_price(value: object) -> bool: if isinstance(value, dict): return bool(value) and not _is_nonzero_price(value) if isinstance(value, (list, tuple)): return not value return isinstance(value, (int, float)) and not isinstance(value, bool) and value == 0 def _raise_if_ptu_deployment_is_priced(*, model_info: Mapping[str, object], supplied: Mapping[str, object]) -> None: """Refuse a rate the caller supplies for a deployment that bills reserved capacity. Separate from the zeroing so the team-model path can run it before it touches the team, whose ACL write autocommits: a refusal raised after it would leave the team changed and the deployment row never written. """ if not is_ptu_cost_attribution_enabled(): return if model_info.get("ptu_count") is None or model_info.get("cost_per_ptu_per_hour") is None: return priced: Final = tuple( sorted( tuple(field for field in _CUSTOM_PRICING_FIELDS if _is_nonzero_price(supplied.get(field))) + tuple(field for field in _PTU_EMPTIED_PRICING_FIELDS if supplied.get(field)) ) ) if not priced: return raise HTTPException( status_code=400, detail=( f"A PTU deployment bills by reserved capacity, so {', '.join(priced)} cannot be charged on " "top of it. Send 0 or no value, or remove ptu_count and cost_per_ptu_per_hour to bill per token." ), ) def _ptu_zeroed_pricing( *, model_info: Mapping[str, object], litellm_params: Mapping[str, object], supplied: Mapping[str, object], ) -> Mapping[str, float | tuple[()] | Mapping[str, float]]: """The pricing a PTU deployment must carry, empty unless one is being stored. Reserved capacity is already billed by the flat cost the rollup writes, so charging the traffic it serves bills the same tokens twice. Left unset the rate falls back to the public cost map, which makes the double charge the default rather than an opt-in. Only a price the caller supplies is refused. A non-zero price already on the row is zeroed instead, so a deployment priced through a path this rule does not cover heals on its next save rather than rejecting every later edit of a field that has nothing to do with pricing. ``supplied`` is the caller's litellm_params alone, because that is the blob a price is authored on. model_info's copy is written by the server, both by the mirror in ``Deployment.__init__`` and by the cost-map defaults /model/info fills in, so a client that round-trips a model_info blob sends back prices it never chose. """ if not is_ptu_cost_attribution_enabled(): return _NO_PRICING_OVERRIDE if model_info.get("ptu_count") is None or model_info.get("cost_per_ptu_per_hour") is None: return _NO_PRICING_OVERRIDE _raise_if_ptu_deployment_is_priced(model_info=model_info, supplied=supplied) stored: Final = frozenset( field for field in _CUSTOM_PRICING_FIELDS if _is_nonzero_price(model_info.get(field)) or _is_nonzero_price(litellm_params.get(field)) ) return MappingProxyType( { **_PTU_ZEROED_PRICING, **dict.fromkeys(_PTU_ZEROED_TABLE_FIELDS, dict.fromkeys(_SEARCH_CONTEXT_SIZES, 0.0)), **dict.fromkeys(stored - _PTU_ZEROED_TABLE_FIELDS, 0.0), } ) def _ptu_pricing_delta( *, stored_model_info: Mapping[str, object], model_info: Mapping[str, object], litellm_params: Mapping[str, object], patch: updateDeployment, ) -> tuple[Mapping[str, float | tuple[()] | Mapping[str, float]], frozenset[str]]: """The pricing a patch must write into both blobs, and the pricing it must drop from them. A patch that takes the deployment off PTU takes the zeroed pricing with it, since the zeros exist only to stop the double charge. Left behind they would serve the deployment for free. Reading the stored row rather than the patch alone keeps that release off a deployment that never carried PTU config, whose zero price is a rate its operator chose. A zero the patch itself carries is released with the rest, because the dashboard echoes the whole stored blob on every save, so a supplied zero cannot be told apart from the one this rule wrote. The release spans every field the zeroing could have written, not just the mirrored ones, or a rate zeroed on the way in (per-second, per-character tiers) would bill nothing forever. """ supplied: Final = patch.litellm_params.model_dump(exclude_none=True) if patch.litellm_params else _EMPTY_MODEL_INFO zeroed: Final = _ptu_zeroed_pricing(model_info=model_info, litellm_params=litellm_params, supplied=supplied) if zeroed: return zeroed, frozenset() was_ptu: Final = any(stored_model_info.get(field) is not None for field in _PTU_PRICED_PAIR) if not was_ptu or not _explicitly_cleared_ptu_fields(patch.model_info) & _PTU_PRICED_PAIR: return _NO_PRICING_OVERRIDE, frozenset() return _NO_PRICING_OVERRIDE, frozenset( field for field in _CUSTOM_PRICING_FIELDS.union(_PTU_ZEROED_PRICING_FIELDS, _PTU_EMPTIED_PRICING_FIELDS) if _is_zero_price(model_info.get(field)) or _is_zero_price(litellm_params.get(field)) ) def _ptu_priced_deployment(model_params: Deployment) -> Deployment: """``model_params`` with PTU pricing applied, or itself when it configures no PTU.""" model_info: Final = model_params.model_info.model_dump(exclude_none=True) litellm_params: Final = model_params.litellm_params.model_dump(exclude_none=True) override: Final = _ptu_zeroed_pricing(model_info=model_info, litellm_params=litellm_params, supplied=litellm_params) if not override: return model_params # model_copy validates nothing, so the emptied tier table has to arrive as the list the field # declares or Pydantic warns on every later dump of it stored: Final = MappingProxyType( {key: [] if isinstance(value, tuple) else value for key, value in override.items()} ) return model_params.model_copy( update=MappingProxyType( { "litellm_params": model_params.litellm_params.model_copy(update=stored), "model_info": model_params.model_info.model_copy(update=stored), } ) ) def update_db_model(db_model: Deployment, updated_patch: updateDeployment) -> PrismaCompatibleUpdateDBModel: if updated_patch.model_info is not None: _raise_if_ptu_cost_attribution_disabled(updated_patch.model_info.model_dump(exclude_none=True)) merged_model_name: Final = updated_patch.model_name or db_model.model_name merged_litellm_params: Final = db_model.litellm_params.model_dump(exclude_none=True) merged_model_info: Final = db_model.model_info.model_dump(exclude_none=True) # update litellm params if updated_patch.litellm_params: # Encrypt any sensitive values encrypted_params: Final = { k: encrypt_value_helper(v) for k, v in updated_patch.litellm_params.model_dump(exclude_none=True).items() } merged_litellm_params.update(encrypted_params) # update model info if updated_patch.model_info: merged_model_info.update(updated_patch.model_info.model_dump(exclude_none=True)) # Honor explicit-null clears LAST, after both merges, so a model_info blob the UI # passes through (which today re-sends the OLD pricing on every save) cannot # silently undo a litellm_params clear via .update(). # # Restricted to SPECIAL_MODEL_INFO_PARAMS (input/output cost per token/character # and cache read/write costs) so this path cannot be used to null out privileged # model_info fields like team_id or access groups. SPECIAL_MODEL_INFO_PARAMS are # mirrored between litellm_params and model_info by Deployment.__init__, so the # clear propagates to both blobs. if updated_patch.litellm_params: for field in updated_patch.litellm_params.model_fields_set: if field in SPECIAL_MODEL_INFO_PARAMS and getattr(updated_patch.litellm_params, field) is None: merged_litellm_params.pop(field, None) merged_model_info.pop(field, None) if updated_patch.model_info: for field in updated_patch.model_info.model_fields_set: if field in SPECIAL_MODEL_INFO_PARAMS and getattr(updated_patch.model_info, field) is None: merged_model_info.pop(field, None) merged_litellm_params.pop(field, None) for field in _explicitly_cleared_ptu_fields(updated_patch.model_info): merged_model_info.pop(field, None) _validate_ptu_model_info(merged_model_info) ptu_pricing, ptu_released = _ptu_pricing_delta( stored_model_info=db_model.model_info.model_dump(exclude_none=True) if db_model.model_info else _EMPTY_MODEL_INFO, model_info=merged_model_info, litellm_params=merged_litellm_params, patch=updated_patch, ) merged_model_info.update(ptu_pricing) merged_litellm_params.update(ptu_pricing) for field in ptu_released: merged_model_info.pop(field, None) merged_litellm_params.pop(field, None) # convert to prisma compatible format for key, value in merged_model_info.items(): if isinstance(value, datetime.datetime): merged_model_info[key] = value.isoformat() prisma_compatible_model_dict: Final = PrismaCompatibleUpdateDBModel( model_name=merged_model_name, litellm_params=json.dumps(merged_litellm_params), model_info=json.dumps(merged_model_info), ) if updated_patch.blocked is not None: prisma_compatible_model_dict["blocked"] = updated_patch.blocked return prisma_compatible_model_dict @router.patch( "/model/{model_id}/update", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def patch_model( model_id: str, # Get model_id from path parameter patch_data: updateDeployment, # Create a specific schema for PATCH operations user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ PATCH Endpoint for partial model updates. Only updates the fields specified in the request while preserving other existing values. Follows proper PATCH semantics by only modifying provided fields. Args: model_id: The ID of the model to update patch_data: The fields to update and their new values user_api_key_dict: User authentication information Returns: Updated model information Raises: ProxyException: For various error conditions including authentication and database errors """ from litellm.proxy.proxy_server import ( litellm_proxy_admin_name, llm_router, premium_user, prisma_client, store_model_in_db, ) try: if prisma_client is None: raise HTTPException( status_code=500, detail={"error": CommonProxyErrors.db_not_connected_error.value}, ) # Verify model exists and is stored in DB if not store_model_in_db: raise ProxyException( message="Model updates only supported for DB-stored models", type=ProxyErrorTypes.validation_error.value, code=status.HTTP_400_BAD_REQUEST, param=None, ) # Fetch existing model db_model: Final = await get_db_model(model_id=model_id, prisma_client=prisma_client) if db_model is None: # Check if model exists in config but not DB if llm_router and llm_router.get_deployment(model_id=model_id) is not None: raise ProxyException( message="Cannot edit config-based model. Store model in DB via /model/new first.", type=ProxyErrorTypes.validation_error.value, code=status.HTTP_400_BAD_REQUEST, param=None, ) raise ProxyException( message=f"Model {model_id} not found on proxy.", type=ProxyErrorTypes.not_found_error, code=status.HTTP_404_NOT_FOUND, param=None, ) await ModelManagementAuthChecks.can_user_make_model_call( model_params=db_model, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, premium_user=premium_user, ) # Pause/resume (`blocked`) is a proxy-admin-only privilege. Team admins # passed the auth check above for team-scoped models, but they must not # be able to unblock (or block) a model their proxy admin has paused. if patch_data.blocked is not None and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: raise ProxyException( message="Only proxy admins can change a model's blocked flag.", type=ProxyErrorTypes.auth_error.value, code=status.HTTP_403_FORBIDDEN, param="blocked", ) _raise_on_strategy_router_write_violation( incoming_params=patch_data.litellm_params, existing_params=db_model.litellm_params, ) # Handle team model updates with proper alias management update_data: Final = await _update_team_model_in_db( db_model=db_model, patch_data=patch_data, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, ) # Add metadata about update update_data["updated_by"] = user_api_key_dict.user_id or litellm_proxy_admin_name update_data["updated_at"] = cast(str, get_utc_datetime()) # Perform partial update updated_model: Final = await _proxy_model_table(prisma_client).update( where={"model_id": model_id}, data=update_data, ) if updated_model is None: raise ProxyException( message=f"Model {model_id} not found on proxy.", type=ProxyErrorTypes.not_found_error, code=status.HTTP_404_NOT_FOUND, param=None, ) # Clear cache and reload models (uses config setting or defaults to preserving config models for DB updates) live_before_reload: Final = live_model_ids_snapshot() reload_outcome: Final = await clear_cache() ## CREATE AUDIT LOG ## asyncio.create_task( create_object_audit_log( object_id=model_id, action="updated", user_api_key_dict=user_api_key_dict, table_name=LitellmTableNames.PROXY_MODEL_TABLE_NAME, before_value=db_model.model_dump_json(exclude_none=True), after_value=updated_model.model_dump_json(exclude_none=True), litellm_changed_by=user_api_key_dict.user_id, litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME, ) ) raise_if_reload_degraded_serving( before=live_before_reload, written_models=[(model_id, getattr(updated_model, "model_info", None))], action="update", still_desired=reload_outcome.still_desired, live_after=reload_outcome.live_after, ) return updated_model except Exception as e: verbose_proxy_logger.exception("Error in patch_model: %s", e) if isinstance(e, (HTTPException, ProxyException)): raise e raise ProxyException( message=f"Error updating model: {e}", type=ProxyErrorTypes.internal_server_error, code=status.HTTP_500_INTERNAL_SERVER_ERROR, param=None, ) async def _set_model_blocked_status( data: BlockModelRequest, user_api_key_dict: UserAPIKeyAuth, blocked: bool, action: Literal["blocked", "unblocked"], litellm_changed_by: str | None, ) -> LiteLLM_ProxyModelTable | None: from litellm.proxy.proxy_server import ( litellm_proxy_admin_name, llm_router, prisma_client, store_model_in_db, ) try: if prisma_client is None: raise HTTPException( status_code=500, detail={"error": CommonProxyErrors.db_not_connected_error.value}, ) if store_model_in_db is not True: raise ProxyException( message="Model updates only supported for DB-stored models", type=ProxyErrorTypes.validation_error.value, code=status.HTTP_400_BAD_REQUEST, param=None, ) if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: raise ProxyException( message="Only proxy admins can change a model's blocked flag.", type=ProxyErrorTypes.auth_error.value, code=status.HTTP_403_FORBIDDEN, param="blocked", ) db_model: Final = await get_db_model( model_id=data.model_id, prisma_client=prisma_client, ) if db_model is None: if llm_router and llm_router.get_deployment(model_id=data.model_id) is not None: raise ProxyException( message="Cannot edit config-based model. Store model in DB via /model/new first.", type=ProxyErrorTypes.validation_error.value, code=status.HTTP_400_BAD_REQUEST, param=None, ) raise ProxyException( message=f"Model {data.model_id} not found on proxy.", type=ProxyErrorTypes.not_found_error, code=status.HTTP_404_NOT_FOUND, param=None, ) updated_model: Final = await ModelRepository(prisma_client).table.update( where={"model_id": data.model_id}, data={ "blocked": blocked, "updated_by": user_api_key_dict.user_id or litellm_proxy_admin_name, "updated_at": cast(str, get_utc_datetime()), }, ) live_before_reload: Final = live_model_ids_snapshot() reload_outcome: Final = await clear_cache() asyncio.create_task( create_object_audit_log( object_id=data.model_id, action=action, user_api_key_dict=user_api_key_dict, table_name=LitellmTableNames.PROXY_MODEL_TABLE_NAME, before_value=db_model.model_dump_json(exclude_none=True), after_value=( updated_model.model_dump_json(exclude_none=True) if isinstance(updated_model, BaseModel) else None ), litellm_changed_by=litellm_changed_by, litellm_proxy_admin_name=litellm_proxy_admin_name, ) ) raise_if_reload_degraded_serving( before=live_before_reload, written_models=[(data.model_id, getattr(updated_model, "model_info", None))], action=action, still_desired=reload_outcome.still_desired, live_after=reload_outcome.live_after, ) return updated_model # pyright: ignore[reportReturnType] # prisma row, coerced by this route's response_model except Exception as e: verbose_proxy_logger.exception("Error in model %s: %s", action, e) if isinstance(e, (HTTPException, ProxyException)): raise e raise ProxyException( message=f"Error updating model blocked status: {e}", type=ProxyErrorTypes.internal_server_error, code=status.HTTP_500_INTERNAL_SERVER_ERROR, param=None, ) @router.post( "/model/block", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def block_model( data: BlockModelRequest, http_request: Request, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), litellm_changed_by: str | None = Header( None, description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability", ), ) -> LiteLLM_ProxyModelTable | None: """ Block a DB-stored model deployment from serving requests. Parameters: - model_id: str - The model deployment id to block. """ return await _set_model_blocked_status( data=data, user_api_key_dict=user_api_key_dict, blocked=True, action="blocked", litellm_changed_by=litellm_changed_by, ) @router.post( "/model/unblock", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def unblock_model( data: BlockModelRequest, http_request: Request, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), litellm_changed_by: str | None = Header( None, description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability", ), ) -> LiteLLM_ProxyModelTable | None: """ Unblock a DB-stored model deployment so it can serve requests again. Parameters: - model_id: str - The model deployment id to unblock. """ return await _set_model_blocked_status( data=data, user_api_key_dict=user_api_key_dict, blocked=False, action="unblocked", litellm_changed_by=litellm_changed_by, ) ################################# Helper Functions ################################# #################################################################################### #################################################################################### #################################################################################### async def _add_model_to_db( model_params: Deployment, user_api_key_dict: UserAPIKeyAuth, prisma_client: PrismaClient, new_encryption_key: str | None = None, should_create_model_in_db: bool = True, ) -> "prisma_models.LiteLLM_ProxyModelTable | LiteLLM_ProxyModelTable | None": # encrypt litellm params # _litellm_params_dict: Final = model_params.litellm_params.dict(exclude_none=True) _original_litellm_model_name: Final = model_params.litellm_params.model for k, v in _litellm_params_dict.items(): encrypted_value = encrypt_value_helper(value=v, new_encryption_key=new_encryption_key) model_params.litellm_params[k] = encrypted_value _data: Final[dict] = { "model_id": model_params.model_info.id, "model_name": model_params.model_name, "litellm_params": model_params.litellm_params.model_dump_json(exclude_none=True), "model_info": model_params.model_info.model_dump_json(exclude_none=True), "created_by": user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME, "updated_by": user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME, } if model_params.model_info.id is not None: _data["model_id"] = model_params.model_info.id _create_data: Final = cast("Mapping[str, object]", _data) # cast-ok: str-keyed json payload built just above if should_create_model_in_db: model_response = await ModelRepository(prisma_client).table.create(data=_create_data) else: model_response = LiteLLM_ProxyModelTable(**_data) return model_response async def _add_team_model_to_db( model_params: Deployment, user_api_key_dict: UserAPIKeyAuth, prisma_client: PrismaClient, ) -> "prisma_models.LiteLLM_ProxyModelTable | LiteLLM_ProxyModelTable | None": """ If 'team_id' is provided, - generate a unique 'model_name' for the model (e.g. 'model_name_{team_id}_{uuid}) - store the model in the db with the unique 'model_name' - add the public model name to the team's allowed models list """ _team_id: Final = model_params.model_info.team_id if _team_id is None: return None # Capture the original public name FIRST, before any mutations original_model_name: Final = model_params.model_name # Set team_public_model_name in model_info using the captured original_model_name # This must happen BEFORE mutating model_params.model_name so _add_model_to_db # serializes the correct team_public_model_name (not the internal UUID name) if original_model_name: model_params.model_info.team_public_model_name = original_model_name # Generate and assign unique internal model_name LAST # (after team_public_model_name is safely stored) unique_model_name: Final = f"model_name_{_team_id}_{uuid.uuid4()}" model_params.model_name = unique_model_name ## CREATE MODEL IN DB ## model_response: Final = await _add_model_to_db( model_params=model_params, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, ) if original_model_name: await team_model_add( data=TeamModelAddRequest( team_id=_team_id, models=[original_model_name], ), http_request=Request(scope={"type": "http"}), user_api_key_dict=user_api_key_dict, ) return model_response async def _update_team_model_in_db( db_model: Deployment, patch_data: updateDeployment, user_api_key_dict: UserAPIKeyAuth, prisma_client: PrismaClient, ) -> PrismaCompatibleUpdateDBModel: """ Handle team model updates with proper alias management. If patch_data contains a team_id: - Creates unique internal model_name and team alias - Adds model to team object - Preserves team_public_model_name for external reference """ # Validate team_id if present in patch_data from litellm.proxy.proxy_server import premium_user await ModelManagementAuthChecks.allow_team_model_action( model_params=patch_data, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, premium_user=premium_user, ) # Validated before any write, beside the premium check the create path already runs # here. The team ACL is updated below and autocommits, so a validator that raises # further down would leave the team mutated and the deployment row never written. # # The merged view is what gets stored, so that is what has to satisfy the invariants. # Validating the patch alone rejected a partial edit of an already valid deployment: # raising the rate on a configured model carries no ptu_effective_from, which the # stored row supplies. if patch_data.model_info is not None: _raise_if_ptu_cost_attribution_disabled(patch_data.model_info.model_dump(exclude_none=True)) _validate_ptu_model_info(_merged_ptu_model_info(db_model=db_model, patch_data=patch_data)) _raise_if_ptu_deployment_is_priced( model_info=_merged_ptu_model_info(db_model=db_model, patch_data=patch_data), supplied=( patch_data.litellm_params.model_dump(exclude_none=True) if patch_data.litellm_params else _EMPTY_MODEL_INFO ), ) patch_team_id: Final = patch_data.model_info.team_id if patch_data.model_info else None # No team_id in patch, proceed with standard update if patch_team_id is None: return update_db_model(db_model=db_model, updated_patch=patch_data) # Determine public model name public_model_name: Final = _get_public_model_name( patch_data=patch_data, db_model=db_model, ) # Ensure model_info exists and set team_public_model_name if patch_data.model_info is None: from litellm.types.router import ModelInfo patch_data.model_info = ModelInfo() patch_data.model_info.team_public_model_name = public_model_name # Check if team assignment is new or changed db_team_id: Final = db_model.model_info.team_id if db_model.model_info else None is_new_team_assignment: Final = db_team_id != patch_team_id if is_new_team_assignment: await _setup_new_team_model_assignment( team_id=patch_team_id, public_model_name=public_model_name, patch_data=patch_data, user_api_key_dict=user_api_key_dict, ) else: await _update_existing_team_model_assignment( team_id=patch_team_id, public_model_name=public_model_name, db_model=db_model, patch_data=patch_data, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, ) return update_db_model(db_model=db_model, updated_patch=patch_data) def _get_public_model_name( patch_data: updateDeployment, db_model: Deployment, ) -> str: """Determine the public model name from patch or existing model. The top-level ``model_name`` is the rename channel. For team-scoped rows the DB ``model_name`` column holds an internal routing key (``model_name_{team_id}_{uuid}``), and ``/model/info`` historically leaked it into the dashboard edit form, so a non-rename save (e.g. a TPM tweak) would PATCH the internal name and the update path would treat it as a rename -- overwriting ``team_public_model_name`` and rewriting the team ACL (see issue #28382). Guard against that by ignoring an incoming ``model_name`` that matches the internal shape, or is a no-op against the current DB column. Anything else is a genuine rename and wins. We deliberately do NOT read ``patch_data.model_info.team_public_model_name``: the dashboard passes the existing ``model_info`` blob through untouched on a rename, so honoring it would return the OLD public name and silently drop the rename. Precedence (highest first): 1. patch_data.model_name -- a genuine rename: not internal-shape and not a no-op against db_model.model_name. 2. db_model.model_info.team_public_model_name -- existing public name. 3. db_model.model_name -- last-resort fallback for legacy rows. """ team_id: Final = (patch_data.model_info.team_id if patch_data.model_info else None) or ( db_model.model_info.team_id if db_model.model_info else None ) def _is_internal_shape(name: str | None) -> bool: if team_id is None or not name: return False return name.startswith(f"model_name_{team_id}_") incoming: Final = patch_data.model_name if incoming and not _is_internal_shape(incoming) and incoming != db_model.model_name: return incoming if db_model.model_info and db_model.model_info.team_public_model_name: return db_model.model_info.team_public_model_name return db_model.model_name async def _setup_new_team_model_assignment( team_id: str, public_model_name: str, patch_data: updateDeployment, user_api_key_dict: UserAPIKeyAuth, ) -> None: """Set up a new team model with unique name and team membership.""" unique_model_name: Final = f"model_name_{team_id}_{uuid.uuid4()}" patch_data.model_name = unique_model_name await team_model_add( data=TeamModelAddRequest( team_id=team_id, models=[public_model_name], ), http_request=Request(scope={"type": "http"}), user_api_key_dict=user_api_key_dict, ) async def _get_team_deployments( team_id: str, prisma_client: PrismaClient, table: _ProxyModelTable | None = None ) -> Sequence[_ProxyModelRow]: """ Fetch all deployments for a given team_id from the database. Centralizes team deployment queries to ensure consistent filtering and error handling. This is the established helper pattern for team deployment DB access in this module. Note: prisma-client-py 0.11.0 does not support JSON path filtering, so we filter by the model_name prefix (team models use "model_name_{team_id}_*") and confirm team_id in model_info with Python-side filtering. Pass ``table`` (a transaction's proxy-model table) to run the read inside an existing transaction. """ prefix: Final = f"model_name_{team_id}_" table = table or _proxy_model_table(prisma_client) response: Final = await table.find_many( where={ "model_name": {"startswith": prefix}, } ) if not response: return [] # Confirm team_id in model_info (defensive check) result: Final = [] for row in response: model_info = model_info_as_mapping(row.model_info) if model_info is not None and model_info.get("team_id") == team_id: result.append(row) return result async def delete_team_models( team_ids: list[str], prisma_client: PrismaClient, llm_router: Router | None, ) -> list[str]: """ Delete every BYOK model owned by the given teams, from the DB and the router. The DB rows are removed inside a single transaction, so deletion is atomic across all team_ids. Each team's rows are deleted by the exact model_ids read in the same transaction, which keeps the deleted set identical to the set handed to the router. The router is synced only after the transaction commits, so a rollback can never leave a deployment live in the router without its row. Returns the model_ids that were deleted. """ deleted_model_ids: Final[list[str]] = [] async with prisma_client.db.tx() as tx_ctx: tx: Final[_TxModelTables] = tx_ctx for team_id in team_ids: rows = await _get_team_deployments(team_id, prisma_client, table=tx.litellm_proxymodeltable) model_ids = [row.model_id for row in rows] if model_ids: await tx.litellm_proxymodeltable.delete_many(where={"model_id": {"in": model_ids}}) deleted_model_ids.extend(model_ids) if deleted_model_ids: await publish_config_change(redis_cache=coordination_redis_cache(), object_type="litellm_proxymodeltable") # Under MODEL_RECONCILE_LOCK, for the same reason as delete_model: the rows are # gone, but a reconcile holding a pre-delete snapshot would upsert these ids back # onto this pod. The lock orders the eviction after any in-flight reconcile. if llm_router is not None: from litellm.proxy.proxy_server import MODEL_RECONCILE_LOCK async with MODEL_RECONCILE_LOCK: for model_id in deleted_model_ids: llm_router.delete_deployment(id=model_id) return deleted_model_ids async def _get_team_public_model_names( team_id: str, prisma_client: PrismaClient, ) -> set[str]: """ Public model names currently backed by a deployment in the team. Called on delete (after the deployment row is removed) so a public name that is load-balanced across several deployments stays in team.models while a replica still serves it. """ deployments: Final = await _get_team_deployments(team_id, prisma_client) public_names: Final[set[str]] = set() for row in deployments: model_info = model_info_as_mapping(row.model_info) if model_info is not None: public_name = model_info.get("team_public_model_name") if public_name: public_names.add(public_name) return public_names async def _remove_unbacked_team_models( model_params: Deployment, prisma_client: PrismaClient, user_api_key_cache: UserApiKeyCache, proxy_logging_obj: ProxyLogging, llm_router: Router | None = None, ) -> None: """ Strip a deleted team model's public name(s) from team.models and refresh the cache. Must be called after the deployment row is deleted: a public name is removed only when no remaining team deployment still backs it, so a load-balanced replica isn't revoked while siblings serve it, and concurrent deletes can't leave a ghost. Legacy team models (created before team_public_model_name existed) store a ``{public_name: "model_name_{team_id}_{uuid}"}`` entry in the team's model_aliases, so the alias scan runs for every team model; skipping it for internal-shaped names left stale aliases that rewrote requests to deployments that no longer exist. Aliases are scrubbed only when the deleted deployment's name no longer resolves in the router, so deleting one replica of a load-balanced group never breaks aliases that still route to the surviving replicas (in any team). A public name that still resolves to a live router deployment (e.g. a gateway-level model group shared with the team) is kept in team.models, so deleting a per-team duplicate does not revoke the team's access to the shared deployment. """ team_id: Final = model_params.model_info.team_id if team_id is None: return deleted_name_still_served: Final = ( llm_router is not None and model_params.model_name in llm_router.model_name_to_deployment_indices ) removed_model_aliases: Final[list[tuple[str, str]]] = ( [] if deleted_name_still_served else await delete_team_model_alias( public_model_name=model_params.model_name, prisma_client=prisma_client, ) ) removed_alias_names: Final = {alias for alias_team_id, alias in removed_model_aliases if alias_team_id == team_id} candidate_names: Final = ( removed_alias_names | {model_params.model_info.team_public_model_name} if model_params.model_info.team_public_model_name is not None else removed_alias_names ) if not candidate_names: return team_backed_names: Final = await _get_team_public_model_names(team_id=team_id, prisma_client=prisma_client) router_served_names: Final = ( frozenset(name for name in candidate_names if name in llm_router.model_name_to_deployment_indices) if llm_router is not None else frozenset() ) names_to_remove: Final = candidate_names - team_backed_names - router_served_names if not names_to_remove: return existing_team_row: Final = await _db_team_table(prisma_client).find_unique(where={"team_id": team_id}) if existing_team_row is None: return updated_team_row: Final[LiteLLM_TeamTable] = await _db_team_table(prisma_client).update( where={"team_id": team_id}, data={"models": [model for model in existing_team_row.models if model not in names_to_remove]}, include={"object_permission": True}, ) await _refresh_cached_team( team_row=updated_team_row, user_api_key_cache=user_api_key_cache, proxy_logging_obj=proxy_logging_obj, ) async def _update_existing_team_model_assignment( team_id: str, public_model_name: str, db_model: Deployment, patch_data: updateDeployment, user_api_key_dict: UserAPIKeyAuth, prisma_client: PrismaClient | None, ) -> None: """Update an existing team model if the public name changed. Note on DB scan: Prisma's JSON filtering does not support compound AND conditions across multiple JSON paths, so we fetch all deployments for the team and filter team_public_model_name in Python. For teams with many deployments this scan grows linearly; if team deployment counts become large this should be revisited. """ def _get_team_public_model_name( model_info: object, ) -> str | None: parsed: Final = model_info_as_mapping(model_info) if parsed is None: return None value: Final = parsed.get("team_public_model_name") return value if isinstance(value, str) else None old_public_name: Final = db_model.model_info.team_public_model_name if db_model.model_info else None if old_public_name and public_model_name != old_public_name: # Clear user-supplied public name from patch before any early return so the # caller does not overwrite the internal UUID-based model_name in the DB. patch_data.model_name = None if prisma_client is None: verbose_proxy_logger.warning( "prisma_client not initialized; skipping public name update entirely to avoid orphaned entries" ) return # Query DB for all team deployments to check for sibling deployments team_deployments: Final = await _get_team_deployments(team_id, prisma_client) other_deployments_with_old_name: Final = [ d for d in team_deployments if d.model_name != db_model.model_name and _get_team_public_model_name(d.model_info) == old_public_name ] # Add new name first, then delete old name to prevent access loss on partial failure await team_model_add( data=TeamModelAddRequest( team_id=team_id, models=[public_model_name], ), http_request=Request(scope={"type": "http"}), user_api_key_dict=user_api_key_dict, ) if not other_deployments_with_old_name: await team_model_delete( data=TeamModelDeleteRequest( team_id=team_id, models=[old_public_name], ), http_request=Request(scope={"type": "http"}), user_api_key_dict=user_api_key_dict, ) elif not old_public_name and public_model_name: # First-time assignment of public name on an existing team deployment: # ensure the team's models list is updated so team routing can resolve it. await team_model_add( data=TeamModelAddRequest( team_id=team_id, models=[public_model_name], ), http_request=Request(scope={"type": "http"}), user_api_key_dict=user_api_key_dict, ) # else: old_public_name == public_model_name (no rename needed) # No team_model_add/delete calls required; public name is already registered # Always clear patch_data.model_name to prevent caller from overwriting # the internal UUID-based model_name in the DB with the user-supplied public name patch_data.model_name = None class ModelManagementAuthChecks: """ Common auth checks for model management endpoints """ @staticmethod def can_user_make_team_model_call( team_id: str, user_api_key_dict: UserAPIKeyAuth, team_obj: LiteLLM_TeamTable | None = None, premium_user: bool = False, ) -> Literal[True]: if premium_user is False: raise HTTPException( status_code=403, detail={"error": CommonProxyErrors.not_premium_user.value}, ) if user_api_key_dict.user_role and user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN: return True elif team_obj is None or not _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj): raise HTTPException( status_code=403, detail={ "error": f"Team ID={team_id} does not match the API key's team ID={user_api_key_dict.team_id}, OR you are not the admin for this team. Check `/user/info` to verify your team admin status." }, ) return True @staticmethod async def allow_team_model_action( model_params: Deployment | updateDeployment, user_api_key_dict: UserAPIKeyAuth, prisma_client: PrismaClient, premium_user: bool, ) -> Literal[True]: if model_params.model_info is None or model_params.model_info.team_id is None: return True if model_params.model_info.team_id is not None and premium_user is not True: raise HTTPException( status_code=403, detail={"error": CommonProxyErrors.not_premium_user.value}, ) _existing_team_row: Final = await _repo_team_table(prisma_client).find_unique( where={"team_id": model_params.model_info.team_id} ) if _existing_team_row is None: raise HTTPException( status_code=400, detail={"error": f"Team id={model_params.model_info.team_id} does not exist in db"}, ) existing_team_row: Final = LiteLLM_TeamTable.model_validate(_existing_team_row.model_dump()) ModelManagementAuthChecks.can_user_make_team_model_call( team_id=model_params.model_info.team_id, user_api_key_dict=user_api_key_dict, team_obj=existing_team_row, premium_user=premium_user, ) return True @staticmethod async def can_user_make_model_call( model_params: Deployment, user_api_key_dict: UserAPIKeyAuth, prisma_client: PrismaClient, premium_user: bool, allow_missing_team: bool = False, ) -> Literal[True]: ## Check team model auth if model_params.model_info is not None and model_params.model_info.team_id is not None: team_obj_row: Final = await _repo_team_table(prisma_client).find_unique( where={"team_id": model_params.model_info.team_id} ) if team_obj_row is None: # The team was deleted. Callers that opt in (e.g. model deletion) may # act on the orphaned model, but only as a proxy admin -- without the # team there is no team-admin membership left to verify. if allow_missing_team: if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN: return True raise HTTPException( status_code=403, detail={"error": "Only a proxy admin can delete a model whose team has been deleted."}, ) raise HTTPException( status_code=400, detail={"error": f"Team id={model_params.model_info.team_id} does not exist in db"}, ) team_obj: Final = LiteLLM_TeamTable.model_validate(team_obj_row.model_dump()) return ModelManagementAuthChecks.can_user_make_team_model_call( team_id=model_params.model_info.team_id, user_api_key_dict=user_api_key_dict, team_obj=team_obj, premium_user=premium_user, ) ## Check non-team model auth elif user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: raise HTTPException( status_code=403, detail={ "error": f"User does not have permission to make this model call. Your role={user_api_key_dict.user_role}. You can only make model calls if you are a PROXY_ADMIN or if you are a team admin, by specifying a team_id in the model_info." }, ) else: return True return True #### [BETA] - This is a beta endpoint, format might change based on user feedback. - https://github.com/BerriAI/litellm/issues/964 @router.post( "/model/delete", description="Allows deleting models in the model list in the config.yaml", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def delete_model( model_info: ModelInfoDelete, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): from litellm.proxy.proxy_server import llm_router try: """ [BETA] - This is a beta endpoint, format might change based on user feedback. - https://github.com/BerriAI/litellm/issues/964 - Check if id in db - Delete """ from litellm.proxy.proxy_server import ( MODEL_RECONCILE_LOCK, llm_router, premium_user, prisma_client, proxy_logging_obj, store_model_in_db, user_api_key_cache, ) if prisma_client is None: raise HTTPException( status_code=500, detail={ "error": "No DB Connected. Here's how to do it - https://docs.litellm.ai/docs/proxy/virtual_keys" }, ) model_in_db: Final = await _proxy_model_table(prisma_client).find_unique(where={"model_id": model_info.id}) if model_in_db is None: raise HTTPException( status_code=400, detail={"error": f"Model with id={model_info.id} not found in db"}, ) model_params: Final = Deployment(**model_in_db.model_dump()) await ModelManagementAuthChecks.can_user_make_model_call( model_params=model_params, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, premium_user=premium_user, allow_missing_team=True, ) # update DB if store_model_in_db is True: """ - store model_list in db - store keys separately """ # encrypt litellm params # result: Final = await _proxy_model_table(prisma_client).delete(where={"model_id": model_info.id}) if result is None: raise HTTPException( status_code=400, detail={"error": f"Model with id={model_info.id} not found in db"}, ) ## DELETE FROM ROUTER ## # Under MODEL_RECONCILE_LOCK. The db row is already gone, but a reconcile # that snapshotted the db BEFORE that delete still lists this id as desired, # and its _add_deployment upserts the deployment straight back -- leaving # this pod serving a model the database no longer has, until the next # reconcile. Taking the lock orders this eviction after any such in-flight # reconcile's re-add, so the eviction is the last word. if llm_router is not None: async with MODEL_RECONCILE_LOCK: llm_router.delete_deployment(id=model_info.id) # Runs after the row delete so the sibling check sees post-delete state. if model_params.model_info.team_id is not None: await _remove_unbacked_team_models( model_params=model_params, prisma_client=prisma_client, user_api_key_cache=user_api_key_cache, proxy_logging_obj=proxy_logging_obj, llm_router=llm_router, ) ## CREATE AUDIT LOG ## asyncio.create_task( create_object_audit_log( object_id=model_info.id, action="deleted", user_api_key_dict=user_api_key_dict, table_name=LitellmTableNames.PROXY_MODEL_TABLE_NAME, before_value=result.model_dump_json(exclude_none=True), after_value=None, litellm_changed_by=user_api_key_dict.user_id, litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME, ) ) return {"message": f"Model: {result.model_id} deleted successfully"} else: raise HTTPException( status_code=500, detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."}, ) except Exception as e: verbose_proxy_logger.exception("Failed to delete model. Due to error - %s", e) if isinstance(e, HTTPException): raise ProxyException( message=getattr(e, "detail", f"Authentication Error({e})"), type=ProxyErrorTypes.auth_error, param=getattr(e, "param", "None"), code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST), ) elif isinstance(e, ProxyException): raise e raise ProxyException( message="Authentication Error, " + str(e), type=ProxyErrorTypes.auth_error, param=getattr(e, "param", "None"), code=status.HTTP_400_BAD_REQUEST, ) async def delete_team_model_alias( public_model_name: str, prisma_client: PrismaClient, ) -> list[tuple[str, str]]: """ Delete a team model alias Iterate through all team model aliases and delete the one that matches the model_id Returns: - List of team id + model alias pairs that were removed """ team_model_aliases: Final = await _model_alias_table(prisma_client).find_many(include={"team": True}) tasks: Final = [] removed_model_aliases: Final[list[tuple[str, str]]] = [] for team_model_alias in team_model_aliases: model_aliases = cast( # cast-ok: prisma types Json columns as `str`; the driver hands back the parsed dict "dict[str, str]", team_model_alias.model_aliases ) id = team_model_alias.id if public_model_name in model_aliases.values(): key = list(model_aliases.keys())[list(model_aliases.values()).index(public_model_name)] if team_model_alias.team is not None: removed_model_aliases.append((team_model_alias.team.team_id, key)) del model_aliases[key] tasks.append( _model_alias_table(prisma_client).update( where={"id": id}, data={"model_aliases": json.dumps(model_aliases)}, ) ) await asyncio.gather(*tasks) return removed_model_aliases #### [BETA] - This is a beta endpoint, format might change based on user feedback. - https://github.com/BerriAI/litellm/issues/964 @router.post( "/model/new", description="Allows adding new models to the model list in the config.yaml", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def add_new_model( model_params: Deployment, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ Add a new model to the proxy. Parameters: - model_name: str - The name users will use to call this model (required) - litellm_params: dict - LiteLLM-specific parameters (required) - model: str - The actual model identifier, e.g., "azure/my-deployment-name" (required - this is the only required field in litellm_params) - api_key: str - API key for the provider (optional) - api_base: str - API base URL (optional) - Other optional params: api_version, timeout, max_retries, etc. - model_info: dict - Additional model metadata returned in /v1/model/info (optional) Example curl: ```bash curl -L -X POST 'http://0.0.0.0:4000/model/new' \ -H 'Authorization: Bearer LITELLM_VIRTUAL_KEY' \ -H 'Content-Type: application/json' \ -d '{ "model_name": "my-azure-model", "litellm_params": { "model": "azure/my-deployment-name", "api_key": "my-azure-api-key", "api_base": "https://my-endpoint.openai.azure.com" }, "model_info": { "my_custom_key": "my_custom_value" } }' ``` Returns: - The created model entry with model_id """ from litellm.proxy.proxy_server import ( general_settings, premium_user, prisma_client, proxy_config, proxy_logging_obj, store_model_in_db, ) try: if prisma_client is None: raise HTTPException( status_code=500, detail={ "error": "No DB Connected. Here's how to do it - https://docs.litellm.ai/docs/proxy/virtual_keys" }, ) ## Auth check await ModelManagementAuthChecks.can_user_make_model_call( model_params=model_params, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, premium_user=premium_user, ) _raise_on_strategy_router_write_violation( incoming_params=model_params.litellm_params, existing_params=None, ) _raise_if_rate_limits_required_but_missing( litellm_params=model_params.litellm_params, enforced=bool(general_settings.get(ENFORCE_RPM_TPM_ON_MODEL_ADD_SETTING, False)), ) model_response: prisma_models.LiteLLM_ProxyModelTable | LiteLLM_ProxyModelTable | None = None # update DB incoming_model_info: Final = model_params.model_info.model_dump(exclude_none=True) _raise_if_ptu_cost_attribution_disabled(incoming_model_info) _validate_ptu_model_info(incoming_model_info) priced_model_params: Final = _ptu_priced_deployment(model_params) if store_model_in_db is True: """ - store model_list in db - store keys separately """ live_before_reload: Final = live_model_ids_snapshot() reload_outcome: ReconcileOutcome = ReconcileOutcome(still_desired=None, live_after=None) try: _original_litellm_model_name: Final = model_params.model_name if model_params.model_info.team_id is None: model_response = await _add_model_to_db( model_params=priced_model_params, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, ) else: model_response = await _add_team_model_to_db( model_params=priced_model_params, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, ) reload_outcome = await proxy_config.add_deployment( prisma_client=prisma_client, proxy_logging_obj=proxy_logging_obj ) # don't let failed slack alert block the /model/new response _alerting: Final = general_settings.get("alerting", []) or [] if "slack" in _alerting: # send notification - new model added await proxy_logging_obj.slack_alerting_instance.model_added_alert( model_name=priced_model_params.model_name, litellm_model_name=_original_litellm_model_name, passed_model_info=priced_model_params.model_info, ) except Exception as e: verbose_proxy_logger.exception("Exception in add_new_model: %s", e) else: raise HTTPException( status_code=500, detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."}, ) if model_response is None: raise HTTPException( status_code=500, detail={"error": "Failed to add model to db. Check your server logs for more details."}, ) ## CREATE AUDIT LOG ## asyncio.create_task( create_object_audit_log( object_id=model_response.model_id, action="created", user_api_key_dict=user_api_key_dict, table_name=LitellmTableNames.PROXY_MODEL_TABLE_NAME, before_value=None, after_value=( model_response.model_dump_json(exclude_none=True) if isinstance(model_response, BaseModel) else None ), litellm_changed_by=user_api_key_dict.user_id, litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME, ) ) raise_if_reload_degraded_serving( before=live_before_reload, written_models=[(model_response.model_id, getattr(model_response, "model_info", None))], action="create", still_desired=reload_outcome.still_desired, live_after=reload_outcome.live_after, ) return model_response except Exception as e: verbose_proxy_logger.exception("litellm.proxy.proxy_server.add_new_model(): Exception occured - %s", e) if isinstance(e, HTTPException): raise ProxyException( message=getattr(e, "detail", f"Authentication Error({e})"), type=ProxyErrorTypes.auth_error, param=getattr(e, "param", "None"), code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST), ) elif isinstance(e, ProxyException): raise e raise ProxyException( message="Authentication Error, " + str(e), type=ProxyErrorTypes.auth_error, param=getattr(e, "param", "None"), code=status.HTTP_400_BAD_REQUEST, ) #### MODEL MANAGEMENT #### @router.post( "/model/update", description="Edit existing model params", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def update_model( model_params: updateDeployment, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ Old endpoint for model update. Makes a PUT request. Use `/model/{model_id}/update` to PATCH the stored model in db. """ from litellm.proxy.proxy_server import ( LITELLM_PROXY_ADMIN_NAME, llm_router, premium_user, prisma_client, store_model_in_db, ) try: if prisma_client is None: raise HTTPException( status_code=500, detail={ "error": "No DB Connected. Here's how to do it - https://docs.litellm.ai/docs/proxy/virtual_keys" }, ) _model_id: str | None = None _model_info: Final[ModelInfo | None] = getattr(model_params, "model_info", None) if _model_info is None: raise Exception("model_info not provided") _model_id = _model_info.id if _model_id is None: raise Exception("model_info.id not provided") _existing_litellm_params = await ModelRepository(prisma_client).table.find_unique(where={"model_id": _model_id}) if _existing_litellm_params is None: if llm_router is not None and llm_router.get_deployment(model_id=_model_id) is not None: raise HTTPException( status_code=400, detail={"error": "Can't edit model. Model in config. Store model in db via `/model/new`. to edit."}, ) else: raise Exception("model not found") deployment: Final = Deployment(**_existing_litellm_params.model_dump()) await ModelManagementAuthChecks.can_user_make_model_call( model_params=deployment, user_api_key_dict=user_api_key_dict, prisma_client=prisma_client, premium_user=premium_user, ) _raise_on_strategy_router_write_violation( incoming_params=model_params.litellm_params, existing_params=deployment.litellm_params, ) # update DB if store_model_in_db is True: existing_model_row: Final = cast( # cast-ok: prisma types Json columns as `str`; the driver parses them "_ExistingModelRow", _existing_litellm_params ) _existing_litellm_params_dict: Final = dict(existing_model_row.litellm_params) if model_params.litellm_params is None: raise Exception("litellm_params not provided") _new_litellm_params_dict: Final = model_params.litellm_params.dict(exclude_none=True) ### ENCRYPT PARAMS ### for k, v in _new_litellm_params_dict.items(): encrypted_value = encrypt_value_helper(value=v) model_params.litellm_params[k] = encrypted_value ### MERGE WITH EXISTING DATA ### merged_dictionary: Final = {} _mp: Final = model_params.litellm_params.dict() for key, value in _mp.items(): if value is not None: merged_dictionary[key] = value elif key in _existing_litellm_params_dict and _existing_litellm_params_dict[key] is not None: merged_dictionary[key] = _existing_litellm_params_dict[key] else: pass _data: Final[dict[str, str]] = { "litellm_params": json.dumps(merged_dictionary), "updated_by": user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME, } model_response: Final = await _proxy_model_table(prisma_client).update( where={"model_id": _model_id}, data=_data, ) # Clear cache and reload models (uses config setting or defaults to preserving config models for DB updates) live_before_reload: Final = live_model_ids_snapshot() reload_outcome: Final = await clear_cache() ## CREATE AUDIT LOG ## asyncio.create_task( create_object_audit_log( object_id=_model_id, action="updated", user_api_key_dict=user_api_key_dict, table_name=LitellmTableNames.PROXY_MODEL_TABLE_NAME, before_value=( existing_model_row.model_dump_json(exclude_none=True) if isinstance(existing_model_row, BaseModel) else None ), after_value=( model_response.model_dump_json(exclude_none=True) if isinstance(model_response, BaseModel) else None ), litellm_changed_by=user_api_key_dict.user_id, litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME, ) ) raise_if_reload_degraded_serving( before=live_before_reload, written_models=[(_model_id, getattr(model_response, "model_info", None))], action="update", still_desired=reload_outcome.still_desired, live_after=reload_outcome.live_after, ) return model_response except Exception as e: verbose_proxy_logger.exception("litellm.proxy.proxy_server.update_model(): Exception occured - %s", e) if isinstance(e, HTTPException): raise ProxyException( message=getattr(e, "detail", f"Authentication Error({e})"), type=ProxyErrorTypes.auth_error, param=getattr(e, "param", "None"), code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST), ) elif isinstance(e, ProxyException): raise e raise ProxyException( message="Authentication Error, " + str(e), type=ProxyErrorTypes.auth_error, param=getattr(e, "param", "None"), code=status.HTTP_400_BAD_REQUEST, ) @router.post( "/model_group/make_public", description="Update which model groups are public", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def update_public_model_groups( request: UpdatePublicModelGroupsRequest, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ Update which model groups are public. This endpoint allows admins to specify which model groups should be publicly accessible. Public model groups are visible via the /public/model_hub endpoint. Args: request: Request containing list of model group names to make public user_api_key_dict: User authentication information Returns: Success message with updated public model groups Raises: ProxyException: For various error conditions including authentication errors """ try: # Update the public model groups import litellm from litellm.proxy.proxy_server import proxy_config, store_model_in_db # Check if user has admin permissions if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: raise HTTPException( status_code=403, detail={ "error": f"Only proxy admins can update public model groups. Your role={user_api_key_dict.user_role}" }, ) # Check if STORE_MODEL_IN_DB is enabled if store_model_in_db is not True: raise HTTPException( status_code=500, detail={"error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."}, ) # Load existing config first (this may overwrite in-memory litellm settings # from DB values via _update_config_from_db), so set the in-memory value AFTER config: Final = await proxy_config.get_config() # Update config with new settings if "litellm_settings" not in config: config["litellm_settings"] = {} config["litellm_settings"]["public_model_groups"] = request.model_groups # Save the updated config await proxy_config.save_config(new_config=config) # Set in-memory value AFTER get_config() and save_config() to avoid # get_config() overwriting with stale DB value litellm.public_model_groups = request.model_groups verbose_proxy_logger.debug( "Updated public model groups to: %s by user: %s", request.model_groups, user_api_key_dict.user_id ) return { "message": "Successfully updated public model groups", "public_model_groups": request.model_groups, "updated_by": user_api_key_dict.user_id, } except Exception as e: verbose_proxy_logger.exception("Error updating public model groups: %s", e) if isinstance(e, HTTPException): raise e raise ProxyException( message=f"Error updating public model groups: {e}", type=ProxyErrorTypes.internal_server_error, code=status.HTTP_500_INTERNAL_SERVER_ERROR, param=None, ) @router.post( "/model_hub/update_useful_links", description="Update useful links", tags=["model management"], dependencies=[Depends(user_api_key_auth)], ) async def update_useful_links( request: UpdateUsefulLinksRequest, user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ Update useful links. """ try: # Update the public model groups import litellm from litellm.proxy.proxy_server import proxy_config # Check if user has admin permissions if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: raise HTTPException( status_code=403, detail={ "error": f"Only proxy admins can update public model groups. Your role={user_api_key_dict.user_role}" }, ) # Load existing config first (this may overwrite in-memory litellm settings # from DB values via _update_config_from_db), so set the in-memory value AFTER config: Final = await proxy_config.get_config() # Update config with new settings if "litellm_settings" not in config: config["litellm_settings"] = {} config["litellm_settings"]["public_model_groups_links"] = request.useful_links # Save the updated config await proxy_config.save_config(new_config=config) # Set in-memory value AFTER get_config() and save_config() to avoid # get_config() overwriting with stale DB value litellm.public_model_groups_links = request.useful_links verbose_proxy_logger.debug( "Updated useful links to: %s by user: %s", request.useful_links, user_api_key_dict.user_id ) return { "message": "Successfully updated useful links", "useful_links": request.useful_links, "updated_by": user_api_key_dict.user_id, } except Exception as e: verbose_proxy_logger.exception("Error updating public model groups: %s", e) if isinstance(e, HTTPException): raise e raise ProxyException( message=f"Error updating public model groups: {e}", type=ProxyErrorTypes.internal_server_error, code=status.HTTP_500_INTERNAL_SERVER_ERROR, param=None, ) def _labeled_tiers_from_query(tier_labels: str | None) -> tuple[tuple[ComplexityTier, str], ...] | None: """Resolve the tier_labels query param into the labeled tiers the rubric is built from. Validated through ComplexityRouterConfig so the editor prefills what the router would send: the same field validators that reject a blank, duplicated, or canonical-name-stealing label on the write path reject it here, rather than this returning a rubric no router could be configured to use. A malformed value is the caller's error, so it surfaces as a 400. None when unset, letting classification_system_prompt apply its own default names. """ if not tier_labels: return None try: return ComplexityRouterConfig(tier_labels=json.loads(tier_labels)).labeled_tiers() except (JSONDecodeError, ValidationError) as e: raise ProxyException( message=f"tier_labels must be a JSON object of tier name to display name: {e}", type=ProxyErrorTypes.bad_request_error, code=status.HTTP_400_BAD_REQUEST, param="tier_labels", ) from e class AutoRouterClassifierPromptPreviewRequest(BaseModel): """A POST rather than query params: classification_prompt is the operator's own text, which must not reach access logs through a URL.""" tier_definitions: tuple[TierDefinition, ...] context_window_size: Annotated[int, Field(ge=0)] = DEFAULT_CLASSIFIER_CONTEXT_WINDOW_SIZE classification_prompt: str | None = None _normalize_prompt = field_validator("classification_prompt")(normalize_classification_prompt) @router.post( "/auto_router/classifier/default_prompt", description="Get the system prompt an auto-router's LLM classifier sends for an edited tier set", tags=["model management"], # mutable-ok: fastapi's decorator signature types tags as a list dependencies=[Depends(user_api_key_auth)], # mutable-ok: fastapi's decorator signature types dependencies as a list ) async def preview_auto_router_classifier_prompt( request: AutoRouterClassifierPromptPreviewRequest, ) -> AutoRouterClassifierDefaultPromptResponse: """ Get the classifier system prompt an edited tier set sends, so the dashboard can show it. Built by the same function the live classifier uses, so the preview cannot drift from what the router sends. Payload validity beyond a renderable definition stays the dry-run's job. """ return AutoRouterClassifierDefaultPromptResponse( system_prompt=custom_tier_classification_prompt( request.tier_definitions, request.classification_prompt, request.context_window_size ) ) @router.get( "/auto_router/classifier/default_prompt", description="Get the built-in system prompt used by an auto-router's LLM classifier", tags=["model management"], # mutable-ok: fastapi's decorator signature types tags as a list dependencies=[Depends(user_api_key_auth)], # mutable-ok: fastapi's decorator signature types dependencies as a list ) async def get_auto_router_classifier_default_prompt( context_window_size: int = DEFAULT_CLASSIFIER_CONTEXT_WINDOW_SIZE, tier_labels: str | None = None, classification_rubric: ClassificationRubric | None = None, ) -> AutoRouterClassifierDefaultPromptResponse: """ Get the classifier system prompt a router would send, so the dashboard can show it. The prompt's closing line depends on whether prior conversation turns are quoted to the classifier, its tier bullets are named by the router's tier_labels, and its calibration examples come from the router's classification rubric, so the caller passes all three to get the text that router would actually send rather than a rubric it does not use. An edited tier set replaces the whole rubric; POST to this path for that prompt, which carries the operator's own instructions and so must not ride in a query string. Parameters: - context_window_size: int - The router's classifier_context_window_size. Defaults to the built-in default. - tier_labels: str | None - The router's tier_labels as a JSON object of canonical tier name to display name, e.g. `{"SIMPLE": "Cheap"}`. Omit or pass an empty object for the default names. - classification_rubric: ClassificationRubric | None - The router's classifier_llm_config.classification_rubric. Omit for the default. """ if context_window_size < 0: raise ProxyException( message="context_window_size must be non-negative", type=ProxyErrorTypes.bad_request_error, code=status.HTTP_400_BAD_REQUEST, param="context_window_size", ) labeled_tiers: Final = _labeled_tiers_from_query(tier_labels) return AutoRouterClassifierDefaultPromptResponse( system_prompt=( classification_system_prompt(context_window_size, classification_rubric=classification_rubric) if labeled_tiers is None else classification_system_prompt( context_window_size, labeled_tiers=labeled_tiers, classification_rubric=classification_rubric ) ) ) def _deduplicate_litellm_router_models(models: list[dict]) -> list[dict]: """ Deduplicate models based on their model_info.id field. Returns a list of unique models keeping only the first occurrence of each model ID. Args: models: List of model dictionaries containing model_info Returns: List of deduplicated model dictionaries """ seen_ids: Final = set() unique_models: Final = [] for model in models: model_id = model.get("model_info", {}).get("id", None) if model_id is not None and model_id not in seen_ids: unique_models.append(model) seen_ids.add(model_id) return unique_models def model_info_as_mapping(model_info: object) -> Mapping[str, object] | None: """A DB row's model_info column arrives as a dict or as its JSON string depending on the query path, and every consumer needs the mapping. Single owner of that parse: returns None when no usable mapping exists (None, an unparseable string, or JSON that is not an object), and callers choose what None means for them.""" if isinstance(model_info, Mapping): return model_info if not isinstance(model_info, str): return None try: parsed: Final = json.loads(model_info) except (TypeError, ValueError): return None return parsed if isinstance(parsed, Mapping) else None def _expects_liveness_on_this_pod(model_info: object) -> bool: from litellm.router import model_info_is_active_for_environment try: return model_info_is_active_for_environment(model_info=model_info_as_mapping(model_info)) except ValueError: return True def live_model_ids_snapshot() -> frozenset[str]: """The ids this pod's router is currently serving, read fresh from the module global because a reload can rebind it. The empirical ground truth every verdict below is computed from; an absent router serves nothing.""" from litellm.proxy.proxy_server import llm_router if llm_router is None: return frozenset() return frozenset(llm_router.get_model_ids()) def reload_serving_verdict( before: frozenset[str], written_models: Sequence[tuple[str, object]], written_must_serve: bool, still_desired: frozenset[str] | None = None, live_after: frozenset[str] | None = None, ) -> tuple[tuple[str, ...], tuple[str, ...]]: """Judge a write-triggered reload by diffing the router's serving state instead of trusting any layer of the reload stack to report its own failure. The full cell matrix, per id: - written, must-serve (the write's purpose is this model's serving state): live now is fine; not live is reported unless the row is deliberately inactive for this pod's LITELLM_ENVIRONMENT; a row whose model_info cannot be read counts as expecting to serve, so its drop is still reported - written, metadata-only (must_not_degrade): live before and gone now is reported; a row that was already not serving stays silent, because its deadness predates this write and blaming it would block unrelated metadata fixes - not written but live before and gone now: collateral degradation of this pod caused by the reload this request triggered (a wholesale re-add failure, or a newly introduced conflict), reported only when the db still wants that id ``still_desired`` is the db + config id set the reload just reconciled against. An id absent from it was deleted on purpose, most often by another pod this one had not yet polled, so the reload dropping it is the reconcile working rather than damage. Without it (no reconcile ran) every drop is reported, which is the safe direction. ``live_after`` is the router's serving state captured by the reload itself, while it still held MODEL_RECONCILE_LOCK. Pass it whenever the caller has it: re-reading the router here instead means sampling it after the lock was released, where the NEXT reconcile's leading wipe (clear_cache un-serves every db model before reloading them) shows up as this reload having dropped them. Falling back to a fresh read is only correct when no reconcile ran and there is nothing to be concurrent with. Returns (written ids violating their obligation, collateral ids no longer served). """ now: Final = live_model_ids_snapshot() if live_after is None else live_after written_ids: Final = frozenset(model_id for model_id, _ in written_models) if written_must_serve: missing = tuple( model_id for model_id, model_info in written_models if model_id not in now and _expects_liveness_on_this_pod(model_info) ) else: missing = tuple(model_id for model_id, _ in written_models if model_id in before and model_id not in now) dropped: Final = before - now - written_ids collateral: Final = tuple(sorted(dropped if still_desired is None else dropped & still_desired)) return (missing, collateral) def raise_if_reload_degraded_serving( before: frozenset[str], written_models: Sequence[tuple[str, object]], action: str, still_desired: frozenset[str] | None = None, live_after: frozenset[str] | None = None, ) -> None: """The caller-visible error this pod's model-write endpoints owe their caller when the model they wrote is not being served after the reload they triggered. The DB write is durable either way and every other pod reloads on its own interval; this speaks only for the handling pod. Callers hold a ReconcileOutcome from the reload; pass BOTH of its fields. Supplying still_desired without live_after mixes a snapshot taken under the reconcile lock with one taken after it was released, which is what makes a concurrent model write look like collateral damage.""" missing, collateral = reload_serving_verdict( before=before, written_models=written_models, written_must_serve=True, still_desired=still_desired, live_after=live_after, ) if not missing and not collateral: return missing_clause: Final = ( f"the model id(s) {list(missing)} are not live in this pod's router after the reload and are not " "being served by this pod." if missing else "the reload it triggered degraded this pod's serving state." ) collateral_clause: Final = ( f" Previously served model id(s) {list(collateral)} are also no longer being served by this pod." if collateral else "" ) raise ProxyException( message=( f"Model {action} was saved to the database, but {missing_clause}{collateral_clause} " "Other pods reload on their own interval. Check server logs for 'Error upserting deployment' or " "'Error creating deployment' for the cause." ), type=ProxyErrorTypes.internal_server_error, code=status.HTTP_500_INTERNAL_SERVER_ERROR, param=None, ) async def clear_cache() -> ReconcileOutcome: """ Clear router caches and reload models. Returns what the reload saw (see ReconcileOutcome) so callers can pass it to raise_if_reload_degraded_serving. Runs under MODEL_RECONCILE_LOCK for its whole extent, not just the reload at the end, so the auto-router reset and the reload that rebuilds those routers are atomic to any other reconcile. The inner call is _add_deployment_locked because add_deployment would re-acquire the same non-reentrant lock and deadlock. """ from litellm.proxy.proxy_server import ( MODEL_RECONCILE_LOCK, llm_router, prisma_client, proxy_config, proxy_logging_obj, verbose_proxy_logger, ) if llm_router is None or prisma_client is None: verbose_proxy_logger.debug("llm_router or prisma_client is None, skipping cache clear") return ReconcileOutcome(still_desired=None, live_after=None) async with MODEL_RECONCILE_LOCK: try: # Only clear DB models, preserve config models verbose_proxy_logger.debug("Clearing only DB models, preserving config models") # Get current models and filter out DB models current_models: Final = llm_router.model_list.copy() config_models: Final = [] db_model_ids: Final = [] db_router_names: Final = set() for model in current_models: model_info = model.get("model_info", {}) if model_info.get("db_model", False): db_model_ids.append(model_info.get("id")) # Auto-router deployments (and only those) are wiped here, in the # same pass, so the reload rebuilds them -- see the comment below. model_name = model.get("model_name") if model_name is not None and str(model.get("litellm_params", {}).get("model", "")).startswith( "auto_router/" ): db_router_names.add(model_name) router_model_id = model_info.get("id") if router_model_id is not None: llm_router.delete_deployment(id=router_model_id) else: # This is a config model, preserved by the reconcile below config_models.append(model) # ORDINARY db deployments are deliberately NOT wiped. This used to # delete_deployment() every db model before the reload put them back, which # left the router serving ZERO db models for the whole width of the reload # -- a real data-plane hole that every inference request landing in it fell # into. It was also redundant for them: the reload's _delete_deployment # evicts exactly the ids the db no longer lists, and upsert_deployment # pops-and-re-adds a deployment whose params changed while no-opping one # that did not, so the reconcile converges on its own. Every mutation is # visible to that comparison -- `blocked` and (for premium) `updated_at` # are written into model_info. # # AUTO-ROUTER db deployments are the exception and ARE wiped -- in the # classification pass above, together with the strategy entries popped # just below. Their strategy registries are keyed # by model_name, which no deployment-id reconcile touches, so they have to # be popped and rebuilt here. But the rebuild only happens on the ADD path: # Router.upsert_deployment returns early when a deployment is unchanged and # never reaches add_deployment -> _add_deployment -> # init_auto_router_deployment, which is what repopulates the registries. # Popping without deleting would therefore strip every db-backed auto, # complexity, adaptive and quality router on this pod and never put it back, # so ANY unrelated model write would leave them unroutable until a restart. # Deleting the deployment forces upsert down the add path, which rebuilds # both the deployment and its strategy entry. # # That pass restricts the wipe to deployments whose model is actually an # auto_router/* so a config router that merely shares a model_name with a # regular db model isn't evicted -- config routers are never re-added by the # reload (it only reloads db models) and would be permanently unroutable. # The auto_router/ prefix also covers quality_router/ and adaptive_router/, # so pop the name from every registry (no-op where absent); a missing # quality/adaptive entry would otherwise make init raise "already exists" # on reload and abort it. for model_name in db_router_names: llm_router.auto_routers.pop(model_name, None) llm_router.complexity_routers.pop(model_name, None) llm_router.adaptive_routers.pop(model_name, None) llm_router.quality_routers.pop(model_name, None) # Reload only DB models. _add_deployment_locked, not add_deployment: this # coroutine already holds MODEL_RECONCILE_LOCK and asyncio.Lock is not # reentrant, so the public wrapper would deadlock against itself. outcome: Final = await proxy_config._add_deployment_locked( prisma_client=prisma_client, proxy_logging_obj=proxy_logging_obj ) verbose_proxy_logger.debug( "Reconciled %s DB models, preserved %s config models", len(db_model_ids), len(config_models) ) return outcome except Exception as e: verbose_proxy_logger.exception("Failed to clear cache and reload models. Due to error - %s", e) return ReconcileOutcome(still_desired=None, live_after=None)