Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:BerriAI Litellm Base Secret Manager

From Leeroopedia
Attribute Value
Sources litellm/secret_managers/base_secret_manager.py
Domains Secret Management, Abstract Base Class, Interface
last_updated 2026-02-15 16:00 GMT

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

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment