Python SDK

pip install agentsecretstore
# or with Poetry:
poetry add agentsecretstore
# or with uv:
uv add agentsecretstore

export ASS_AGENT_KEY="ass_your_key_here"

import asyncio
import os
from agentsecretstore import AgentVault

async def main():
    vault = AgentVault(
        agent_key=os.environ["ASS_AGENT_KEY"],
        base_url="https://api.agentsecretstore.com",  # default
        default_ttl="1h",
        timeout=30.0,  # seconds, default: 30
        retries=3,      # default: 3
    )

    await vault.close()  # clean up connections

asyncio.run(main())

import asyncio
import os
from agentsecretstore import AgentVault

async def main():
    # Recommended: use as async context manager
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        value = await vault.get_secret("production/gemini/GEMINI_API_KEY")
        print(value)
    # connections closed automatically

asyncio.run(main())

from agentsecretstore import AgentVaultSync
import os

# For non-async code (e.g., scripts, Django views, legacy code)
vault = AgentVaultSync(agent_key=os.environ["ASS_AGENT_KEY"])

value = vault.get_secret("production/gemini/GEMINI_API_KEY")
print(value)

# All async methods have sync equivalents with identical signatures
vault.set_secret("staging/key", "value")
secrets = vault.list_secrets("production")

import asyncio
import os
from agentsecretstore import AgentVault

async def example():
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        # Basic retrieval
        value = await vault.get_secret("production/gemini/GEMINI_API_KEY")
        print(value)  # str: "gemini-api-key-example"

        metadata = await vault.get_secret_with_metadata("production/gemini/GEMINI_API_KEY")
        print(metadata.version)      # int: 1
        print(metadata.secret_type)  # "api_key"
        print(metadata.access_tier)  # "sensitive"
        print(metadata.metadata)     # dict

asyncio.run(example())

import asyncio
import os
from agentsecretstore import AgentVault

async def example():
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        # Full options
        result = await vault.set_secret(
            path="production/gemini/GEMINI_API_KEY",
            value="gemini-api-key-example",
            secret_type="api_key",
            access_tier="sensitive",
            metadata={"team": "ml", "env": "prod"},
        )
        print(result.version)

        # Minimal — just path + value
        await vault.set_secret("staging/feature-flag", "true")

asyncio.run(example())

import asyncio
import os
from agentsecretstore import AgentVault

async def example():
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        # List all secrets in a namespace prefix
        result = await vault.list_secrets("production/gemini")
        print(f"Total: {result.total}")

        for item in result.items:
            print(
                f"{item.namespace}/{item.key}  "
                f"v{item.version}  [{item.access_tier}] expired={item.expired}"
            )

asyncio.run(example())

import asyncio
import os
from pathlib import Path
from agentsecretstore import AgentVault

async def example():
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        # Bulk-import KEY=VALUE pairs from a .env-formatted string.
        # Each line becomes a secret; blank lines and # comments are ignored.
        content = Path(".env.production").read_text()

        result = await vault.import_env(
            content,
            "production/gemini",
            secret_type="api_key",
            access_tier="sensitive",
            metadata={"source": ".env.production"},
        )

        print(result.imported_count)  # int: 4
        print(result.imported)        # ["GEMINI_API_KEY", "DATABASE_URL", ...]
        print(result.namespace)       # "production/gemini"

asyncio.run(example())

# Sync equivalent: AgentVaultSync(...).import_env(content, "production/gemini")

import asyncio
import os
from agentsecretstore import AgentVault

async def example():
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        # Standard scoped token
        token = await vault.request_token(
            scopes=["secrets:read:production/gemini/*"],
            ttl="1h",
        )

        print(token.token)       # JWT bearer token
        print(token.expires_at)  # ISO timestamp

asyncio.run(example())

import asyncio
import os
from agentsecretstore import AgentVault, VaultError

async def example():
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        try:
            await vault.delete_secret("production/gemini/GEMINI_API_KEY")
        except VaultError as exc:
            assert exc.code == "HUMAN_AUTH_REQUIRED"

asyncio.run(example())

import asyncio
import os
from agentsecretstore import (
    AgentVault,
    ApprovalRequiredError,
    AuthError,
    NetworkError,
    RateLimitError,
    ScopeError,
    SecretNotFoundError,
    ValidationError,
    VaultError,
)

async def safe_get_secret(path: str) -> str | None:
    async with AgentVault(agent_key=os.environ["ASS_AGENT_KEY"]) as vault:
        try:
            return await vault.get_secret(path)

        except AuthError as e:
            print(f"Auth failed: {e}")
            return None

        except ScopeError as e:
            print(f"Permission denied: {e}")
            return None

        except SecretNotFoundError as e:
            print(f"Secret not found: {e}")
            return None

        except ApprovalRequiredError as e:
            print(f"Approval pending: {e.approval_id}")
            return None

        except RateLimitError as e:
            print(f"Rate limited. Retry after {e.retry_after}s")
            return None

        except ValidationError as e:
            print(f"Invalid request: {e}")
            return None

        except NetworkError as e:
            print(f"Network error: {e}")
            return None

        except VaultError as e:
            print(f"Vault error {e.code}: {e}")
            return None

from pydantic import BaseModel
from typing import Literal

SecretType = Literal["api_key", "oauth_token", "db_credential", "ssh_key", "certificate", "custom"]
AccessTier = Literal["standard", "sensitive", "critical"]

class TokenResponse(BaseModel):
    token: str | None = None
    token_id: str | None = None
    expires_at: str | None = None
    requires_approval: bool = False
    approval_id: str | None = None

class SecretResponse(BaseModel):
    value: str
    secret_type: SecretType
    access_tier: AccessTier
    metadata: dict
    version: int

class SecretListItem(BaseModel):
    namespace: str
    key: str
    version: int
    secret_type: SecretType
    access_tier: AccessTier
    metadata: dict
    created_at: str
    updated_at: str
    expired: bool = False

class SecretListResponse(BaseModel):
    namespace: str
    items: list[SecretListItem]
    total: int = 0

# Pattern: LangChain / CrewAI agent that fetches its own Gemini credentials
# pip install google-genai

import asyncio
import os
from agentsecretstore import AgentVault
from google import genai

async def run_inference_agent(task: str, agent_key: str) -> str:
    """
    Agent receives its Agent Secret Store key from the runtime.
    The SDK exchanges it for short-lived scoped tokens automatically.
    """
    async with AgentVault(agent_key=agent_key) as vault:
        api_key = await vault.get_secret("production/gemini/GEMINI_API_KEY")

    # Use the secret and let it fall out of scope
    client = genai.Client(api_key=api_key)
    response = await client.aio.models.generate_content(
        model=os.environ.get("GEMINI_MODEL", "gemini-2.5-flash"),
        contents=task,
    )
    return response.text or ""