scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-03 02:22:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

refactor(proxy): read and write config sections through ProxyRuntimeConfig attributes

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
mateo 2026-09-21 23:11:57 +00:00
parent bafd09e658
commit 415032118f
13 changed files with 130 additions and 143 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/proxy/_experimental/mcp_server/tool_registry.py
View file

@ -1,5 +1,5 @@
import json
from collections.abc import Callable
from collections.abc import Callable, Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
from litellm._logging import verbose_logger
@ -86,7 +86,7 @@ class MCPToolRegistry:
def load_tools_from_config(
self,
mcp_tools_config: dict[str, Any] | None = None,
mcp_tools_config: Iterable[Mapping[str, Any]] | None = None,
config_file_path: str | None = None,
) -> None:
"""
@ -105,7 +105,7 @@ class MCPToolRegistry:
for tool_config in mcp_tools_config:
if not isinstance(tool_config, dict):
raise ValueError("mcp_tools_config must be a list of dictionaries")
raise ValueError("mcp_tools_config must be a list of dictionaries") # pyright: ignore[reportUnnecessaryIsInstance] # entries arrive as raw YAML dicts at runtime
name = tool_config.get("name")
description = tool_config.get("description")

25
litellm/proxy/agent_endpoints/endpoints.py
View file

@ -950,12 +950,12 @@ async def make_agent_public(
)
updated_public_agent_groups: Final = [*current_public_agent_groups, agent.agent_id]
if "litellm_settings" not in config or config["litellm_settings"] is None:
config["litellm_settings"] = {}
config["litellm_settings"]["public_agent_groups"] = updated_public_agent_groups
await proxy_config.save_config(new_config=config)
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "public_agent_groups": updated_public_agent_groups}, # mutable-ok: replacement section for save_config
)
)
litellm.public_agent_groups = updated_public_agent_groups
@ -1054,14 +1054,13 @@ async def make_agents_public(
litellm.public_agent_groups = request.agent_ids
# Update config with new settings
if "litellm_settings" not in config or config["litellm_settings"] is None:
config["litellm_settings"] = {}
config["litellm_settings"]["public_agent_groups"] = litellm.public_agent_groups
# Save the updated config
await proxy_config.save_config(new_config=config)
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "public_agent_groups": litellm.public_agent_groups}, # mutable-ok: replacement section for save_config
)
)
verbose_proxy_logger.debug(
"Updated public agent groups to: %s by user: %s", litellm.public_agent_groups, user_api_key_dict.user_id

8
litellm/proxy/guardrails/guardrail_endpoints.py
View file

@ -167,11 +167,11 @@ async def list_guardrails():
"""
from litellm.proxy.proxy_server import proxy_config
config: Final = proxy_config.config
_guardrails_config: Final = [ # mutable-ok: dict copies for the response layer
dict(guardrail) for guardrail in proxy_config.config.guardrails
]
_guardrails_config: Final = cast(list[dict] | None, config.get("guardrails"))
if _guardrails_config is None:
if not _guardrails_config:
return _get_guardrails_list_response([])
return _get_guardrails_list_response(_guardrails_config)

2
litellm/proxy/management_endpoints/coordination_redis_endpoints.py
View file

@ -196,7 +196,7 @@ async def _current_coordination_redis_settings() -> dict[str, object] | None:
if persisted is not None:
return persisted
config_state: Final = _SETTINGS_ADAPTER.validate_python(proxy_config.get_config_state())
config_state: Final = _SETTINGS_ADAPTER.validate_python(proxy_config.get_config_state().to_mapping())
general_settings: Final = config_state.get(_GENERAL_SETTINGS_PARAM_NAME)
if not isinstance(general_settings, Mapping):
return None

48
litellm/proxy/management_endpoints/cost_tracking_settings.py
View file

