feat(proxy): add admin toggle to block requests for models without pricing

litellm/__init__.py

@@ -453,6 +453,7 @@ def _dev_env_hot_reload_enabled() -> bool:
 # backwards compatibility — arbitrary client-supplied identifiers still
 # pass through unchanged.
 validate_end_user_id_in_db: bool = False
+block_requests_for_models_without_pricing: bool = False
 disable_end_user_cost_tracking: Optional[bool] = None
 disable_end_user_cost_tracking_prometheus_only: Optional[bool] = None
 enable_end_user_cost_tracking_prometheus_only: Optional[bool] = None

litellm/constants.py

@@ -1580,6 +1580,7 @@
     "public_model_groups_links",
     "cost_discount_config",
     "cost_margin_config",
+    "block_requests_for_models_without_pricing",
     "budget_exceeded_throttle_percentage",
     # Every field editable from the Admin UI (proxy_server._GENERAL_SETTINGS_UI_LITELLM_FIELDS)
     # must be listed here so a DB write from one worker overrides the live litellm attribute on

litellm/proxy/_types.py

@@ -3721,6 +3721,8 @@ class ProxyErrorTypes(str, enum.Enum):
     Project does not have access to the model
     """
 
+    model_cost_map_missing = "model_cost_map_missing"
+
     expired_key = "expired_key"
     """
     Key has expired

litellm/proxy/auth/auth_checks.py

@@ -464,6 +464,103 @@ def _is_cost_explicitly_configured(model: str, llm_router: "Router") -> bool:
     return False
 
 
