What is AI Agent Secrets Management?
AI Agent Secrets Management is the systematic practice of securely storing, distributing, rotating, and auditing the sensitive credentials that autonomous AI agents rely on to perform their work. These secrets include API keys for large language model providers (OpenAI, Anthropic, Google), database connection strings, third-party service tokens, cloud provider access keys, signing certificates, and even internal service mesh credentials. Unlike traditional application secrets, AI agent secrets carry additional risk because agents often operate with elevated autonomy — they make decisions, chain actions, and access multiple services in unpredictable sequences. A compromised agent secret can cascade into data exfiltration, unauthorized model usage, or complete infrastructure takeover.
Why Secrets Management Matters for AI Agents
AI agents are fundamentally different from static microservices. An agent may dynamically select tools, query vector databases, invoke external APIs, and spawn sub-agents — all based on natural language instructions that could be manipulated through prompt injection. When secrets are hardcoded in agent code, configuration files, or prompt templates, they become trivially extractable. Beyond the immediate security risk, unmanaged secrets create operational nightmares: expired keys bring agents down silently, rotated credentials require manual restarts, and debugging secret-related failures becomes a scavenger hunt across scattered configuration files. Proper secrets management transforms these brittle, dangerous patterns into a resilient, auditable system where agents receive exactly the credentials they need, for exactly as long as they need them, with full cryptographic guarantees.
Real-World Risks
- Prompt injection exfiltration: An attacker crafts input that tricks an agent into echoing its system prompt, which may contain inline API keys.
- Tool-calling side effects: An agent with access to a file-writing tool might accidentally log secrets to disk.
- Sub-agent sprawl: A parent agent spawns child agents that inherit credentials without proper scope limitation.
- Vendor lock-in through expired keys: Hardcoded keys for a deprecated model provider break the agent silently when the key is revoked.
Core Components of a Secrets Management Architecture
A production-grade secrets management architecture for AI agents typically combines two complementary technologies: a secrets storage and distribution engine like HashiCorp Vault, and a Key Management Service (KMS) from your cloud provider that acts as the cryptographic root of trust. Vault handles the operational lifecycle of secrets — dynamic generation, leasing, rotation, revocation — while KMS provides envelope encryption so that even Vault's storage backend is protected by keys you control in the cloud provider's hardware security modules.
1. HashiCorp Vault
Vault is an open-source secrets management platform that centralizes all secrets behind a unified API with strict access control policies. For AI agents, Vault's most powerful features are dynamic secrets (credentials that are generated on-demand with a time-to-live), fine-grained ACLs via policies, and audit logging that records every single secret access. Instead of an agent holding a permanent database password, it requests a temporary credential from Vault that automatically expires, limiting the blast radius of any compromise.
Setting Up Vault for AI Agent Secrets
First, install Vault in development mode for testing. In production, you would run a clustered deployment backed by integrated storage (Raft) or a dedicated backend like Consul.
# Start Vault in development mode (never use for production)
vault server -dev -dev-root-token-id="root" &
# Set the VAULT_ADDR and VAULT_TOKEN environment variables
export VAULT_ADDR='http://127.0.0.1:8200'
export VAULT_TOKEN='root'
# Enable the key-value v2 secrets engine at the path "ai-agent"
vault secrets enable -path=ai-agent kv-v2
# Store secrets for an AI agent named "research-agent"
vault kv put ai-agent/research-agent \
openai_api_key="sk-proj-abc123..." \
anthropic_api_key="sk-ant-xyz789..." \
pinecone_api_key="pc-123456..." \
database_url="postgresql://user:pass@host:5432/vectordb"
Retrieving Secrets from Vault in Python
Here is a complete Python module that an AI agent can use to fetch secrets from Vault at runtime. It uses the hvac library and implements retry logic for resilience.
import hvac
import os
import time
from functools import lru_cache
class VaultSecretsClient:
"""
A secrets client for AI agents that retrieves credentials from HashiCorp Vault.
Secrets are fetched once and cached for the duration of the agent's execution.
"""
def __init__(self, vault_url=None, vault_token=None, mount_point="ai-agent"):
self.vault_url = vault_url or os.environ.get("VAULT_ADDR", "http://127.0.0.1:8200")
self.vault_token = vault_token or os.environ.get("VAULT_TOKEN")
self.mount_point = mount_point
self.client = hvac.Client(url=self.vault_url, token=self.vault_token)
if not self.client.is_authenticated():
raise RuntimeError("Vault client is not authenticated. Check VAULT_TOKEN.")
def get_agent_secrets(self, agent_name: str, max_retries: int = 3) -> dict:
"""
Fetch all secrets for a named agent from the KV store.
Returns a dictionary of secret key-value pairs.
"""
path = f"{self.mount_point}/data/{agent_name}"
for attempt in range(max_retries):
try:
response = self.client.secrets.kv.v2.read_secret_version(
path=path,
mount_point=self.mount_point
)
return response["data"]["data"]
except hvac.exceptions.Forbidden:
raise RuntimeError(
f"Access denied to secrets for agent '{agent_name}'. "
f"Verify Vault policy permits read on {path}."
)
except hvac.exceptions.VaultDown as e:
if attempt == max_retries - 1:
raise RuntimeError(f"Vault is unreachable after {max_retries} attempts: {e}")
time.sleep(2 ** attempt)
return {}
def get_secret(self, agent_name: str, secret_key: str) -> str:
"""
Fetch a single secret value for a named agent.
"""
secrets = self.get_agent_secrets(agent_name)
if secret_key not in secrets:
raise KeyError(
f"Secret '{secret_key}' not found for agent '{agent_name}'. "
f"Available keys: {list(secrets.keys())}"
)
return secrets[secret_key]
# Usage example within an AI agent runtime
def initialize_agent_credentials():
client = VaultSecretsClient()
# Fetch all secrets for this agent instance
secrets = client.get_agent_secrets("research-agent")
# Inject into environment or agent context
os.environ["OPENAI_API_KEY"] = secrets["openai_api_key"]
os.environ["ANTHROPIC_API_KEY"] = secrets["anthropic_api_key"]
os.environ["PINECONE_API_KEY"] = secrets["pinecone_api_key"]
print(f"Loaded {len(secrets)} secrets for research-agent")
return secrets
Dynamic Secrets for AI Agent Databases
One of Vault's most powerful features is the ability to generate temporary database credentials. This is dramatically safer than storing static credentials. The agent requests a credential, uses it, and the credential automatically expires within a short TTL.
# Enable the PostgreSQL secrets engine
vault secrets enable database
# Configure Vault to connect to PostgreSQL and create temporary roles
vault write database/config/postgres-db \
plugin_name="postgresql-database-plugin" \
allowed_roles="agent-readonly,agent-readwrite" \
connection_url="postgresql://{{username}}:{{password}}@postgres-host:5432/vectordb" \
username="vault_admin" \
password="vault_admin_password"
# Create a role that generates temporary read-only credentials (TTL: 1 hour)
vault write database/roles/agent-readonly \
db_name="postgres-db" \
creation_statements="CREATE ROLE \"{{name}}\" WITH LOGIN PASSWORD '{{password}}' \
VALID UNTIL '{{expiration}}'; \
GRANT SELECT ON ALL TABLES IN SCHEMA public TO \"{{name}}\";" \
default_ttl="1h" \
max_ttl="4h"
# Now an AI agent requests temporary credentials via the Vault API
vault read database/creds/agent-readonly
Here is the Python code an agent uses to obtain and manage these dynamic credentials, including proactive renewal before expiration:
import hvac
import psycopg2
import threading
import time
class DynamicDatabaseConnection:
"""
Manages a database connection backed by Vault dynamic secrets.
Automatically renews the lease before expiration.
"""
def __init__(self, vault_client, role_name="agent-readonly"):
self.vault_client = vault_client
self.role_name = role_name
self.credentials = None
self.lease_id = None
self.lease_duration = None
self.connection = None
self._renewal_thread = None
self._stop_renewal = threading.Event()
def connect(self):
"""Obtain dynamic credentials from Vault and establish a connection."""
response = self.vault_client.secrets.database.generate_credentials(
name=self.role_name
)
self.credentials = response["data"]
self.lease_id = response["lease_id"]
self.lease_duration = response["lease_duration"]
self.connection = psycopg2.connect(
host="postgres-host",
database="vectordb",
user=self.credentials["username"],
password=self.credentials["password"]
)
# Start background renewal at 80% of TTL
renewal_interval = int(self.lease_duration * 0.8)
self._renewal_thread = threading.Thread(
target=self._renew_lease, daemon=True
)
self._renewal_thread.start()
print(f"Obtained dynamic credentials. Lease: {self.lease_id}, "
f"TTL: {self.lease_duration}s, Renewing every {renewal_interval}s")
return self.connection
def _renew_lease(self):
"""Background thread that renews the lease before expiration."""
while not self._stop_renewal.is_set():
time.sleep(int(self.lease_duration * 0.8))
try:
self.vault_client.sys.renew_lease(
lease_id=self.lease_id,
increment=self.lease_duration
)
print(f"Renewed lease {self.lease_id}")
except Exception as e:
print(f"Lease renewal failed: {e}. Reconnecting...")
self.connect()
def close(self):
"""Revoke the lease and close the connection."""
self._stop_renewal.set()
if self.connection:
self.connection.close()
if self.lease_id:
self.vault_client.sys.revoke_lease(self.lease_id)
print(f"Revoked lease {self.lease_id}")
2. Cloud KMS (Key Management Services)
While Vault manages the operational lifecycle of secrets, KMS provides the cryptographic foundation. KMS services (AWS KMS, Google Cloud KMS, Azure Key Vault) manage master encryption keys in hardware security modules (HSMs) that are FIPS 140-2 certified. In a defense-in-depth architecture, you use KMS to encrypt Vault's storage backend (envelope encryption) and to encrypt individual secrets at the application layer before they even reach Vault. This means even if Vault's storage is compromised, the attacker cannot decrypt secrets without access to the KMS key — which itself is protected by cloud IAM.
AWS KMS: Envelope Encryption for Agent Secrets
The following example demonstrates envelope encryption: you generate a data encryption key (DEK) using AWS KMS, encrypt your secrets with that DEK locally, and then store the encrypted DEK alongside the ciphertext. Only entities with kms:Decrypt permission can recover the plaintext DEK and thus the secrets.
import boto3
import json
import os
from cryptography.fernet import Fernet
import base64
class AWSKMSEnvelopeEncryption:
"""
Implements envelope encryption using AWS KMS.
A KMS-managed Customer Master Key (CMK) protects a data encryption key (DEK)
that encrypts the actual secrets payload.
"""
def __init__(self, kms_key_id, region="us-east-1"):
self.kms_client = boto3.client("kms", region_name=region)
self.kms_key_id = kms_key_id
def encrypt_secrets_payload(self, secrets_dict: dict) -> dict:
"""
Encrypts a dictionary of secrets using envelope encryption.
Returns a JSON-serializable envelope containing the encrypted DEK and ciphertext.
"""
# 1. Generate a data encryption key (DEK) from KMS
response = self.kms_client.generate_data_key(
KeyId=self.kms_key_id,
KeySpec="AES_256"
)
plaintext_dek = response["Plaintext"] # 32 bytes of raw key material
encrypted_dek = response["CiphertextBlob"] # DEK encrypted under the CMK
# 2. Encrypt the secrets payload with the plaintext DEK using Fernet
# Fernet requires a base64-encoded 32-byte key
fernet_key = base64.urlsafe_b64encode(plaintext_dek)
fernet = Fernet(fernet_key)
secrets_json = json.dumps(secrets_dict).encode("utf-8")
encrypted_payload = fernet.encrypt(secrets_json)
# 3. Return the envelope
return {
"encrypted_dek": base64.b64encode(encrypted_dek).decode("utf-8"),
"encrypted_payload": base64.b64encode(encrypted_payload).decode("utf-8"),
"kms_key_arn": self.kms_key_id
}
def decrypt_secrets_payload(self, envelope: dict) -> dict:
"""
Decrypts an envelope back to the original secrets dictionary.
Requires kms:Decrypt permission on the CMK.
"""
encrypted_dek = base64.b64decode(envelope["encrypted_dek"])
encrypted_payload = base64.b64decode(envelope["encrypted_payload"])
# 1. Decrypt the DEK using KMS
response = self.kms_client.decrypt(
KeyId=self.kms_key_id,
CiphertextBlob=encrypted_dek
)
plaintext_dek = response["Plaintext"]
# 2. Decrypt the payload with the recovered DEK
fernet_key = base64.urlsafe_b64encode(plaintext_dek)
fernet = Fernet(fernet_key)
decrypted_json = fernet.decrypt(encrypted_payload)
return json.loads(decrypted_json.decode("utf-8"))
# Usage: Encrypt secrets before storing in Vault or S3
kms = AWSKMSEnvelopeEncryption(kms_key_id="arn:aws:kms:us-east-1:123456789012:key/abc-def-1234")
agent_secrets = {
"openai_api_key": "sk-secret-value",
"database_password": "supersecurepassword"
}
# Encrypt
envelope = kms.encrypt_secrets_payload(agent_secrets)
print(f"Encrypted envelope: {json.dumps(envelope, indent=2)}")
# Later: decrypt (requires AWS IAM permission)
decrypted = kms.decrypt_secrets_payload(envelope)
print(f"Decrypted secrets: {list(decrypted.keys())}")
Google Cloud KMS: Symmetric Encryption for Agent Secrets
Google Cloud KMS offers a similar envelope encryption pattern. Here is a complete example using the Python client library with symmetric encryption.
from google.cloud import kms
import google_crc32c
import json
class GoogleCloudKMSSecrets:
"""
Encrypts and decrypts agent secrets using Google Cloud KMS symmetric keys.
"""
def __init__(self, project_id, location_id, key_ring_id, key_id):
self.client = kms.KeyManagementServiceClient()
self.key_name = self.client.crypto_key_path(
project_id, location_id, key_ring_id, key_id
)
def encrypt_secrets(self, secrets_dict: dict) -> bytes:
"""
Encrypts a secrets dictionary into ciphertext bytes using Cloud KMS.
"""
plaintext = json.dumps(secrets_dict).encode("utf-8")
# Compute CRC32C checksum for integrity verification
plaintext_crc32c = google_crc32c.Checksum(plaintext).intdigest()
response = self.client.encrypt(
request={
"name": self.key_name,
"plaintext": plaintext,
"plaintext_crc32c": plaintext_crc32c,
}
)
# Verify the response integrity
if not response.verified_plaintext_crc32c:
raise ValueError("Encryption response CRC32C mismatch - possible tampering")
return response.ciphertext
def decrypt_secrets(self, ciphertext: bytes) -> dict:
"""
Decrypts ciphertext back to the original secrets dictionary.
"""
ciphertext_crc32c = google_crc32c.Checksum(ciphertext).intdigest()
response = self.client.decrypt(
request={
"name": self.key_name,
"ciphertext": ciphertext,
"ciphertext_crc32c": ciphertext_crc32c,
}
)
plaintext = response.plaintext.decode("utf-8")
return json.loads(plaintext)
# Usage
gcp_kms = GoogleCloudKMSSecrets(
project_id="my-ai-project",
location_id="global",
key_ring_id="agent-secrets-ring",
key_id="agent-encryption-key"
)
secrets = {"anthropic_api_key": "sk-ant-secret123", "db_password": "secure-db-pass"}
encrypted_blob = gcp_kms.encrypt_secrets(secrets)
print(f"Encrypted to {len(encrypted_blob)} bytes (ciphertext)")
decrypted = gcp_kms.decrypt_secrets(encrypted_blob)
print(f"Decrypted successfully: {list(decrypted.keys())}")
Azure Key Vault: Secrets as a Service
Azure Key Vault combines secrets storage and key management into a single managed service. Secrets are stored directly in the vault with versioning and access policies.
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import os
class AzureKeyVaultAgentSecrets:
"""
Retrieves and manages AI agent secrets stored in Azure Key Vault.
Uses managed identity (DefaultAzureCredential) for zero-code authentication.
"""
def __init__(self, vault_url=None):
self.vault_url = vault_url or os.environ.get(
"AZURE_KEY_VAULT_URL", "https://agent-vault.vault.azure.net/"
)
credential = DefaultAzureCredential()
self.client = SecretClient(vault_url=self.vault_url, credential=credential)
def get_agent_secret(self, secret_name: str) -> str:
"""
Retrieves the latest version of a secret from Azure Key Vault.
Secret names follow convention: agent-{agentname}-{provider}-apikey
"""
retrieved_secret = self.client.get_secret(secret_name)
return retrieved_secret.value
def set_agent_secret(self, secret_name: str, secret_value: str) -> None:
"""
Creates or updates a secret in the vault. New version is created automatically.
"""
self.client.set_secret(secret_name, secret_value)
def get_all_agent_secrets(self, agent_prefix: str) -> dict:
"""
Fetch all secrets matching a prefix for a given agent.
Example: agent_prefix = "agent-research-"
"""
secrets_dict = {}
secret_properties = self.client.list_secrets()
for secret_prop in secret_properties:
if secret_prop.name.startswith(agent_prefix):
secret = self.client.get_secret(secret_prop.name)
# Strip prefix to get a clean key name
clean_key = secret_prop.name[len(agent_prefix):]
secrets_dict[clean_key] = secret.value
return secrets_dict
# Usage with managed identity (no keys in code)
vault = AzureKeyVaultAgentSecrets()
openai_key = vault.get_agent_secret("agent-research-openai-apikey")
print(f"Retrieved secret with length: {len(openai_key)}")
Best Practices for AI Agent Secrets Management
The following best practices form a comprehensive security posture. They apply regardless of which specific tools (Vault, KMS, or managed services) you choose.
1. Never Hardcode Secrets
This is the cardinal rule. Secrets must never appear in source code, configuration files, Docker images, prompt templates, or agent memory. Even "example" or "test" secrets in comments can leak. Use a secrets manager as the single source of truth and fetch credentials at runtime only.
# BAD: Hardcoded secret in agent code
openai_api_key = "sk-proj-abc123realapikey" # NEVER DO THIS
# GOOD: Fetch at runtime from environment injected by secrets manager
import os
openai_api_key = os.environ.get("OPENAI_API_KEY")
if not openai_api_key:
raise RuntimeError("OPENAI_API_KEY not set — did secrets injection fail?")
2. Principle of Least Privilege
Each AI agent should receive exactly the secrets it needs and nothing more. Create per-agent Vault policies or per-agent KMS grants. A "research-agent" that only queries vector databases should never hold credentials to your billing system. If an agent is compromised, the blast radius is limited to its specific scope.
# Vault policy for research-agent — minimal access
# File: research-agent-policy.hcl
path "ai-agent/data/research-agent" {
capabilities = ["read"]
}
path "database/creds/agent-readonly" {
capabilities = ["read"]
}
# No access to other agents' secrets, no write access, no admin paths
3. Secrets Rotation
Rotate secrets regularly and automatically. Vault's dynamic secrets handle this natively — credentials expire and are regenerated. For static secrets (API keys that cannot be dynamically generated), implement scheduled rotation workflows that update the secret in the store and trigger agent re-authentication gracefully.
import schedule
import time
from vault_client import VaultSecretsClient
def rotate_openai_key():
"""
Scheduled job: generate a new OpenAI API key via their management API,
update it in Vault, and trigger agent restart or hot-reload.
"""
# 1. Call OpenAI API to create a new key
# new_key = openai_client.create_api_key(...)
# 2. Update Vault with the new key
client = VaultSecretsClient()
# vault kv patch ai-agent/research-agent openai_api_key=new_key
# 3. Notify agents to reload secrets (e.g., via webhook or signal)
print("Rotated OpenAI API key and notified agents")
# Run rotation every 30 days
schedule.every(30).days.do(rotate_openai_key)
while True:
schedule.run_pending()
time.sleep(3600)
4. Audit Logging & Monitoring
Enable Vault's audit log to track every secret access. Stream these logs to your SIEM (Security Information and Event Management) system. For cloud KMS, enable CloudTrail (AWS), Cloud Audit Logs (GCP), or Azure Activity Logs. Monitor for anomalous access patterns: an agent accessing secrets it never uses, access from unexpected IPs, or spike in secret retrieval frequency that could indicate exfiltration.
# Enable Vault audit log to file (production would use socket or syslog)
vault audit enable file file_path=/var/log/vault_audit.log
# Example audit log entry showing a secret read
# {
# "type": "response",
# "auth": {"client_token": "hmac-sha256:...", "policies": ["research-agent"]},
# "request": {
# "path": "ai-agent/data/research-agent",
# "operation": "read"
# },
# "timestamp": "2025-01-15T14:22:10Z"
# }
# Python snippet: Parse audit logs and alert on anomalies
def analyze_vault_audit_logs(log_file_path):
"""
Scans Vault audit logs for suspicious patterns.
"""
with open(log_file_path, 'r') as f:
for line in f:
entry = json.loads(line)
path = entry.get("request", {}).get("path", "")
operation = entry.get("request", {}).get("operation", "")
# Alert: Any access to billing-agent secrets from research-agent policy
if "billing-agent" in path and "research-agent" in str(entry.get("auth", {}).get("policies", [])):
trigger_alert(f"Cross-agent secret access detected: {entry}")
# Alert: Bulk secret reads in short time window
# (Implementation would track frequency per token)
5. Environment Variable Injection at Runtime
Secrets should be injected into the agent's runtime environment moments before execution, and never baked into container images or deployment artifacts. Use a secrets injection sidecar (like Vault Agent Injector for Kubernetes) or a pre-launch script that fetches secrets and exports them as environment variables with minimal exposure. The agent itself should validate that required secrets are present and fail fast with a clear error if injection has failed.
# Kubernetes deployment with Vault Agent Injector sidecar
# The injector mounts secrets to a tmpfs volume and sets env vars
apiVersion: apps/v1
kind: Deployment
metadata:
name: research-agent
annotations:
vault.hashicorp.com/agent-inject: "true"
vault.hashicorp.com/role: "research-agent"
vault.hashicorp.com/agent-inject-secret-openai: "ai-agent/data/research-agent"
vault.hashicorp.com/agent-inject-template-openai: |
{{ with secret "ai-agent/data/research-agent" }}
export OPENAI_API_KEY="{{ .Data.data.openai_api_key }}"
{{ end }}
spec:
template:
spec:
containers:
- name: agent
image: my-agent-image:latest # No secrets in image!
command: ["/bin/bash", "-c"]
args:
- |
# Source injected secrets and start agent
source /vault/secrets/openai
python -m agent.main
6. Encrypt Secrets at Rest and in Transit
Every link in the chain must be encrypted. Secrets in Vault's storage must be encrypted (Vault does this automatically with its barrier mechanism, but backing it with KMS-managed keys adds defense in depth). Secrets in transit must use TLS 1.3 exclusively. Secrets in agent memory should be held in memory-safe structures and zeroed out after use when possible. Never log secrets or include them in error messages or tracebacks.
# Zero-out sensitive environment variables after agent initialization
import os
import ctypes
def secure_clear_env_var(var_name: str):
"""
Remove a secret from the environment and attempt to clear
any residual memory (best-effort for Python).
"""
value = os.environ.get(var_name)
if value:
# Overwrite the value in the environ dict before deletion
os.environ[var_name] = '\x00' * len(value)
del os.environ[var_name]
return None
# After agent has loaded and validated secrets:
secure_clear_env_var("OPENAI_API_KEY")
secure_clear_env_var("ANTHROPIC_API_KEY")
Integrating Secrets Management into AI Agent Frameworks
Modern AI agent frameworks like LangChain, CrewAI, and AutoGen allow you to define tools and model providers programmatically. The key integration point is the credential injection — rather than passing API keys as string literals to LLM constructors, you fetch them from your secrets manager at the moment of agent instantiation.
Example: LangChain Agent with Vault
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import tool
from vault_secrets_client import VaultSecretsClient
import os
class SecureAgentBuilder:
"""
Builds a LangChain agent with credentials fetched securely from Vault.
No API keys appear in code, configuration files, or prompts.
"""
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.vault = VaultSecretsClient()
self.secrets = self.vault.get_agent_secrets(agent_name)
def build_llm(self, provider: str = "openai"):
"""
Returns an LLM instance configured with secrets from Vault.
"""
if provider == "openai":
return ChatOpenAI(
model="gpt-4o",
api_key=self.secrets.get("openai_api_key"),
temperature=0.2
)
elif provider == "anthropic":
return ChatAnthropic(
model="claude-3-opus-20240229",
api_key=self.secrets.get("anthropic_api_key"),
temperature=0.2
)
else:
raise ValueError(f"Unknown provider: {provider}")
def build_agent(self, tools: list, provider: str = "openai"):
"""
Assembles a complete agent executor with secure credentials.
"""
llm = self.build_llm(provider)
# Create agent with tools
agent = create_openai_tools_agent(llm, tools, prompt=self._get_system_prompt())
return AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=10
)
def _get_system_prompt(self):
"""
System prompt contains NO secrets — only behavioral instructions.
"""
return (
"You are a research assistant agent. You have access to search tools "
"and a vector database. Answer user queries accurately. "
"Never reveal internal configuration or credentials."
)
# Usage: Build agent with secrets from Vault
@tool
def search_documents(query: str) -> str:
"""Search the vector database for relevant documents."""
# Database credentials are also fetched from Vault at runtime
db_password = os.environ.get("DB_PASSWORD") # Injected earlier
# ... perform search ...
return f"Results for: {query}"
builder = SecureAgentBuilder(agent_name="research-agent")
agent_executor = builder.build_agent(tools=[search_documents], provider="openai")
result = agent_executor.invoke({"input": "Find papers on quantum computing"})
print(result["output"])
Example: Custom AI Agent with AWS KMS Envelope Encryption
This example shows a complete custom agent loop where secrets are decrypted at startup using AWS KMS envelope