Overview
The AWS Secret Manager V2 module provides a full-featured async/sync integration with AWS Secrets Manager, supporting secret reading, writing, updating (PutSecretValue), deleting, and rotating with SigV4-signed HTTP requests.
Description
This module provides the AWSSecretsManagerV2 class, which extends both BaseAWSLLM (for AWS credential handling and SigV4 signing) and BaseSecretManager (for the standard secret manager interface). Unlike the V1 module that uses boto3 directly, V2 uses raw HTTP requests with SigV4 authentication, enabling fully async operations via httpx. The class supports multiple AWS authentication methods including IAM roles, STS assume-role, web identity tokens, and profiles. It implements all CRUD operations for secrets: GetSecretValue, CreateSecret, PutSecretValue, and DeleteSecret. It also supports reading from a "primary secret" (a JSON-structured secret containing multiple key-value pairs) and smart secret rotation that uses in-place PutSecretValue when the secret name does not change, avoiding ResourceExistsException.
Usage
Import and configure this class when using AWS Secrets Manager as the key management backend for the LiteLLM proxy, typically via key_management_system: aws_secret_manager in the proxy config.
Code Reference
Source Location
litellm/secret_managers/aws_secret_manager_v2.py
Class: AWSSecretsManagerV2
class AWSSecretsManagerV2(BaseAWSLLM, BaseSecretManager):
def __init__(
self,
aws_region_name: Optional[str] = None,
aws_role_name: Optional[str] = None,
aws_session_name: Optional[str] = None,
aws_external_id: Optional[str] = None,
aws_profile_name: Optional[str] = None,
aws_web_identity_token: Optional[str] = None,
aws_sts_endpoint: Optional[str] = None,
**kwargs,
):
Key Methods
| Method |
Signature |
Description
|
load_aws_secret_manager |
@classmethod def load_aws_secret_manager(cls, use_aws_secret_manager, key_management_settings=None) |
Class factory that initializes and registers the V2 client with LiteLLM
|
async_read_secret |
async def async_read_secret(self, secret_name, optional_params=None, timeout=None, primary_secret_name=None) -> Optional[str] |
Reads a secret value asynchronously; supports primary secret lookups
|
sync_read_secret |
def sync_read_secret(self, secret_name, optional_params=None, timeout=None, primary_secret_name=None) -> Optional[str] |
Reads a secret value synchronously; avoids infinite loops for AWS credential env vars
|
async_write_secret |
async def async_write_secret(self, secret_name, secret_value, description=None, optional_params=None, timeout=None, tags=None) -> dict |
Creates a new secret with optional tags (CreateSecret)
|
async_put_secret_value |
async def async_put_secret_value(self, secret_name, secret_value, optional_params=None, timeout=None) -> dict |
Updates an existing secret's value in place (PutSecretValue)
|
async_rotate_secret |
async def async_rotate_secret(self, current_secret_name, new_secret_name, new_secret_value, optional_params=None, timeout=None) -> dict |
Rotates a secret; uses PutSecretValue for in-place updates, or create+delete for name changes
|
async_delete_secret |
async def async_delete_secret(self, secret_name, recovery_window_in_days=7, optional_params=None, timeout=None) -> dict |
Deletes a secret with configurable recovery window
|
_prepare_request |
def _prepare_request(self, action, secret_name, secret_value=None, optional_params=None, request_data=None) -> tuple[str, Any, bytes] |
Builds and SigV4-signs an AWS Secrets Manager HTTP request
|
Import
from litellm.secret_managers.aws_secret_manager_v2 import AWSSecretsManagerV2
I/O Contract
Inputs (async_read_secret)
| Parameter |
Type |
Description
|
secret_name |
str |
Name/ID of the secret to read
|
optional_params |
Optional[dict] |
Additional AWS parameters (region, role, etc.)
|
timeout |
Optional[Union[float, httpx.Timeout]] |
Request timeout
|
primary_secret_name |
Optional[str] |
If set, reads the primary secret and extracts the value for secret_name as a key
|
Outputs (async_read_secret)
| Return Type |
Description
|
Optional[str] |
The secret string value, or None if not found or on error
|
Inputs (async_write_secret)
| Parameter |
Type |
Description
|
secret_name |
str |
Name for the new secret
|
secret_value |
str |
Value to store
|
description |
Optional[str] |
Optional description
|
tags |
Optional[Union[dict, list]] |
Tags as dict ({"Key": "Value"}) or AWS-format list
|
Outputs (async_write_secret)
| Return Type |
Description
|
dict |
AWS response containing ARN, Name, VersionId
|
Usage Examples
from litellm.secret_managers.aws_secret_manager_v2 import AWSSecretsManagerV2
# Initialize with IAM role
client = AWSSecretsManagerV2(
aws_region_name="us-east-1",
aws_role_name="arn:aws:iam::123456789:role/my-role",
aws_session_name="litellm-session",
)
# Read a secret
value = await client.async_read_secret(secret_name="my-api-key")
# Read from a primary secret (JSON bundle)
value = await client.async_read_secret(
secret_name="OPENAI_API_KEY",
primary_secret_name="litellm/production-keys",
)
# Write a new secret with tags
response = await client.async_write_secret(
secret_name="litellm/new-key",
secret_value="sk-abc123",
description="OpenAI API key for production",
tags={"Environment": "Production", "ManagedBy": "LiteLLM"},
)
# Rotate a secret (in-place update)
response = await client.async_rotate_secret(
current_secret_name="litellm/api-key",
new_secret_name="litellm/api-key", # same name = PutSecretValue
new_secret_value="sk-new-value",
)
# Delete a secret
response = await client.async_delete_secret(
secret_name="litellm/old-key",
recovery_window_in_days=7,
)
Related Pages