@ -215,7 +215,7 @@ async def get_cost_discount_config(
config: Final = await proxy_config.get_config()
# Get cost_discount_config from litellm_settings
litellm_settings: Final = config.get("litellm_settings", {})
litellm_settings: Final = config.litellm_settings
cost_discount_config: Final = litellm_settings.get("cost_discount_config", {})
return {"values": cost_discount_config}
@ -294,15 +294,13 @@ async def update_cost_discount_config(
# Load existing config
config: Final = await proxy_config.get_config()
# Ensure litellm_settings exists
if "litellm_settings" not in config:
config["litellm_settings"] = {}
# Update cost_discount_config
config["litellm_settings"]["cost_discount_config"] = cost_discount_config
# Save the updated config to DB
await proxy_config.save_config(new_config=config)
# Update cost_discount_config and save
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "cost_discount_config": cost_discount_config}, # mutable-ok: replacement section for save_config
)
)
# Update in-memory litellm.cost_discount_config
litellm.cost_discount_config = cost_discount_config
@ -348,7 +346,7 @@ async def get_cost_margin_config(
config: Final = await proxy_config.get_config()
# Get cost_margin_config from litellm_settings
litellm_settings: Final = config.get("litellm_settings", {})
litellm_settings: Final = config.litellm_settings
cost_margin_config: Final = litellm_settings.get("cost_margin_config", {})
return {"values": cost_margin_config}
@ -468,15 +466,13 @@ async def update_cost_margin_config(
# Load existing config
config: Final = await proxy_config.get_config()
# Ensure litellm_settings exists
if "litellm_settings" not in config:
config["litellm_settings"] = {}
# Update cost_margin_config
config["litellm_settings"]["cost_margin_config"] = cost_margin_config
# Save the updated config to DB
await proxy_config.save_config(new_config=config)
# Update cost_margin_config and save
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "cost_margin_config": cost_margin_config}, # mutable-ok: replacement section for save_config
)
)
# Update in-memory litellm.cost_margin_config
litellm.cost_margin_config = cost_margin_config
@ -546,11 +542,13 @@ async def update_block_requests_for_models_without_pricing(
)
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)
config: Final = await proxy_config.get_config()
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "block_requests_for_models_without_pricing": request.enabled}, # mutable-ok: replacement section for save_config
)
)
litellm.block_requests_for_models_without_pricing = request.enabled
verbose_proxy_logger.info("Updated block_requests_for_models_without_pricing: %s", request.enabled)

14
litellm/proxy/management_endpoints/fallback_management_endpoints.py
View file

@ -11,7 +11,7 @@ DELETE /fallback/{model} - Delete fallbacks for a specific model
# pyright: reportMissingImports=false
import json
from typing import TYPE_CHECKING, Final, Literal
from typing import TYPE_CHECKING, Final, Literal, cast
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import UserAPIKeyAuth
@ -124,7 +124,7 @@ async def create_fallback(
# Load existing config
config: Final = await proxy_config.get_config()
router_settings: Final = config.get("router_settings", {})
router_settings: Final = dict(config.router_settings) # mutable-ok: fallback list is rewritten below before the DB upsert
# Get the appropriate fallback list based on type
fallback_key = "fallbacks"
@ -134,7 +134,9 @@ async def create_fallback(
fallback_key = "content_policy_fallbacks"
# Get existing fallbacks
existing_fallbacks: Final[list[dict[str, list[str]]]] = router_settings.get(fallback_key, [])
existing_fallbacks: Final[list[dict[str, list[str]]]] = list( # mutable-ok: copy so edits stay off the frozen config
cast("list[dict[str, list[str]]]", router_settings.get(fallback_key)) # cast-ok: entries are model -> fallback models maps
)
# Update or add the fallback configuration
fallback_updated = False
@ -293,7 +295,7 @@ async def delete_fallback(
# Load existing config
config: Final = await proxy_config.get_config()
router_settings: Final = config.get("router_settings", {})
router_settings: Final = dict(config.router_settings) # mutable-ok: fallback list is rewritten below before the DB upsert
# Get the appropriate fallback list based on type
fallback_key = "fallbacks"
@ -303,7 +305,9 @@ async def delete_fallback(
fallback_key = "content_policy_fallbacks"
# Get existing fallbacks
existing_fallbacks: Final[list[dict[str, list[str]]]] = router_settings.get(fallback_key, [])
existing_fallbacks: Final[list[dict[str, list[str]]]] = list( # mutable-ok: copy so edits stay off the frozen config
cast("list[dict[str, list[str]]]", router_settings.get(fallback_key)) # cast-ok: entries are model -> fallback models maps
)
# Find and remove the fallback configuration
fallback_found = False

13
litellm/proxy/management_endpoints/mcp_management_endpoints.py
View file

@ -3012,14 +3012,13 @@ if MCP_AVAILABLE:
litellm.public_mcp_servers = request.mcp_server_ids
# Update config with new settings
if "litellm_settings" not in config or config["litellm_settings"] is None:
config["litellm_settings"] = {}
config["litellm_settings"]["public_mcp_servers"] = litellm.public_mcp_servers
# Save the updated config
await proxy_config.save_config(new_config=config)
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "public_mcp_servers": litellm.public_mcp_servers}, # mutable-ok: replacement section for save_config
)
)
verbose_proxy_logger.debug(
"Updated public mcp servers to: %s by user: %s", litellm.public_mcp_servers, user_api_key_dict.user_id

26
litellm/proxy/management_endpoints/model_management_endpoints.py
View file

@ -2777,14 +2777,13 @@ async def update_public_model_groups(
# 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)
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "public_model_groups": request.model_groups}, # mutable-ok: replacement section for save_config
)
)
# Set in-memory value AFTER get_config() and save_config() to avoid
# get_config() overwriting with stale DB value
@ -2844,14 +2843,13 @@ async def update_useful_links(
# 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)
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, "public_model_groups_links": request.useful_links}, # mutable-ok: replacement section for save_config
)
)
# Set in-memory value AFTER get_config() and save_config() to avoid
# get_config() overwriting with stale DB value

