Python SDK

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

export ASS_AGENT_KEY="ass_live_your_key_here"

import asyncio
from agentsecretstore import AgentVault

async def main():
    # Reads ASS_AGENT_KEY from environment automatically
    vault = AgentVault()

    # Or provide explicitly
    vault = AgentVault(
        agent_key="ass_live_your_key_here",
        base_url="https://api.agentsecretstore.com/v1",  # default
        timeout=10.0,   # seconds, default: 10
        retries=3,      # default: 3
    )

    await vault.close()  # clean up connections

asyncio.run(main())

import asyncio
from agentsecretstore import AgentVault

async def main():
    # Recommended: use as async context manager
    async with AgentVault() as vault:
        secret = await vault.get_secret("production/openai/api-key")
        print(secret.value)
    # connections closed automatically

asyncio.run(main())

from agentsecretstore import SyncAgentVault

# For non-async code (e.g., scripts, Django views, legacy code)
vault = SyncAgentVault()

secret = vault.get_secret("production/openai/api-key")
print(secret.value)

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

from agentsecretstore import AgentVault
from agentsecretstore.models import Secret

async def example():
    async with AgentVault() as vault:
        # Basic retrieval
        secret: Secret = await vault.get_secret("production/openai/api-key")

        print(secret.value)        # str: "sk-proj-abc123..."
        print(secret.version)      # int: 1
        print(secret.path)         # str: "production/openai/api-key"
        print(secret.tier)         # Literal["standard","sensitive","critical"]
        print(secret.description)  # str | None
        print(secret.tags)         # dict[str, str]
        print(secret.created_at)   # datetime
        print(secret.updated_at)   # datetime
        print(secret.accessed_at)  # datetime
        print(secret.expires_at)   # datetime | None

        # Specific version
        secret_v1 = await vault.get_secret(
            "production/openai/api-key",
            version=1,
        )

        # Using a scoped token
        secret_with_token = await vault.get_secret(
            "production/openai/api-key",
            token="ast_tok_received_from_orchestrator",
        )

from agentsecretstore.models import SecretTier
from datetime import datetime, timezone, timedelta

async def example():
    async with AgentVault() as vault:
        # Full options
        await vault.set_secret(
            path="production/openai/api-key",
            value="sk-proj-abc123...",
            tier=SecretTier.SENSITIVE,       # or "sensitive"
            description="OpenAI production key for inference cluster",
            tags={"team": "ml", "env": "prod", "rotated_by": "ci"},
            expires_at=datetime.now(timezone.utc) + timedelta(days=365),
        )

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

from agentsecretstore.models import SecretMetadata

async def example():
    async with AgentVault() as vault:
        # List all secrets in a namespace prefix
        result = await vault.list_secrets(
            namespace="production/openai",
            limit=50,       # default: 50, max: 200
            cursor=None,    # for pagination
        )

        items: list[SecretMetadata] = result.items
        for item in items:
            print(f"{item.path}  v{item.version}  [{item.tier}]")

        # Paginate
        while result.next_cursor:
            result = await vault.list_secrets(
                "production/openai",
                cursor=result.next_cursor,
            )
            for item in result.items:
                print(item.path)

from agentsecretstore.models import ScopedToken

async def example():
    async with AgentVault() as vault:
        # Standard scoped token
        token: ScopedToken = await vault.request_token(
            scope="secrets:read:production/openai/*",
            ttl_seconds=3600,
            description="Inference agent token for batch run #4821",
            allowed_ips=["10.0.1.50"],   # Optional IP restriction
            max_uses=None,               # None = unlimited within TTL
        )

        print(token.value)               # "ast_tok_..."
        print(token.scope)               # "secrets:read:production/openai/*"
        print(token.expires_at)          # datetime
        print(token.approval_status)     # "approved" | "pending"

        # Single-use token
        one_time = await vault.request_token(
            scope="secrets:read:production/stripe/api-key",
            ttl_seconds=300,
            max_uses=1,
            description="Single payment processing run",
        )

async def example():
    async with AgentVault() as vault:
        result = await vault.rotate_secret(
            path="production/openai/api-key",
            new_value="sk-proj-rotated789...",
            grace_period_seconds=300,  # old version readable for 5 min
        )

        print(f"Rotated to version {result.new_version}")
        print(f"Old version {result.old_version} expires at {result.expires_at}")

import asyncio
from agentsecretstore import AgentVault
from agentsecretstore.exceptions import (
    AgentVaultError,
    AgentVaultAuthError,
    AgentVaultPermissionError,
    AgentVaultNotFoundError,
    AgentVaultRateLimitError,
    AgentVaultApprovalPendingError,
    AgentVaultNetworkError,
)

async def safe_get_secret(path: str) -> str | None:
    async with AgentVault() as vault:
        try:
            secret = await vault.get_secret(path)
            return secret.value

        except AgentVaultAuthError as e:
            # 401: Invalid or expired agent key / token
            print(f"Auth failed: {e}")
            return None

        except AgentVaultPermissionError as e:
            # 403: Token scope doesn't cover this path
            print(f"Permission denied: {e.path} — need {e.required_scope}")
            return None

        except AgentVaultNotFoundError as e:
            # 404: Secret doesn't exist
            print(f"Secret not found: {e.path}")
            return None

        except AgentVaultApprovalPendingError as e:
            # 202: Needs human approval
            print(f"Approval pending: {e.approval_request_id}")
            return None

        except AgentVaultRateLimitError as e:
            # 429: Too many requests
            print(f"Rate limited. Retry after {e.retry_after}s")
            return None

        except AgentVaultNetworkError as e:
            # Timeout or connectivity issue
            print(f"Network error: {e}")
            return None

from typing import Literal
from datetime import datetime
from dataclasses import dataclass

SecretTier = Literal["standard", "sensitive", "critical"]
ApprovalStatus = Literal["approved", "pending", "denied"]

@dataclass
class Secret:
    path: str
    value: str
    version: int
    tier: SecretTier
    description: str | None
    tags: dict[str, str]
    created_at: datetime
    updated_at: datetime
    accessed_at: datetime
    expires_at: datetime | None

@dataclass
class SecretMetadata:
    path: str
    version: int
    tier: SecretTier
    description: str | None
    tags: dict[str, str]
    created_at: datetime
    updated_at: datetime
    expires_at: datetime | None
    # Note: no 'value' field

@dataclass
class ScopedToken:
    value: str
    scope: str
    ttl_seconds: int
    expires_at: datetime
    description: str | None
    allowed_ips: list[str] | None
    max_uses: int | None
    uses_remaining: int | None
    approval_status: ApprovalStatus
    approval_request_id: str | None

# Pattern: LangChain / CrewAI agent that fetches its own credentials

import asyncio
from agentsecretstore import AgentVault
from openai import AsyncOpenAI

async def run_inference_agent(task: str, token: str) -> str:
    """
    Agent receives a scoped token from the orchestrator.
    Only reads the specific secrets its scope allows.
    """
    async with AgentVault() as vault:
        # Use scoped token — not the master key
        secret = await vault.get_secret(
            "production/openai/api-key",
            token=token,  # scoped: "secrets:read:production/openai/*"
        )

    # Use the secret and let it fall out of scope
    client = AsyncOpenAI(api_key=secret.value)
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": task}],
    )
    return response.choices[0].message.content or ""