+_EMPTY_COST_ENTRY: Final[Mapping[str, object]] = MappingProxyType({})
+
+
+def _is_positive_cost(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0
+
+
+def _entry_has_priced_metric(entry: Mapping[str, object]) -> bool:
+    if entry.get("tiered_pricing") is not None:
+        return True
+    for key, value in entry.items():
+        if "cost_per" not in key:
+            continue
+        if _is_positive_cost(value):
+            return True
+        if isinstance(value, dict) and any(_is_positive_cost(nested) for nested in value.values()):
+            return True
+    return False
+
+
+def _entry_declares_price(entry: Mapping[str, object]) -> bool:
+    return any("cost_per" in key or key == "tiered_pricing" for key in entry)
+
+
+def _model_group_has_pricing(model: str, llm_router: "Router") -> bool:
+    """
+    A model group counts as priced when a deployment overrides any *cost_per* field or
+    tiered_pricing in its litellm_params, even at zero, or when its resolved model info carries
+    tiered_pricing or a positive price on any billed metric (tokens, characters, seconds, pages,
+    images, queries, ...), so models billed by a non-token metric are not treated as unpriced.
+    """
+    for deployment in llm_router.get_model_list(model_name=model) or ():
+        litellm_params = deployment.get("litellm_params") or _EMPTY_COST_ENTRY
+        if _entry_declares_price(litellm_params):
+            return True
+
+        model_id = (deployment.get("model_info") or _EMPTY_COST_ENTRY).get("id")
+        if model_id is None:
+            continue
+
+        model_info = llm_router.get_deployment_model_info(
+            model_id=model_id, model_name=litellm_params.get("model") or ""
+        )
+        if model_info is not None and _entry_has_priced_metric(model_info):
+            return True
+
+    return False
+
+
+def _group_declares_explicit_cost(model: str, llm_router: "Router") -> bool:
+    """
+    Alias-aware counterpart to ``_is_cost_explicitly_configured``, which resolves the model group
+    the same way ``_model_group_has_pricing`` does. A deployment that prices itself through its
+    ``model_info`` block lands in the cost map under its deployment id rather than in its
+    litellm_params, and reaching that entry through the router's own resolution keeps an alias
+    pointing at such a group from being read as unpriced.
+    """
+    for deployment in llm_router.get_model_list(model_name=model) or ():
+        model_id = (deployment.get("model_info") or _EMPTY_COST_ENTRY).get("id")
+        if model_id is None:
+            continue
+        raw_entry = litellm.model_cost.get(model_id, _EMPTY_COST_ENTRY)
+        if "input_cost_per_token" in raw_entry or "output_cost_per_token" in raw_entry:
+            return True
+    return False
+
+
+def model_has_no_cost_mapping(model: str | None, llm_router: Router | None) -> bool:
+    if not model or llm_router is None:
+        return False
+
+    if llm_router.get_model_group_info(model_group=model) is None:
+        return False
+
+    if _model_group_has_pricing(model=model, llm_router=llm_router):
+        return False
+
+    return not _group_declares_explicit_cost(model=model, llm_router=llm_router)
+
+
+def _unpriced_models_in_request(model: str | list[str] | None, llm_router: Router | None) -> tuple[str, ...]:
+    candidates: Final = (model,) if isinstance(model, str) else tuple(model or ())
+    return tuple(
+        candidate for candidate in candidates if model_has_no_cost_mapping(model=candidate, llm_router=llm_router)
+    )
+
+
+def _unpriced_models_block_message(models: tuple[str, ...]) -> str:
+    names: Final = ", ".join(f"'{model}'" for model in models)
+    subject: Final = f"Model {names} has" if len(models) == 1 else f"Models {names} have"
+    return (
+        f"{subject} no pricing in the cost map, so litellm cannot price the request. "
+        "Requests for unpriced models are blocked because 'block_requests_for_models_without_pricing' "
+        "is enabled. Add pricing (input_cost_per_token/output_cost_per_token) to allow the request."
+    )
+
+
 async def _run_project_checks(
     project_object: LiteLLM_ProjectTableCachedObj | None,
     _model: str | list[str] | None,
@@ -734,6 +831,19 @@ async def common_checks(
         and (route in MODEL_DISCOVERY_ROUTES or not RouteChecks.is_llm_api_route(route=route))
     )
 
+    unpriced_models: Final = (
+        _unpriced_models_in_request(model=_model, llm_router=llm_router)
+        if litellm.block_requests_for_models_without_pricing and RouteChecks.is_llm_api_route(route=route)
+        else ()
+    )
+    if unpriced_models:
+        raise ProxyException(
+            message=_unpriced_models_block_message(unpriced_models),
+            type=ProxyErrorTypes.model_cost_map_missing,
+            param="model",
+            code=status.HTTP_403_FORBIDDEN,
+        )
+
     # 1. If team is blocked
     if team_object is not None and team_object.blocked is True:
         raise Exception(f"Team={team_object.team_id} is blocked. Update via `/team/unblock` if you're an admin.")

litellm/proxy/management_endpoints/cost_tracking_settings.py

@@ -15,6 +15,7 @@
 from typing import Final
 
 from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
 
 import litellm
 from litellm._logging import verbose_proxy_logger
@@ -439,6 +440,76 @@ async def update_cost_margin_config(
         )
 
 
+class BlockUnpricedModelsRequest(BaseModel):
+    enabled: bool
+
+
+class BlockUnpricedModelsResponse(BaseModel):
+    enabled: bool
+
+
+@router.get(
+    "/config/block_requests_for_models_without_pricing",
+    tags=("Cost Tracking",),
+    dependencies=(Depends(user_api_key_auth),),
+    response_model=BlockUnpricedModelsResponse,
+)
+async def get_block_requests_for_models_without_pricing() -> BlockUnpricedModelsResponse:
+    return BlockUnpricedModelsResponse(enabled=bool(litellm.block_requests_for_models_without_pricing))
+
+
+@router.patch(
+    "/config/block_requests_for_models_without_pricing",
+    tags=("Cost Tracking",),
+    dependencies=(Depends(user_api_key_auth),),
+    response_model=BlockUnpricedModelsResponse,
+)
+async def update_block_requests_for_models_without_pricing(
+    request: BlockUnpricedModelsRequest,
+) -> BlockUnpricedModelsResponse:
+    from litellm.proxy.proxy_server import (
+        prisma_client,
+        proxy_config,
+        store_model_in_db,
+    )
+
+    if prisma_client is None:
+        raise HTTPException(
+            status_code=500,
+            detail={  # mutable-ok: HTTPException detail must be a plain mapping
+                "error": CommonProxyErrors.db_not_connected_error.value
+            },
+        )
+
+    if store_model_in_db is not True:
+        raise HTTPException(
+            status_code=500,
+            detail={  # mutable-ok: HTTPException detail must be a plain mapping
+                "error": "Set `'STORE_MODEL_IN_DB='True'` in your env to enable this feature."
+            },
+        )
+
+    try:
+        config = await proxy_config.get_config()
+        if "litellm_settings" not in config:
+            config["litellm_settings"] = {}  # mutable-ok: config is a plain-dict payload for save_config
+        config["litellm_settings"]["block_requests_for_models_without_pricing"] = request.enabled
+        await proxy_config.save_config(new_config=config)
+
+        litellm.block_requests_for_models_without_pricing = request.enabled
+        verbose_proxy_logger.info("Updated block_requests_for_models_without_pricing: %s", request.enabled)
+
+        return BlockUnpricedModelsResponse(enabled=request.enabled)
+    except Exception as e:  # noqa: BLE001  # any config persistence failure must surface as a 500 response, not a crash
+        verbose_proxy_logger.error("Error updating block_requests_for_models_without_pricing: %s", e)
+        raise HTTPException(
+            status_code=500,
+            detail={  # mutable-ok: HTTPException detail must be a plain mapping
+                "error": f"Failed to update setting: {e!s}"
+            },
+        )
+
+
 @router.post(
     "/cost/estimate",
     tags=["Cost Tracking"],

litellm/proxy/proxy_server.py

@@ -6824,6 +6824,20 @@ async def _init_non_llm_objects_in_db(self, prisma_client: PrismaClient):
         if self._should_load_db_object(object_type="config_overrides"):
             await self._init_hashicorp_vault_config_override(prisma_client=prisma_client)
 
+        await self._apply_safe_litellm_settings_overrides_from_db(prisma_client=prisma_client)
+
+    async def _apply_safe_litellm_settings_overrides_from_db(self, prisma_client: PrismaClient) -> None:
+        config_record: Final = await get_config_param(prisma_client, "litellm_settings")
+        if config_record is None or config_record.param_value is None:
+            return
+        raw_settings: Final = config_record.param_value
+        litellm_settings: Final = json.loads(raw_settings) if isinstance(raw_settings, str) else raw_settings
+        if not isinstance(litellm_settings, dict):
+            return
+        for key, value in litellm_settings.items():
+            if key in LITELLM_SETTINGS_SAFE_DB_OVERRIDES:
+                setattr(litellm, key, value)
+
     async def _init_semantic_filter_settings_in_db(self, prisma_client: PrismaClient):
         """
         Initialize MCP semantic filter settings from database.