Implementation:BerriAI Litellm Custom Secret Manager Base

Attribute	Value
Sources	litellm/integrations/custom_secret_manager.py
Domains	Secret Management, Integrations, Extensibility
Last Updated	2026-02-15 16:00 GMT

Overview

CustomSecretManager is an abstract base class that provides a standard interface for implementing custom secret management integrations in LiteLLM, allowing users to plug in any secret vault or credential store.

Description

The CustomSecretManager class extends BaseSecretManager and defines the contract that all custom secret manager implementations must follow. It requires subclasses to implement both synchronous (sync_read_secret) and asynchronous (async_read_secret) methods for reading secrets. Optionally, subclasses can override async_write_secret, async_delete_secret, validate_environment, and async_health_check methods for write, delete, validation, and health check operations respectively. The class is designed to integrate seamlessly with LiteLLM's key management system by assigning a custom secret manager instance to litellm.secret_manager_client.

Usage

Import and subclass CustomSecretManager when you need to integrate a custom secret vault (e.g., HashiCorp Vault, CyberArk, custom databases) with LiteLLM. After implementing the required abstract methods, set the instance as litellm.secret_manager_client and configure litellm._key_management_system = KeyManagementSystem.CUSTOM.

Code Reference

Source Location

litellm/integrations/custom_secret_manager.py

Signature

class CustomSecretManager(BaseSecretManager):
    def __init__(self, secret_manager_name: Optional[str] = None, **kwargs)
    async def async_read_secret(self, secret_name: str, optional_params: Optional[dict] = None, timeout: Optional[Union[float, httpx.Timeout]] = None) -> Optional[str]  # abstract
    def sync_read_secret(self, secret_name: str, optional_params: Optional[dict] = None, timeout: Optional[Union[float, httpx.Timeout]] = None) -> Optional[str]  # abstract
    async def async_write_secret(self, secret_name: str, secret_value: str, description: Optional[str] = None, optional_params: Optional[dict] = None, timeout: Optional[Union[float, httpx.Timeout]] = None, tags: Optional[Union[dict, list]] = None) -> Dict[str, Any]
    async def async_delete_secret(self, secret_name: str, recovery_window_in_days: Optional[int] = 7, optional_params: Optional[dict] = None, timeout: Optional[Union[float, httpx.Timeout]] = None) -> dict
    def validate_environment(self) -> bool
    async def async_health_check(self, timeout: Optional[Union[float, httpx.Timeout]] = None) -> bool

Import

from litellm.integrations.custom_secret_manager import CustomSecretManager

I/O Contract

Inputs

Parameter	Type	Description
secret_manager_name	Optional[str]	A descriptive name for the secret manager, used for logging and debugging. Defaults to "custom_secret_manager".
secret_name	str	The name or path of the secret to read, write, or delete.
secret_value	str	The value to store when writing a secret.
optional_params	Optional[dict]	Additional parameters specific to the secret manager implementation.
timeout	Optional[Union[float, httpx.Timeout]]	Request timeout for the operation.
description	Optional[str]	Description of the secret (for write operations).
tags	Optional[Union[dict, list]]	Tags to apply to the secret (for write operations).
recovery_window_in_days	Optional[int]	Days before permanent deletion (for delete operations). Defaults to 7.

Outputs

Method	Return Type	Description
async_read_secret / sync_read_secret	Optional[str]	The secret value if found, None otherwise.
async_write_secret	Dict[str, Any]	Response from the secret manager with write operation details.
async_delete_secret	dict	Response from the secret manager with deletion details.
validate_environment	bool	True if environment is valid.
async_health_check	bool	True if the secret manager is healthy.

Usage Examples

from litellm.integrations.custom_secret_manager import CustomSecretManager
import httpx

class MyVaultSecretManager(CustomSecretManager):
    def __init__(self, vault_url: str, token: str):
        super().__init__(secret_manager_name="my_vault")
        self.vault_url = vault_url
        self.token = token

    async def async_read_secret(self, secret_name, optional_params=None, timeout=None):
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.vault_url}/v1/secret/{secret_name}",
                headers={"X-Vault-Token": self.token},
                timeout=timeout,
            )
            return response.json()["data"]["value"]

    def sync_read_secret(self, secret_name, optional_params=None, timeout=None):
        with httpx.Client() as client:
            response = client.get(
                f"{self.vault_url}/v1/secret/{secret_name}",
                headers={"X-Vault-Token": self.token},
                timeout=timeout,
            )
            return response.json()["data"]["value"]

# Register the custom secret manager
import litellm
from litellm.types.secret_managers.main import KeyManagementSystem

litellm.secret_manager_client = MyVaultSecretManager(
    vault_url="https://vault.example.com",
    token="my-vault-token",
)
litellm._key_management_system = KeyManagementSystem.CUSTOM

Related Pages

• Implementation:BerriAI_Litellm_Custom_Auth_Template - Template for custom authentication that may use secrets
• Implementation:BerriAI_Litellm_Proxy_Auth_Credentials - Proxy authentication credentials that may leverage secret managers
• Implementation:BerriAI_Litellm_Custom_SSO_Template - Custom SSO handler that may require secret retrieval