feat: added /v1/models/{model}/next-fallback endpoint

litellm/proxy/auth/model_checks.py

@@ -74,16 +74,13 @@ async def get_mcp_server_ids(
     if prisma_client is None:
         return []
 
-
     if user_api_key_dict.object_permission_id is None:
         return []
 
-
     # Make a direct SQL query to get just the mcp_servers
     try:
-
         result = await prisma_client.db.litellm_objectpermissiontable.find_unique(
-                where={"object_permission_id": user_api_key_dict.object_permission_id},
+            where={"object_permission_id": user_api_key_dict.object_permission_id},
         )
         if result and result.mcp_servers:
             return result.mcp_servers
@@ -287,3 +284,123 @@ def _get_wildcard_models(
         unique_models.remove(model)
 
     return all_wildcard_models
+
+
+def get_next_fallback(
+    model: str,
+    user_api_key_dict: UserAPIKeyAuth,
+    llm_router: Optional[Router] = None,
+    fallback_type: str = "general",
+) -> Optional[str]:
+    """
+    Returns the next immediate fallback for a specific model.
+
+    Args:
+        model: The current model to find fallback for
+        user_api_key_dict: User API key authentication info
+        llm_router: Router instance containing fallback configurations
+        fallback_type: Type of fallback ("general", "context_window", "content_policy")
+
+    Returns:
+        The next fallback model name, or None if no fallback exists
+
+    Example:
+        For fallback config: {"claude-4-sonnet": ["bedrock-claude-sonnet-4", "google-claude-sonnet-4"]}
+        get_next_fallback("claude-4-sonnet") -> "bedrock-claude-sonnet-4"
+        get_next_fallback("bedrock-claude-sonnet-4") -> "google-claude-sonnet-4"
+        get_next_fallback("google-claude-sonnet-4") -> None
+    """
+    if llm_router is None:
+        return None
+
+    # Get the appropriate fallback configuration based on type
+    fallbacks = None
+    if fallback_type == "context_window":
+        fallbacks = llm_router.context_window_fallbacks
+    elif fallback_type == "content_policy":
+        fallbacks = llm_router.content_policy_fallbacks
+    else:  # general or default
+        fallbacks = llm_router.fallbacks
+
+    if not fallbacks:
+        return None
+
+    # Search through all fallback configurations to find where this model appears
+    for fallback_dict in fallbacks:
+        if isinstance(fallback_dict, dict):
+            primary_model = list(fallback_dict.keys())[0]
+            fallback_list = fallback_dict[primary_model]
+
+            # Check if this model is the primary model in this fallback configuration
+            if primary_model == model or _check_stripped_model_group(
+                model, primary_model
+            ):
+                # This is a primary model, return the first fallback
+                if fallback_list and len(fallback_list) > 0:
+                    return fallback_list[0]
+
+            # Check if this model appears anywhere in the fallback list
+            elif model in fallback_list:
+                # This model is in the fallback chain, find its position and return the next one
+                current_index = fallback_list.index(model)
+                if current_index + 1 < len(fallback_list):
+                    return fallback_list[current_index + 1]
+                # If we're at the end of the chain, no more fallbacks
+                return None
+
+            # Also check with stripped model names for provider prefixes
+            else:
+                # Check if model matches primary with stripped names
+                if _check_stripped_model_group(
+                    model, primary_model
+                ) or _check_stripped_model_group(primary_model, model):
+                    if fallback_list and len(fallback_list) > 0:
+                        return fallback_list[0]
+
+                # Check if model matches any fallback with stripped names
+                for idx, fallback_model in enumerate(fallback_list):
+                    if _check_stripped_model_group(
+                        model, fallback_model
+                    ) or _check_stripped_model_group(fallback_model, model):
+                        # Found current model in fallback chain, return next one
+                        if idx + 1 < len(fallback_list):
+                            return fallback_list[idx + 1]
+                        return None
+
+    # Check for generic wildcard fallbacks
+    for fallback_dict in fallbacks:
+        if isinstance(fallback_dict, dict):
+            primary_model = list(fallback_dict.keys())[0]
+            if primary_model == "*":  # Generic fallback
+                fallback_list = fallback_dict["*"]
+                if fallback_list and len(fallback_list) > 0:
+                    return fallback_list[0]
+
+    return None
+
+
+def _check_stripped_model_group(model_group: str, fallback_key: str) -> bool:
+    """
+    Handles wildcard routing scenario - reused from router_utils for consistency
+
+    where fallbacks set like:
+    [{"gpt-3.5-turbo": ["claude-3-haiku"]}]
+
+    but model_group is like:
+    "openai/gpt-3.5-turbo"
+
+    Returns:
+    - True if the stripped model group == fallback_key
+    """
+    for provider in litellm.provider_list:
+        from enum import Enum
+
+        if isinstance(provider, Enum):
+            _provider = provider.value
+        else:
+            _provider = provider
+        if model_group.startswith(f"{_provider}/"):
+            stripped_model_group = model_group.replace(f"{_provider}/", "")
+            if stripped_model_group == fallback_key:
+                return True
+    return False

litellm/proxy/proxy_server.py

@@ -3736,6 +3736,162 @@ async def model_list(
     )
 
 
+@router.get(
+    "/v1/models/{model}/next-fallback",
+    dependencies=[Depends(user_api_key_auth)],
+    tags=["model management"],
+)
+async def get_model_next_fallback(
+    model: str,
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+    fallback_type: str = "general",
+):
+    """
+    Returns the next immediate fallback for a specific model.
+
+    Args:
+        model: The current model to find fallback for
+        fallback_type: Type of fallback ("general", "context_window", "content_policy")
+
+    Returns:
+        JSON response with the next fallback model or 404 if no fallback exists
+
+    Example:
+        GET /v1/models/claude-4-sonnet/next-fallback
+        Returns: {"current_model": "claude-4-sonnet", "next_fallback": "bedrock-claude-sonnet-4", ...}
+    """
+    from litellm.proxy.auth.model_checks import get_next_fallback
+
+    global llm_router
+
+    if llm_router is None:
+        raise HTTPException(
+            status_code=404,
+            detail={
+                "error": {
+                    "message": "No router configured - fallbacks not available",
+                    "type": "not_found", 
+                    "code": "no_router_configured"
+                }
+            }
+        )
+
+    # Validate fallback_type parameter
+    valid_fallback_types = ["general", "context_window", "content_policy"]
+    if fallback_type not in valid_fallback_types:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": {
+                    "message": f"Invalid fallback_type. Must be one of: {valid_fallback_types}",
+                    "type": "invalid_request_error",
+                    "code": "invalid_fallback_type"
+                }
+            }
+        )
+
+    # Check if user has access to the requested model
+    proxy_model_list = llm_router.get_model_names() if llm_router else []
+    model_access_groups = llm_router.get_model_access_groups() if llm_router else {}
+
+    key_models = get_key_models(
+        user_api_key_dict=user_api_key_dict,
+        proxy_model_list=proxy_model_list,
+        model_access_groups=model_access_groups,
+    )
+    team_models = get_team_models(
+        team_models=user_api_key_dict.team_models,
+        proxy_model_list=proxy_model_list,
+        model_access_groups=model_access_groups,
+    )
+    all_accessible_models = get_complete_model_list(
+        key_models=key_models,
+        team_models=team_models,
+        proxy_model_list=proxy_model_list,
+        user_model=None,
+        infer_model_from_keys=general_settings.get("infer_model_from_keys", False),
+        llm_router=llm_router,
+        model_access_groups=model_access_groups,
+    )
+
+    # Check if the user has access to the requested model
+    if model not in all_accessible_models:
+        raise HTTPException(
+            status_code=404,
+            detail={
+                "error": {
+                    "message": f"Model '{model}' not found or not accessible",
+                    "type": "not_found",
+                    "code": "model_not_found"
+                }
+            }
+        )
+
+    # Get the next fallback
+    next_fallback = get_next_fallback(
+        model=model,
+        user_api_key_dict=user_api_key_dict,
+        llm_router=llm_router,
+        fallback_type=fallback_type,
+    )
+
+    if next_fallback is None:
+        raise HTTPException(
+            status_code=404,
+            detail={
+                "error": {
+                    "message": f"No fallback available for model: {model}",
+                    "type": "not_found",
+                    "code": "no_fallback_available"
+                }
+            }
+        )
+
+    # Validate user has access to the fallback model
+    if next_fallback not in all_accessible_models:
+        # If user doesn't have access to this fallback, try to find the next one they do have access to
+        # This handles cases where some fallbacks might be restricted
+        from litellm.proxy.auth.model_checks import get_next_fallback
+
+        # Try to get the fallback after this one
+        temp_fallback = get_next_fallback(
+            model=next_fallback,
+            user_api_key_dict=user_api_key_dict,
+            llm_router=llm_router,
+            fallback_type=fallback_type,
+        )
+
+        # Keep looking until we find an accessible one or run out
+        while temp_fallback and temp_fallback not in all_accessible_models:
+            temp_fallback = get_next_fallback(
+                model=temp_fallback,
+                user_api_key_dict=user_api_key_dict,
+                llm_router=llm_router,
+                fallback_type=fallback_type,
+            )
+
+        if temp_fallback and temp_fallback in all_accessible_models:
+            next_fallback = temp_fallback
+        else:
+            raise HTTPException(
+                status_code=404,
+                detail={
+                    "error": {
+                        "message": f"No accessible fallback available for model: {model}",
+                        "type": "not_found",
+                        "code": "no_accessible_fallback"
+                    }
+                }
+            )
+
+    return {
+        "current_model": model,
+        "next_fallback": next_fallback,
+        "fallback_type": fallback_type,
+        "object": "next_fallback"
+    }
+
+
 @router.post(
     "/v1/chat/completions",
     dependencies=[Depends(user_api_key_auth)],