Platform
ScaiWave ScaiGrid ScaiCore ScaiBot ScaiDrive ScaiKey Models Tools & Services
Solutions
Organisations Developers Internet Service Providers Managed Service Providers AI-in-a-Box
Resources
Support Documentation Blog Downloads
Company
About Research Careers Investment Opportunities Contact
Log in

Python SDK

Async-first Python client with a sync wrapper. Typed models, automatic token refresh, retry with exponential backoff, connection pooling.

Install#

bash
1
pip install scaivault-sdk

Python 3.10+.

Authenticate#

python
1
2
3
4
5
6
7
import os
from scaivault_sdk import ScaiVaultClient

client = ScaiVaultClient(
    base_url="https://scaivault.scailabs.ai",
    token=os.environ["SCAIVAULT_TOKEN"],
)

For client-credentials refresh, pass client_id / client_secret instead of a raw token — the SDK handles the token exchange and automatically refreshes on expiry:

python
1
2
3
4
5
client = ScaiVaultClient(
    base_url="https://scaivault.scailabs.ai",
    client_id=os.environ["SCAIKEY_CLIENT_ID"],
    client_secret=os.environ["SCAIKEY_CLIENT_SECRET"],
)

Async vs sync#

ScaiVaultClient is async. Use it inside async def with await:

python
1
2
3
4
5
6
7
8
import asyncio

async def main():
    async with ScaiVaultClient(base_url=..., token=...) as client:
        secret = await client.secrets.read("app/db/password")
        print(secret.data["password"])

asyncio.run(main())

For sync code, use SyncScaiVaultClient:

python
1
2
3
4
5
from scaivault_sdk import SyncScaiVaultClient

client = SyncScaiVaultClient(base_url=..., token=...)
secret = client.secrets.read("app/db/password")
print(secret.data["password"])

Both expose the same methods, just with/without async.

Secrets#

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# Read
secret = await client.secrets.read("environments/production/salesforce/oauth")
secret.version           # int
secret.data              # dict[str, Any]
secret.metadata          # SecretMetadata
secret.secret_type       # SecretType enum

# Write
result = await client.secrets.write(
    path="environments/production/salesforce/oauth",
    data={"client_id": "...", "client_secret": "..."},
    secret_type="json",
    metadata={"tags": ["salesforce"]},
    max_versions=10,
)

# Update (metadata only)
await client.secrets.update_metadata(
    path="app/db/password",
    metadata={"tags": ["critical"]},
)

# Delete
await client.secrets.delete("old/secret")
await client.secrets.delete("old/secret", permanent=True)

# List
listing = await client.secrets.list(prefix="environments/production/", limit=50)
for item in listing.data:
    print(item.path, item.version)

# Paginate
while listing.has_more:
    listing = await client.secrets.list(prefix=..., cursor=listing.cursor)

# Read specific version
old = await client.secrets.read("app/db/password", version=1)

# Rotate
rotated = await client.secrets.rotate(
    "app/db/password",
    reason="compromise",
    new_value={"password": "new-value"},
    grace_period="1h",
)

# Batch
results = await client.secrets.batch_read([
    "integrations/salesforce/oauth",
    "integrations/stripe/api-key",
])
# results is a BatchResult with .secrets (dict) and .errors (dict)

Policies#

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from scaivault_sdk.models import PolicyRule

# Create
policy = await client.policies.create(
    name="production-read-only",
    rules=[
        PolicyRule(
            path_pattern="environments/production/**",
            permissions=["read", "list"],
            conditions={"ip_ranges": ["10.0.0.0/8"], "require_mfa": True},
        ),
    ],
    description="Developers read production from VPN + MFA",
)

# Bind
await client.policies.bind(
    policy_id=policy.id,
    identity_type="group",
    identity_id="group:developers",
)

# Test
result = await client.policies.test(
    identity_id="user:alice@acme.example",
    path="environments/production/salesforce/oauth",
    permission="read",
    context={"source_ip": "10.0.1.50"},
)
print(result.allowed)  # bool

Rotation#

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# Create a rotation policy
policy = await client.rotation.create(
    name="quarterly",
    interval="90d",
    grace_period="48h",
    warn_before="7d,1d",
    auto_generate=False,
)

