safenode 0.1.0

Official Python SDK for SafeNode — zero-knowledge credential vault

SafeNode Python SDK

Official Python SDK for SafeNode — zero-knowledge credential vault.

Installation

pip install safenode

Quick Start

MCP Token (recommended)

from safenode import SafeNode

client = SafeNode(token="kvt_your_token_here")

Or use the environment variable:

export SAFENODE_TOKEN=kvt_your_token_here

import os
from safenode import SafeNode

client = SafeNode()  # reads SAFENODE_TOKEN automatically

Email + Password

from safenode import SafeNode

client = SafeNode(
    email="[e-mail]",
    password="your-password",
    passphrase="your-vault-passphrase",
    vault_id="your-vault-uuid",
)

Usage

Proxy Requests

Forward HTTP requests through a stored credential:

response = client.proxy.request(
    credential="my-api-key",
    method="GET",
    path="/users",
    query_params={"page": "1"},
)

print(response.status_code)   # 200
print(response.body)          # parsed JSON body
print(response.response_time_ms)

POST with body and headers:

response = client.proxy.request(
    credential="my-api-key",
    method="POST",
    path="/messages",
    body={"content": "Hello"},
    headers={"X-Custom-Header": "value"},
    timeout=60,
)

SSH Commands

Execute commands on remote servers:

result = client.ssh.exec(
    credential="my-server",
    command="df -h",
    timeout=30,
)

print(result.stdout)
print(result.stderr)
print(result.exit_code)
print(result.execution_time_ms)

Database Queries

Run SQL queries through stored database credentials:

result = client.db.query(
    credential="my-database",
    sql="SELECT * FROM users WHERE active = %s",
    params=[True],
    timeout=30,
)

print(result.columns)     # ["id", "name", "email"]
print(result.rows)        # [[1, "Alice", "[e-mail]"], ...]
print(result.row_count)   # 42

Credentials Management

List credentials:

credentials = client.credentials.list()
credentials = client.credentials.list(type="api_key", environment="production")
credentials = client.credentials.list(search="github")

Get a credential:

cred = client.credentials.get("credential-uuid")

Create a credential:

cred = client.credentials.create(
    name="my-new-api",
    type="api_key",
    data={"api_key": "secret-value"},
    environment="production",
    tags=["external", "payments"],
    base_url="https://api.example.com",
    description="Payment provider API key",
)

Update a credential:

cred = client.credentials.update(
    "credential-uuid",
    description="Updated description",
    tags=["external"],
)

Delete a credential:

client.credentials.delete("credential-uuid")

Error Handling

from safenode import (
    SafeNode,
    SafeNodeError,
    AuthError,
    VaultLockedError,
    RateLimitError,
    NotFoundError,
    ProxyError,
    TimeoutError,
)
import time

client = SafeNode(token="kvt_your_token")

try:
    response = client.proxy.request("my-api", "GET", "/data")
except VaultLockedError:
    print("Vault is locked — re-authentication required")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
    time.sleep(e.retry_after)
except NotFoundError:
    print("Credential not found")
except ProxyError as e:
    print(f"Upstream error: {e}")
except TimeoutError as e:
    print(f"Request timed out: {e}")
except AuthError as e:
    print(f"Authentication failed: {e}")
except SafeNodeError as e:
    print(f"SafeNode error {e.status_code}: {e}")

Context Manager

with SafeNode(token="kvt_your_token") as client:
    result = client.db.query("my-db", "SELECT 1")

Environment Variables

Variable	Description
SAFENODE_TOKEN	MCP token (kvt_xxx) for auth

Requirements

• Python 3.9+
• httpx >= 0.27