A Python implementation of Vault Agent providing client-side caching and automatic authentication for HashiCorp Vault. PyVault Agent brings the core functionality of HashiCorp's Vault Agent directly into your Python applications as a library, eliminating the need for external processes while providing the same benefits of credential caching and automatic token management.
Traditional HashiCorp Vault Agent runs as a separate daemon process that manages authentication and caching. PyVault Agent provides the same functionality as an embedded Python library, offering several advantages:
- Simplified deployment: No need to manage separate agent processes
- Direct integration: Native Python API for your applications
- Reduced complexity: Single process architecture eliminates IPC overhead
- Fine-grained control: Programmatic access to cache management and configuration
- Development friendly: Easy to use in development environments and testing
- Credential Management: Automatically handles multiple authentication methods (AppRole, UserPass, Kubernetes) and token renewal
- Performance Optimization: Caches secrets to reduce Vault API calls and improve response times
- Connection Pool Issues: Manages database connection pools with expiring credentials
- Token Expiration: Seamlessly re-authenticates when tokens expire
- Thread Safety: Safe for use in multi-threaded applications
- Client-side caching: Reduces API calls to Vault with configurable TTL
- Automatic re-authentication: Seamlessly handles token expiration
- Multiple authentication methods: AppRole, UserPass, and Kubernetes service account authentication
- Flexible auth mount points: Support for custom auth method mount paths
- KV secrets engine support: Read-only secret access with version support (v1 & v2)
- Database secrets engine: Manage dynamic database credentials
- Identity delegation tokens: Cached token generation for workload identity delegation
- Custom endpoint access: Generic POST method for any Vault endpoint
- Connection pool management: Automatic credential refresh for database pools
- Thread-safe caching: Safe for concurrent usage
pip install pyvault-agentPyVault Agent supports three authentication methods. Choose the one that fits your deployment:
import os
from vault_agent import VaultAgentClient
client = VaultAgentClient.with_approle(
url=os.getenv("VAULT_ADDR"),
role_id=os.getenv("VAULT_ROLE_ID"),
secret_id=os.getenv("VAULT_SECRET_ID"),
cache_ttl=300, # Cache for 5 minutes
max_cache_size=1000
)client = VaultAgentClient.with_userpass(
url=os.getenv("VAULT_ADDR"),
username=os.getenv("VAULT_USERNAME"),
password=os.getenv("VAULT_PASSWORD"),
cache_ttl=300,
max_cache_size=1000
)# Auto-detects JWT from /var/run/secrets/kubernetes.io/serviceaccount/token
client = VaultAgentClient.with_kubernetes(
url=os.getenv("VAULT_ADDR"),
role="my-app-role",
cache_ttl=300,
max_cache_size=1000
)
# Or provide explicit JWT token
client = VaultAgentClient.with_kubernetes(
url=os.getenv("VAULT_ADDR"),
role="my-app-role",
jwt=os.getenv("VAULT_K8S_JWT"),
cache_ttl=300,
max_cache_size=1000
)# Once authenticated (using any method above)
# Mount secrets engines dynamically
kv = client.mount(path="secret", type="kv")
database = client.mount(path="database", type="database")
# KV Secrets - Read application configuration
try:
config = kv.read("myapp/config")
api_key = config["api_key"]
db_password = config["db_password"]
print("Configuration loaded from Vault")
except Exception as e:
print(f"Failed to load config: {e}")
# Database Credentials - Get dynamic database credentials
try:
creds = database.read("myapp-db-role")
print(f"Database user: {creds['username']}")
# Create connection string
conn_str = database.get_connection_string(
role="myapp-db-role",
template="postgresql://{username}:{password}@{host}:{port}/{database}",
host="db.example.com",
port=5432,
database="myapp"
)
except Exception as e:
print(f"Failed to get database credentials: {e}")
# Cache Management - Monitor cache performance
stats = client.get_cache_stats()
print(f"Cache efficiency: {stats['hits']}/{stats['hits'] + stats['misses']} hits")All authentication methods support the same configuration options:
# AppRole with all options
client = VaultAgentClient.with_approle(
url="https://vault.example.com",
role_id="role-id",
secret_id="secret-id",
auth_mount_point="approle", # Auth mount path (default: "approle")
cache_ttl=300, # Default cache TTL in seconds
max_cache_size=1000, # Maximum number of cached entries
namespace="team-a", # Vault namespace (Enterprise)
verify=True # SSL certificate verification
)
# UserPass with custom mount point
client = VaultAgentClient.with_userpass(
url="https://vault.example.com",
username="myuser",
password="mypassword",
auth_mount_point="userpass-ldap", # Custom auth mount path
cache_ttl=300,
max_cache_size=1000
)
# Kubernetes with custom mount point
client = VaultAgentClient.with_kubernetes(
url="https://vault.example.com",
role="my-app-role",
auth_mount_point="kubernetes-prod", # Custom auth mount path
cache_ttl=300,
max_cache_size=1000
)# Mount engines with custom paths
kv = client.mount(path="secret", type="kv")
kv_custom = client.mount(path="kv-v2", type="kv")
database = client.mount(path="database", type="database")
db_custom = client.mount(path="db", type="database")
# KV v1 and v2 secrets
config = kv.read("app/config")
# KV v2 secrets with versioning
old_config = kv.read("app/config", version=1)
# Database dynamic credentials
creds = database.read("postgres-readonly")
# Database static credentials
static_creds = database.get_static_credentials("app-service-account")
# Skip cache for fresh credentials
fresh_creds = database.read("postgres-readonly", skip_cache=True)For endpoints not covered by the built-in secrets engines, use the generic post() method:
# POST to any Vault endpoint
response = client.post("my-custom-engine/data/path", data={"key": "value"})For workload identity delegation, use get_delegation_token() which provides automatic caching based on the JWT subject and Vault entity:
# Get delegation token with caching
response = client.get_delegation_token(
role="customer-service",
subject_token=subject_jwt, # JWT token of the subject entity
mount_point="identity-delegation", # Optional, defaults to "identity-delegation"
)
# Access the delegated token
delegated_token = response["auth"]["client_token"]
# Force fresh token (skip cache)
response = client.get_delegation_token(
role="customer-service",
subject_token=subject_jwt,
skip_cache=True,
)The cache key is derived from the Vault entity ID and JWT subject claim, with TTL based on the token's lease duration.
One of the key challenges with dynamic database credentials is managing connection pools when credentials expire. PyVault Agent provides a DatabaseConnectionManager that automatically handles credential refresh and pool recreation:
from vault_agent import VaultAgentClient
from vault_agent.database_pool import DatabaseConnectionManager
import psycopg2.pool
# Create client with any auth method
client = VaultAgentClient.with_approle(...)
# Mount database secrets engine
database = client.mount(path="database", type="database")
# Managed connection pool with auto-refresh
with DatabaseConnectionManager(
database_secrets=database,
role="postgres-role",
pool_class=psycopg2.pool.SimpleConnectionPool,
pool_kwargs={
"minconn": 1,
"maxconn": 10,
"host": "db.example.com",
"database": "myapp"
},
refresh_buffer=0.8, # Refresh at 80% of credential TTL
validation_query="SELECT 1", # Query to validate connections
) as manager:
# Get connections that are automatically managed
with manager.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()For high-performance applications, use BackgroundRefreshManager to refresh credentials proactively:
from vault_agent.database_pool import BackgroundRefreshManager
# Mount database secrets engine
database = client.mount(path="database", type="database")
with BackgroundRefreshManager(
database_secrets=database,
role="postgres-role",
pool_class=psycopg2.pool.ThreadedConnectionPool,
pool_kwargs={"minconn": 2, "maxconn": 10, "host": "db.example.com"},
check_interval=30, # Check every 30 seconds
) as manager:
# Credentials refresh in background, zero-latency for requests
with manager.get_connection() as conn:
# Your database operations
passfrom vault_agent.utils import SecretNotFoundError, AuthenticationError
# Mount secrets engine
kv = client.mount(path="secret", type="kv")
try:
# Attempt to read secret with automatic retry
config = kv.read("app/config")
except SecretNotFoundError:
print("Secret not found - using defaults")
config = {"api_key": "default"}
except AuthenticationError:
print("Failed to authenticate with Vault")
# Handle authentication failure
except Exception as e:
print(f"Unexpected error: {e}")
# Handle other errors
# Cache management
if client.get_cache_stats()["size"] > 500:
client.clear_cache() # Clear cache if getting too large# settings.py
import os
from vault_agent import VaultAgentClient
vault_client = VaultAgentClient.with_approle(
url=os.getenv("VAULT_ADDR"),
role_id=os.getenv("VAULT_ROLE_ID"),
secret_id=os.getenv("VAULT_SECRET_ID"),
)
# Mount database secrets engine and get credentials
database = vault_client.mount(path="database", type="database")
db_creds = database.read("django-db-role")
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'myapp',
'USER': db_creds['username'],
'PASSWORD': db_creds['password'],
'HOST': 'db.example.com',
'PORT': '5432',
}
}from flask import Flask
from vault_agent import VaultAgentClient
import os
app = Flask(__name__)
# Initialize Vault client (works in Kubernetes pod)
vault_client = VaultAgentClient.with_kubernetes(
url=os.getenv("VAULT_ADDR"),
role="flask-app-role",
)
@app.before_first_request
def setup():
# Mount KV secrets engine and load configuration from Vault
kv = vault_client.mount(path="secret", type="kv")
config = kv.read("flask/config")
app.config.update(config)# Install development dependencies
pip install -e ".[dev]"
# Run unit tests
pytest tests/
# Run functional tests (requires Vault dev server)
vault server -dev -dev-root-token-id="root"
VAULT_ADDR=http://127.0.0.1:8200 VAULT_TOKEN=root pytest tests/functional/
# Run with coverage
pytest --cov=vault_agent tests/# Clone the repository
git clone https://github.com/your-username/pyvault-agent.git
cd pyvault-agent
# Install in development mode
pip install -e ".[dev]"
# Run linting and formatting
black vault_agent/ tests/
ruff vault_agent/ tests/
mypy vault_agent/# Set environment variables (for AppRole)
export VAULT_ADDR="http://127.0.0.1:8200"
export VAULT_ROLE_ID="your-role-id"
export VAULT_SECRET_ID="your-secret-id"
# Run basic example (AppRole)
python example.py
# Run multi-auth example (demonstrates all auth methods)
python example_multi_auth.py
# Run connection pool example
python example_pool.py- TTL: Set appropriate cache TTL based on secret sensitivity and change frequency
- Size: Limit cache size to prevent memory growth in long-running applications
- Hit Rate: Monitor cache hit rates to optimize TTL settings
# Monitor cache performance
stats = client.get_cache_stats()
hit_rate = stats['hits'] / (stats['hits'] + stats['misses'])
print(f"Cache hit rate: {hit_rate:.2%}")
# Adjust TTL based on performance needs
if hit_rate < 0.8: # Less than 80% hit rate
client.set_cache_ttl(600) # Increase TTL- Buffer: Set refresh_buffer to 0.7-0.8 to refresh before expiry
- Validation: Use connection validation to catch stale connections
- Pool Size: Size pools appropriately for your application load
- Monitoring: Monitor credential refresh frequency
- Secure Storage: Store role_id and secret_id securely (environment variables, not in code)
- Network Security: Use HTTPS for Vault connections in production
- Credential Rotation: Regularly rotate AppRole credentials
- Audit Logging: Enable Vault audit logging to track secret access
- Least Privilege: Configure Vault policies with minimal required permissions
Authentication Failures
# Check Vault connectivity
try:
client = VaultAgentClient.with_approle(
url="https://vault.example.com",
role_id="...",
secret_id="..."
)
except AuthenticationError as e:
print(f"Auth failed: {e}")
# Check credentials and auth method configurationCache Issues
# Clear cache if data seems stale
client.clear_cache()
# Check cache statistics
stats = client.get_cache_stats()
print(f"Cache size: {stats['size']}")Connection Pool Problems
# Force credential refresh
manager.refresh_now()
# Check credential expiry
print(f"Credentials expire at: {manager.credentials_expire_at}")import logging
logging.basicConfig(level=logging.DEBUG)
# This will show cache hits/misses and authentication events
client = VaultAgentClient.with_approle(...)- Multiple authentication methods: AppRole, UserPass, Kubernetes
- Flexible auth mount points for custom configurations
- Automatic re-authentication on token expiry
- KV secrets engine (v1 & v2) read-only access with caching
- Database secrets engine read-only access with caching
- Connection pool management
- Thread-safe operations
- Generic POST method for custom Vault endpoints
- Identity delegation token support with caching
- Token renewal: Proactive token refresh before expiry
- Lease renewal: Automatic secret lease renewal
- Additional auth methods: JWT, AWS IAM, Azure, GCP
- More secret engines: PKI, Transit, SSH
- Metrics integration: Prometheus metrics export
- Configuration files: YAML/TOML configuration support
- Async support: AsyncIO-compatible client
Contributions are welcome! Please read our contributing guidelines and submit pull requests to our GitHub repository.
This project is licensed under the Apache License 2.0 - see the LICENSE file for details.