# Attach a secret
await client.rotation.assign_secret(
    policy_id=policy.id,
    secret_path="environments/production/salesforce/oauth",
)

# Trigger now
await client.rotation.trigger_rotation(
    policy_id=policy.id,
    secret_paths=["environments/production/salesforce/oauth"],
)

# History
history = await client.rotation.get_history(policy.id, limit=100)
for item in history.data:
    print(item.secret_path, item.status, item.rotated_at)

# Due for rotation
due = await client.rotation.get_secrets_due(within_hours=168)

PKI#

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from scaivault_sdk.models import CSRSubject

# Create CA
ca = await client.pki.create_ca(
    name="acme-root",
    common_name="Acme Root CA",
    ca_type="root",
    key_type="ec",
    key_size=256,
    validity_days=3650,
)

# Issue cert against a role
cert = await client.pki.issue(
    role="svc-mtls",
    common_name="billing.svc.cluster.local",
    alt_names=["billing-api.svc.cluster.local"],
    ttl="168h",
)
# cert.certificate_pem, cert.private_key_pem, cert.ca_chain

# Generate CSR in-vault
csr = await client.pki.generate_csr(
    subject=CSRSubject(common_name="vendor.example"),
    san_dns=["vendor-api.example"],
    key_type="ec",
    key_size=256,
)

# Import an external CSR and approve it
imported = await client.pki.import_csr(csr_pem="-----BEGIN...")
await client.pki.approve_csr(imported.id)
signed = await client.pki.sign_csr_by_id(imported.id, ca_id=ca.id, validity_days=90)

# Validate
result = await client.pki.validate_certificate(
    certificate_pem="-----BEGIN...",
    chain_pem="-----BEGIN...",
    check_revocation=True,
)
print(result.valid, result.errors)

ACME#

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# Register account
account = await client.acme.create_account(
    name="letsencrypt-production",
    provider="letsencrypt",
    environment="production",
    email="certs@acme.example",
)

# Issue
order = await client.acme.issue(
    account_id=account.id,
    domains=["api.acme.example"],
    challenge_type="dns-01",
    auto_renew=True,
)
# order.status starts as "pending"; poll until "valid"

Dynamic secrets#

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
# Generate credentials
lease = await client.dynamic.generate_credentials(
    engine="support-db",
    role="readonly",
    ttl="2h",
)
connection_url = lease.data["connection_url"]

try:
    # Use the credentials
    ...
finally:
    await client.dynamic.revoke_lease(lease.lease_id)

Or as a context manager:

python
1
2
3
4
async with client.dynamic.credentials("support-db", "readonly", ttl="2h") as creds:
    # creds is the lease data
    run_query(creds["connection_url"])
# Lease revoked automatically on exit

Error handling#

python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from scaivault_sdk.errors import (
    ScaiVaultError,
    AuthenticationError,
    AccessDeniedError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)

try:
    secret = await client.secrets.read("app/db/password")
except NotFoundError:
    # Path doesn't exist
    ...
except AccessDeniedError as e:
    # No policy allows this
    print(e.details)
except RateLimitError as e:
    # Caller is rate-limited
    await asyncio.sleep(e.retry_after)
except ScaiVaultError as e:
    # Catch-all
    print(e.code, e.message, e.request_id)

All exceptions subclass ScaiVaultError. code, message, details, request_id, and status_code are available on every one.

Retries and timeouts#

The SDK retries transient failures (rate_limited, service_unavailable, internal_error, connection errors) with exponential backoff. Configure:

python
1
2
3
4
5
6
7
client = ScaiVaultClient(
    base_url=...,
    token=...,
    timeout=10.0,         # per-request, seconds
    max_retries=5,        # default 3
    retry_backoff=1.5,    # multiplier
)

Non-retryable errors (auth, access denied, validation) raise immediately.

Configuration via environment#

The SDK reads these env vars as defaults if arguments aren't passed:

Variable Default for
SCAIVAULT_BASE_URL base_url
SCAIVAULT_TOKEN token
SCAIKEY_CLIENT_ID client_id
SCAIKEY_CLIENT_SECRET client_secret

So ScaiVaultClient() with no arguments picks up env in the obvious way.

What's next#

Updated 2026-05-17 13:26:51 View source (.md) rev 2