Overview
The Base Secret Manager module defines the abstract base class that all LiteLLM secret manager implementations must extend, providing a standard interface for reading, writing, deleting, and rotating secrets.
Description
This module provides the BaseSecretManager abstract class (ABC) that establishes the contract for secret manager integrations. It defines four abstract methods that subclasses must implement: async_read_secret, sync_read_secret, async_write_secret, and async_delete_secret. Additionally, it provides a concrete async_rotate_secret method that implements rotation as a create-verify-delete sequence: it reads the current secret to verify it exists, creates a new secret with the new name and value, verifies the new secret was created, then deletes the old secret with a 7-day recovery window. This base rotation logic can be overridden by subclasses for provider-specific optimizations (e.g., AWS V2 uses PutSecretValue for in-place updates).
Usage
Import BaseSecretManager when creating a new secret manager integration. All concrete implementations (AWS Secrets Manager V2, HashiCorp Vault, CyberArk Conjur) inherit from this class.
Code Reference
Source Location
litellm/secret_managers/base_secret_manager.py
Class: BaseSecretManager
class BaseSecretManager(ABC):
"""Abstract base class for secret management implementations."""
Abstract Methods
| Method |
Signature |
Description
|
async_read_secret |
async def async_read_secret(self, secret_name: str, optional_params: Optional[dict] = None, timeout: Optional[Union[float, httpx.Timeout]] = None) -> Optional[str] |
Asynchronously reads a secret value
|
sync_read_secret |
def sync_read_secret(self, secret_name: str, optional_params: Optional[dict] = None, timeout: Optional[Union[float, httpx.Timeout]] = None) -> Optional[str] |
Synchronously reads a secret value
|
async_write_secret |
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] |
Asynchronously writes a secret
|
async_delete_secret |
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 |
Asynchronously deletes a secret
|
Concrete Methods
| Method |
Signature |
Description
|
async_rotate_secret |
async def async_rotate_secret(self, current_secret_name: str, new_secret_name: str, new_secret_value: str, optional_params: Optional[dict] = None, timeout: Optional[Union[float, httpx.Timeout]] = None) -> dict |
Rotates a secret by creating a new one and deleting the old one (default implementation)
|
Import
from litellm.secret_managers.base_secret_manager import BaseSecretManager
I/O Contract
Inputs (async_rotate_secret)
| Parameter |
Type |
Description
|
current_secret_name |
str |
Name of the existing secret to rotate
|
new_secret_name |
str |
New name for the rotated secret
|
new_secret_value |
str |
New value for the secret
|
optional_params |
Optional[dict] |
Additional provider-specific parameters
|
timeout |
Optional[Union[float, httpx.Timeout]] |
Request timeout
|
Outputs (async_rotate_secret)
| Return Type |
Description
|
dict |
Response from the write operation containing the new secret details
|
Raises ValueError |
If the current secret is not found or the new secret verification fails
|
Usage Examples
from litellm.secret_managers.base_secret_manager import BaseSecretManager
from typing import Any, Dict, Optional, Union
import httpx
class MyCustomSecretManager(BaseSecretManager):
async def async_read_secret(
self, secret_name: str,
optional_params: Optional[dict] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
) -> Optional[str]:
# Implementation for reading secrets
...
def sync_read_secret(
self, secret_name: str,
optional_params: Optional[dict] = None,
timeout: Optional[Union[float, httpx.Timeout]] = None,
) -> Optional[str]:
# Implementation for sync reading
...
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]:
# Implementation for writing secrets
...
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:
# Implementation for deleting secrets
...
# The async_rotate_secret method is inherited and works out of the box
Related Pages