2
litellm/proxy/management_endpoints/router_settings_endpoints.py
View file

@ -92,7 +92,7 @@ async def get_router_settings(
# Try to get router settings from config
config: Final = await proxy_config.get_config()
router_settings_from_config: Final = config.get("router_settings", {})
router_settings_from_config: Final = config.router_settings
current_values: Final[dict[str, Any]] = {}
if llm_router is not None:

4
litellm/proxy/management_endpoints/scim/scim_v2.py
View file

@ -373,7 +373,7 @@ async def _get_scim_upsert_user_setting() -> bool:
from litellm.proxy.proxy_server import proxy_config
config: Final = await proxy_config.get_config()
litellm_settings: Final = config.get("litellm_settings", {}) or {}
litellm_settings: Final = config.litellm_settings
scim_upsert_user: Final = litellm_settings.get("scim_upsert_user", True)
# Default to True if not set (backward compatibility)
@ -412,7 +412,7 @@ async def _get_scim_admin_group() -> str | None:
from litellm.proxy.proxy_server import proxy_config
config: Final = await proxy_config.get_config()
litellm_settings: Final = config.get("litellm_settings", {}) or {}
litellm_settings: Final = config.litellm_settings
return litellm_settings.get("scim_admin_group") or None
except Exception as e:
verbose_proxy_logger.warning("Error reading scim_admin_group setting, defaulting to None: %s", e)

11
litellm/proxy/proxy_cli.py
View file

@ -1195,10 +1195,9 @@ def run_server(
_config: Final = asyncio.run(proxy_config.get_config(config_file_path=config))
### LITELLM SETTINGS ###
litellm_settings = _config.get("litellm_settings", None)
litellm_settings = _config.litellm_settings
if (
litellm_settings is not None
and "json_logs" in litellm_settings
"json_logs" in litellm_settings
and litellm_settings["json_logs"] is True
):
import litellm
@ -1207,12 +1206,10 @@ def run_server(
litellm._turn_on_json()
### GENERAL SETTINGS ###
general_settings = _config.get("general_settings", {})
if general_settings is None:
general_settings = {}
general_settings = _config.general_settings
### LOAD KEY MANAGEMENT SETTINGS FIRST (needed for custom secret manager) ###
key_management_settings: Final = general_settings.get("key_management_settings", None)
if key_management_settings is not None:
if isinstance(key_management_settings, Mapping):
import litellm
litellm._key_management_settings = KeyManagementSettings(**key_management_settings)

2
litellm/proxy/read_model_list.py
View file

@ -26,4 +26,4 @@ def read_model_list(config_path: str) -> list[dict[str, Any]]:
from litellm.proxy.proxy_server import ProxyConfig
config: Final = asyncio.run(ProxyConfig().get_config(config_file_path=config_path))
return config.get("model_list") or []
return [dict(model) for model in config.model_list] # mutable-ok: callers consume plain dicts

112
litellm/proxy/ui_crud_endpoints/proxy_setting_endpoints.py
View file

@ -517,7 +517,7 @@ class WebSearchInterceptionSettingsResponse(SettingsResponse):
)
def _with_websearch_enabled_resolved(config: Mapping[str, object]) -> dict[str, object]:
def _with_websearch_enabled_resolved(config: ProxyRuntimeConfig) -> ProxyRuntimeConfig:
"""
Answer with the stored flag when there is one, and only otherwise with what
this process is running.
@ -535,19 +535,19 @@ def _with_websearch_enabled_resolved(config: Mapping[str, object]) -> dict[str,
WebSearchInterceptionLogger,
)
litellm_settings: Final[Mapping[str, object]] = _as_settings_section(config.get("litellm_settings"))
litellm_settings: Final[Mapping[str, object]] = config.litellm_settings
stored: Final[Mapping[str, object]] = _as_settings_section(litellm_settings.get("websearch_interception_params"))
if "enabled" in stored:
return dict(config)
return config
resolved: Final = {
resolved: Final = { # mutable-ok: replacement value for the stored section
**stored,
"enabled": bool(litellm.logging_callback_manager.get_custom_loggers_for_type(WebSearchInterceptionLogger)),
}
return {
**config,
"litellm_settings": {**litellm_settings, "websearch_interception_params": resolved},
}
return config.with_section(
"litellm_settings",
{**litellm_settings, "websearch_interception_params": resolved}, # mutable-ok: replacement section for save_config
)
def _as_settings_section(value: object) -> Mapping[str, object]:
@ -616,24 +616,24 @@ async def add_allowed_ip(
# Load existing config
config: Final = await proxy_config.get_config()
verbose_proxy_logger.debug("Loaded config: %s", config)
if "general_settings" not in config:
config["general_settings"] = {}
if "allowed_ips" not in config["general_settings"]:
config["general_settings"]["allowed_ips"] = []
before_allowed_ips: Final = tuple(config.general_settings.get("allowed_ips") or ())
updated_allowed_ips: Final = (
(*before_allowed_ips, ip_address.ip) if ip_address.ip not in before_allowed_ips else before_allowed_ips
)
updated: Final = config.with_section(
"general_settings",
{**config.general_settings, "allowed_ips": list(updated_allowed_ips)}, # mutable-ok: replacement section for save_config
)
before_allowed_ips: Final = list(config["general_settings"]["allowed_ips"])
if ip_address.ip not in config["general_settings"]["allowed_ips"]:
config["general_settings"]["allowed_ips"].append(ip_address.ip)
await proxy_config.save_config(new_config=config)
await proxy_config.save_config(new_config=updated)
asyncio.create_task(
create_config_audit_log(
param_name="general_settings",
action="updated",
before_value={"allowed_ips": before_allowed_ips},
after_value={"allowed_ips": config["general_settings"]["allowed_ips"]},
before_value={"allowed_ips": list(before_allowed_ips)}, # mutable-ok: audit log wants a JSON list
after_value={"allowed_ips": list(updated_allowed_ips)}, # mutable-ok: audit log wants a JSON list
user_api_key_dict=user_api_key_dict,
)
)
@ -667,24 +667,22 @@ async def delete_allowed_ip(
# Load existing config
config: Final = await proxy_config.get_config()
verbose_proxy_logger.debug("Loaded config: %s", config)
if "general_settings" not in config:
config["general_settings"] = {}
if "allowed_ips" not in config["general_settings"]:
config["general_settings"]["allowed_ips"] = []
before_allowed_ips: Final = tuple(config.general_settings.get("allowed_ips") or ())
updated_allowed_ips: Final = tuple(ip for ip in before_allowed_ips if ip != ip_address.ip)
updated: Final = config.with_section(
"general_settings",
{**config.general_settings, "allowed_ips": list(updated_allowed_ips)}, # mutable-ok: replacement section for save_config
)
before_allowed_ips: Final = list(config["general_settings"]["allowed_ips"])
if ip_address.ip in config["general_settings"]["allowed_ips"]:
config["general_settings"]["allowed_ips"].remove(ip_address.ip)
await proxy_config.save_config(new_config=config)
await proxy_config.save_config(new_config=updated)
asyncio.create_task(
create_config_audit_log(
param_name="general_settings",
action="deleted",
before_value={"allowed_ips": before_allowed_ips},
after_value={"allowed_ips": config["general_settings"]["allowed_ips"]},
before_value={"allowed_ips": list(before_allowed_ips)}, # mutable-ok: audit log wants a JSON list
after_value={"allowed_ips": list(updated_allowed_ips)}, # mutable-ok: audit log wants a JSON list
user_api_key_dict=user_api_key_dict,
)
)
@ -751,7 +749,7 @@ def _root_schema(settings_class: type[BaseModel]) -> _RootSchema:
async def _get_settings_with_schema(
settings_key: str,
settings_class: type[BaseModel],
config: dict,
config: ProxyRuntimeConfig | Mapping[str, object],
) -> dict:
"""
Common utility function to get settings with schema information.
@ -759,10 +757,15 @@ async def _get_settings_with_schema(
Args:
settings_key: The key in litellm_settings to get
settings_class: The Pydantic class to use for schema
config: The config dictionary
config: The resolved proxy config, or a config-shaped mapping
"""
litellm_settings: Final = config.get("litellm_settings", {}) or {}
settings_data: Final = litellm_settings.get(settings_key, {}) or {}
litellm_settings: Final[Mapping[str, object]] = (
config.litellm_settings
if isinstance(config, ProxyRuntimeConfig)
else _as_settings_section(config.get("litellm_settings"))
)
raw_settings_data: Final = litellm_settings.get(settings_key, {}) or {}
settings_data: Final = raw_settings_data if isinstance(raw_settings_data, Mapping) else MappingProxyType({})
# Create the settings object
settings: Final = settings_class(**(settings_data))
@ -990,19 +993,18 @@ async def _update_litellm_setting(
# because get_config() may overwrite litellm.<key> with stale DB values
# via LITELLM_SETTINGS_SAFE_DB_OVERRIDES.
config: Final = await proxy_config.get_config()
before_value: Final = config.get("litellm_settings", {}).get(settings_key)
before_value: Final = config.litellm_settings.get(settings_key)
# Update the in-memory settings (after get_config to avoid stale override)
setattr(litellm, settings_key, in_memory_var)
# Update config with new settings
if "litellm_settings" not in config:
config["litellm_settings"] = {}
config["litellm_settings"][settings_key] = in_memory_var
# Save the updated config
await proxy_config.save_config(new_config=config)
await proxy_config.save_config(
new_config=config.with_section(
"litellm_settings",
{**config.litellm_settings, settings_key: in_memory_var}, # mutable-ok: replacement section for save_config
)
)
# Fire-and-forget so an audit-log failure (transient DB blip, etc.)
# never surfaces as a 500 after save_config has already committed,
@ -1185,18 +1187,7 @@ async def update_sso_settings(
if isinstance(stored, dict):
before_sso_data = proxy_config._decrypt_db_variables(stored)
# Load existing config
config: Final = await proxy_config.get_config()
# Update config with new environment variables
if "environment_variables" not in config:
config["environment_variables"] = {}
# Update general_settings for user_email (admin email)
if "general_settings" not in config:
config["general_settings"] = {}
# Update environment variables in config and in memory
# Update environment variables in memory
sso_data: Final = sso_config.model_dump()
for field_name, value in sso_data.items():
if field_name in SSO_FIELD_ENV_VARS:
@ -1363,15 +1354,16 @@ async def update_ui_theme_settings(
# Load existing config
config: Final = await proxy_config.get_config()
before_theme: Final = config.get("litellm_settings", {}).get("ui_theme_config")
before_theme: Final = config.litellm_settings.get("ui_theme_config")
# Convert theme config to dict
theme_data: Final = 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
updated: Final = config.with_section(
"litellm_settings",
{**config.litellm_settings, "ui_theme_config": theme_data}, # mutable-ok: replacement section for save_config
)
# The vars below are the only environment variables this endpoint owns, and
# they must stay in step with _UI_THEME_FIELD_ENV_VARS. A non-empty value
@ -1395,7 +1387,7 @@ async def update_ui_theme_settings(
# Persist the theme config (litellm_settings). save_config defaults to
# include_env_vars=False, so it does not snapshot environment_variables.
await proxy_config.save_config(new_config=config)
await proxy_config.save_config(new_config=updated)
# Persist only the two owned env vars, merged against the existing DB row.
await proxy_config.save_environment_variables(env_updates)
@ -1731,7 +1723,7 @@ async def get_ui_settings():
await user_api_key_cache.async_set_cache(key=UI_SETTINGS_CACHE_KEY, value=ui_settings, ttl=UI_SETTINGS_CACHE_TTL)
# Build config-like object for schema helper
config: Final[dict[str, object]] = {"litellm_settings": {"ui_settings": ui_settings}}
config: Final[Mapping[str, object]] = {"litellm_settings": {"ui_settings": ui_settings}} # mutable-ok: schema helper only reads it
settings: Final = await _get_settings_with_schema(
settings_key="ui_settings",