fix(proxy_setting_endpoints.py): add new GET /in_product_nudges route

allows for context-based nudges
This commit is contained in:
Krrish Dholakia 2026-01-18 07:01:52 +05:30
parent 60dd04ac95
commit e2593bdda2
36 changed files with 122 additions and 50 deletions

View file

@ -0,0 +1,2 @@
-- This is an empty migration.

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View file

@ -1,5 +1,6 @@
#### Analytics Endpoints #####
import os
from fastapi import APIRouter
from litellm.types.proxy.discovery_endpoints.ui_discovery_endpoints import (
@ -14,10 +15,12 @@ router = APIRouter()
"/litellm/.well-known/litellm-ui-config", response_model=UiDiscoveryEndpoints
) # if mounted at root path
async def get_ui_config():
from litellm.proxy.utils import get_proxy_base_url, get_server_root_path
from litellm.proxy.auth.auth_utils import _has_user_setup_sso
from litellm.proxy.utils import get_proxy_base_url, get_server_root_path
auto_redirect_ui_login_to_sso = os.getenv("AUTO_REDIRECT_UI_LOGIN_TO_SSO", "true").lower() == "true"
auto_redirect_ui_login_to_sso = (
os.getenv("AUTO_REDIRECT_UI_LOGIN_TO_SSO", "true").lower() == "true"
)
admin_ui_disabled = os.getenv("DISABLE_ADMIN_UI", "false").lower() == "true"
return UiDiscoveryEndpoints(

View file

@ -1,8 +1,8 @@
#### CRUD ENDPOINTS for UI Settings #####
import json
from typing import Any, Dict, List, Union, Optional
from typing import Any, Dict, List, Optional, Union
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
import litellm
from litellm._logging import verbose_proxy_logger
@ -10,6 +10,7 @@ from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.types.proxy.management_endpoints.ui_sso import (
DefaultTeamSSOParams,
InProductNudgeResponse,
SSOConfig,
)
@ -22,11 +23,11 @@ class IPAddress(BaseModel):
class UIThemeConfig(BaseModel):
"""Configuration for UI theme customization"""
# Logo configuration
logo_url: Optional[str] = Field(
default=None,
description="URL or path to custom logo image. Can be a local file path or HTTP/HTTPS URL"
description="URL or path to custom logo image. Can be a local file path or HTTP/HTTPS URL",
)
@ -85,7 +86,10 @@ class UISettingsResponse(SettingsResponse):
# Allowlist of UI settings that can be stored
ALLOWED_UI_SETTINGS_FIELDS = {"disable_model_add_for_internal_users", "disable_team_admin_delete_team_user"}
ALLOWED_UI_SETTINGS_FIELDS = {
"disable_model_add_for_internal_users",
"disable_team_admin_delete_team_user",
}
@router.get(
@ -434,7 +438,7 @@ async def get_sso_settings():
# Initialize with defaults
sso_settings_dict = {}
if sso_db_record and sso_db_record.sso_settings:
# Load settings from database
sso_settings_dict = dict(sso_db_record.sso_settings)
@ -444,26 +448,43 @@ async def get_sso_settings():
role_mappings = None
if role_mappings_data:
from litellm.types.proxy.management_endpoints.ui_sso import RoleMappings
if isinstance(role_mappings_data, dict):
role_mappings = RoleMappings(**role_mappings_data)
elif isinstance(role_mappings_data, RoleMappings):
role_mappings = role_mappings_data
decrypted_sso_settings_dict = proxy_config._decrypt_and_set_db_env_variables(environment_variables=sso_settings_dict)
decrypted_sso_settings_dict = proxy_config._decrypt_and_set_db_env_variables(
environment_variables=sso_settings_dict
)
# Build SSO config with database values or environment fallback
sso_config = SSOConfig(
google_client_id=decrypted_sso_settings_dict.get("google_client_id", None),
google_client_secret=decrypted_sso_settings_dict.get("google_client_secret", None),
microsoft_client_id=decrypted_sso_settings_dict.get("microsoft_client_id", None),
microsoft_client_secret=decrypted_sso_settings_dict.get("microsoft_client_secret", None),
google_client_secret=decrypted_sso_settings_dict.get(
"google_client_secret", None
),
microsoft_client_id=decrypted_sso_settings_dict.get(
"microsoft_client_id", None
),
microsoft_client_secret=decrypted_sso_settings_dict.get(
"microsoft_client_secret", None
),
microsoft_tenant=decrypted_sso_settings_dict.get("microsoft_tenant", None),
generic_client_id=decrypted_sso_settings_dict.get("generic_client_id", None),
generic_client_secret=decrypted_sso_settings_dict.get("generic_client_secret", None),
generic_authorization_endpoint=decrypted_sso_settings_dict.get("generic_authorization_endpoint", None),
generic_token_endpoint=decrypted_sso_settings_dict.get("generic_token_endpoint", None),
generic_userinfo_endpoint=decrypted_sso_settings_dict.get("generic_userinfo_endpoint", None),
generic_client_secret=decrypted_sso_settings_dict.get(
"generic_client_secret", None
),
generic_authorization_endpoint=decrypted_sso_settings_dict.get(
"generic_authorization_endpoint", None
),
generic_token_endpoint=decrypted_sso_settings_dict.get(
"generic_token_endpoint", None
),
generic_userinfo_endpoint=decrypted_sso_settings_dict.get(
"generic_userinfo_endpoint", None
),
proxy_base_url=decrypted_sso_settings_dict.get("proxy_base_url", None),
user_email=decrypted_sso_settings_dict.get("user_email"),
ui_access_mode=decrypted_sso_settings_dict.get("ui_access_mode"),
@ -506,10 +527,14 @@ async def update_sso_settings(sso_config: SSOConfig):
"""
Update SSO configuration by saving to the dedicated SSO table.
"""
import os
import json
import os
from litellm.proxy.proxy_server import prisma_client, store_model_in_db, proxy_config
from litellm.proxy.proxy_server import (
prisma_client,
proxy_config,
store_model_in_db,
)
if prisma_client is None:
raise HTTPException(
@ -562,7 +587,9 @@ async def update_sso_settings(sso_config: SSOConfig):
# Clear environment variable if value is null/empty
os.environ.pop(env_var_name, None)
encrypted_sso_data = proxy_config._encrypt_env_variables(environment_variables=sso_data)
encrypted_sso_data = proxy_config._encrypt_env_variables(
environment_variables=sso_data
)
# Save to dedicated SSO table
await prisma_client.db.litellm_ssoconfig.upsert(
@ -655,9 +682,10 @@ async def update_ui_theme_settings(theme_config: UIThemeConfig):
Update UI theme configuration.
Updates logo settings for the admin UI.
"""
from litellm.proxy.proxy_server import proxy_config, store_model_in_db
import os
from litellm.proxy.proxy_server import proxy_config, store_model_in_db
if store_model_in_db is not True:
raise HTTPException(
status_code=500,
@ -668,28 +696,30 @@ async def update_ui_theme_settings(theme_config: UIThemeConfig):
# Load existing config
config = await proxy_config.get_config()
# Update config with UI theme settings
if "general_settings" not in config:
config["general_settings"] = {}
if "environment_variables" not in config:
config["environment_variables"] = {}
# Convert theme config to dict
theme_data = theme_config.model_dump(exclude_none=True)
# Store UI theme config in litellm_settings (where it's retrieved from)
if "litellm_settings" not in config:
config["litellm_settings"] = {}
config["litellm_settings"]["ui_theme_config"] = theme_data
# Update UI_LOGO_PATH environment variable if logo_url is provided
# If logo_url is empty string, None, or null, remove the environment variable to use default
logo_url = theme_data.get("logo_url")
verbose_proxy_logger.debug(f"Updating logo_url: {logo_url}")
if logo_url and isinstance(logo_url, str) and logo_url.strip(): # Check if logo_url exists and is not empty/whitespace
if (
logo_url and isinstance(logo_url, str) and logo_url.strip()
): # Check if logo_url exists and is not empty/whitespace
config["environment_variables"]["UI_LOGO_PATH"] = logo_url
os.environ["UI_LOGO_PATH"] = logo_url
verbose_proxy_logger.debug(f"Set UI_LOGO_PATH to: {logo_url}")
@ -704,12 +734,15 @@ async def update_ui_theme_settings(theme_config: UIThemeConfig):
# Handle environment variable encryption if needed
stored_config = config.copy()
if "environment_variables" in stored_config and len(stored_config["environment_variables"]) > 0:
if (
"environment_variables" in stored_config
and len(stored_config["environment_variables"]) > 0
):
# Only encrypt if there are environment variables to encrypt
stored_config["environment_variables"] = proxy_config._encrypt_env_variables(
environment_variables=stored_config["environment_variables"]
)
# Save the updated config
await proxy_config.save_config(new_config=stored_config)
@ -720,6 +753,34 @@ async def update_ui_theme_settings(theme_config: UIThemeConfig):
}
@router.get(
"/in_product_nudges",
tags=["UI Settings"],
dependencies=[Depends(user_api_key_auth)],
response_model=InProductNudgeResponse,
)
async def get_in_product_nudges():
"""
Get in-product nudges configuration.
"""
from litellm.proxy.proxy_server import prisma_client
if prisma_client is None:
raise HTTPException(
status_code=500,
detail={"error": "Database not connected. Please connect a database."},
)
db_record = await prisma_client.db.litellm_dailytagspend.find_first(
where={"tag": "User-Agent: claude-cli"}
)
if db_record:
return InProductNudgeResponse(is_claude_code_enabled=True)
return InProductNudgeResponse(is_claude_code_enabled=False)
@router.get(
"/get/ui_settings",
tags=["UI Settings"],
@ -752,7 +813,9 @@ async def get_ui_settings():
ui_settings = dict(ui_settings_json)
# Sanitize any unexpected keys from persisted config before returning
ui_settings = {k: v for k, v in ui_settings.items() if k in ALLOWED_UI_SETTINGS_FIELDS}
ui_settings = {
k: v for k, v in ui_settings.items() if k in ALLOWED_UI_SETTINGS_FIELDS
}
# Build config-like object for schema helper
config: Dict[str, Any] = {"litellm_settings": {"ui_settings": ui_settings}}
@ -823,6 +886,7 @@ async def update_ui_settings(
"settings": ui_settings,
}
@router.post(
"/upload/logo",
tags=["UI Theme Settings"],
@ -839,35 +903,35 @@ async def upload_logo(file: UploadFile = File(...)):
# Validate file type
allowed_extensions = {".png", ".jpg", ".jpeg", ".svg"}
file_extension = Path(file.filename or "").suffix.lower()
if file_extension not in allowed_extensions:
raise HTTPException(
status_code=400,
detail=f"Invalid file type. Allowed types: {', '.join(allowed_extensions)}"
detail=f"Invalid file type. Allowed types: {', '.join(allowed_extensions)}",
)
# Validate file size (max 5MB)
file_content = await file.read()
if len(file_content) > 5 * 1024 * 1024: # 5MB
raise HTTPException(
status_code=400,
detail="File size too large. Maximum size is 5MB."
status_code=400, detail="File size too large. Maximum size is 5MB."
)
# Create uploads directory if it doesn't exist
current_dir = os.path.dirname(os.path.abspath(__file__))
upload_dir = os.path.join(current_dir, "..", "uploads")
os.makedirs(upload_dir, exist_ok=True)
# Generate unique filename
from litellm._uuid import uuid
unique_filename = f"logo_{uuid.uuid4().hex}{file_extension}"
file_path = os.path.join(upload_dir, unique_filename)
# Save the file
with open(file_path, "wb") as buffer:
buffer.write(file_content)
return {
"message": "Logo uploaded successfully",
"status": "success",

View file

@ -1,11 +1,10 @@
from typing import Dict, List, Literal, Optional, Union
from pydantic import Field
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
from litellm.types.utils import LiteLLMPydanticObjectBase
from litellm.proxy._types import LitellmUserRoles
from litellm.types.utils import LiteLLMPydanticObjectBase
class LiteLLM_UpperboundKeyGenerateParams(LiteLLMPydanticObjectBase):
@ -28,6 +27,7 @@ class LiteLLM_UpperboundKeyGenerateParams(LiteLLMPydanticObjectBase):
tpm_limit: Optional[int] = None
rpm_limit: Optional[int] = None
class MicrosoftGraphAPIUserGroupDirectoryObject(TypedDict, total=False):
"""Model for Microsoft Graph API directory object"""
@ -65,7 +65,7 @@ class AccessControl_UI_AccessMode(LiteLLMPydanticObjectBase):
class RoleMappings(LiteLLMPydanticObjectBase):
"""
Configuration for mapping SSO groups to LiteLLM roles.
The system will look at the group_claim field in the SSO token to determine
which role to assign the user based on the roles mapping.
"""
@ -78,11 +78,11 @@ class RoleMappings(LiteLLMPydanticObjectBase):
)
default_role: Optional[LitellmUserRoles] = Field(
default=None,
description="Default role to assign if user's groups don't match any role mappings. Must be a valid LitellmUserRoles value (e.g., 'proxy_admin', 'internal_user', 'proxy_admin_viewer')"
description="Default role to assign if user's groups don't match any role mappings. Must be a valid LitellmUserRoles value (e.g., 'proxy_admin', 'internal_user', 'proxy_admin_viewer')",
)
roles: Dict[LitellmUserRoles, List[str]] = Field(
default_factory=dict,
description="Mapping of LiteLLM role names to arrays of SSO group names. Example: {'proxy_admin': ['group-1', 'group-2'], 'proxy_admin_viewer': ['group-3']}"
description="Mapping of LiteLLM role names to arrays of SSO group names. Example: {'proxy_admin': ['group-1', 'group-2'], 'proxy_admin_viewer': ['group-3']}",
)
@ -185,3 +185,10 @@ class DefaultTeamSSOParams(LiteLLMPydanticObjectBase):
default=None,
description="Default rpm limit for new automatically created teams",
)
class InProductNudgeResponse(BaseModel):
is_claude_code_enabled: bool = Field(
default=False,
description="Whether the Claude Code nudge should be shown.",
)