Sovaria Python SDK

Async Python SDK for the Sovaria gateway.

Install

pip install sovaria-sdk

Local workspace development:

pip install -e apps/sdk/python

Supabase-Style Bootstrap (Portal Config)

from sovaria import create_client

client = create_client(
    {
        "url": "https://gateway.sovaria.in",
        "org_id": "11111111-1111-1111-1111-111111111111",
        "project_id": "22222222-2222-2222-2222-222222222222",
        "env_id": "33333333-3333-3333-3333-333333333333",
        # optional: portal-issued publishable/anon key
        "publishable_key": "<token>",
    }
)

Or initialize directly from environment variables:

from sovaria import SovariaClient

client = SovariaClient.from_env()

Supported environment variables:

• SOVARIA_URL
• SOVARIA_ORG_ID
• SOVARIA_PROJECT_ID
• SOVARIA_ENV_ID
• SOVARIA_ACCESS_TOKEN or SOVARIA_PUBLISHABLE_KEY or SOVARIA_ANON_KEY

Quick Start

import asyncio
from sovaria import SovariaClient


async def main() -> None:
    async with SovariaClient.from_env() as sov:
        await sov.console_login("you@example.com", "password")
        me = await sov.get_me()

        # If your token includes scope claims, get_me() auto-hydrates scope.
        # Otherwise, set scope explicitly for SQL/realtime routes:
        # sov.set_scope(org_id="...", project_id="...", env_id="...")

        health = await sov.get_engines_health()
        print(me)
        print(health)

        # Realtime usage
        await sov.stream_subscribe("db.table.users", "py-client-1")
        presence = await sov.stream_presence("db.table.users")
        print(presence)


asyncio.run(main())

Sync Quick Start

from sovaria import SovariaSyncClient

with SovariaSyncClient.from_env() as sov:
    sov.console_login("you@example.com", "password")
    me = sov.get_me()
    print(me)

Use SovariaSyncClient for WSGI workers, cron jobs, and scripts where an explicit async event loop would add unnecessary complexity. SovariaClient.run_sync() is deprecated and raises if a loop is already running.

Auth lanes and session KEK

Passwordless flows cannot populate the Argon2id KEK. Call E2EE helpers with password= or follow up with a password login.

logout() sends Authorization: Bearer plus {"refresh_token": ...} and always clears local tokens/KEK, including when the HTTP call fails.

Refresh has two supported modes. Do not mix them:

• CLI / SDK: JSON body {"refresh_token": "..."} (token stored in keyring or memory).
• Console / browser: empty JSON body plus the HttpOnly refresh cookie (withCredentials). Browser embeds should not copy the CLI body pattern; CLI/SDK clients should not expect cookies.

Admin API lanes (TypeScript parity)

Portal bootstrap: pass expires_in when the token is a user JWT. Omitting it selects static-key mode (long local TTL, no refresh). Do not put a user JWT in publishable_key.

Coverage

• Axis: auth, context, graph
• Atlas-Doc: put/get/delete/scan plain docs
• Atlas-Rel: SQL, table lifecycle, migrations
• Pulse: stream, e2ee key lifecycle, websocket URL helper
• Guard: templates, CRUD, evaluate, rollback, history
• Echo: audit + simulation
• Chrono: commit/timelines/snapshots/converge/variance
• Gateway: engine health

Project Scaffold from sovaria.config.json

You can scaffold a new client project directly from the SDK package.

from sovaria import (
    DEFAULT_CONFIG_FILENAME,
    create_project_from_config_file,
    write_project_config_template,
)

# 1) Write template config JSON
write_project_config_template(DEFAULT_CONFIG_FILENAME, name="my-app")

# 2) Generate project folder structure from config
result = create_project_from_config_file(DEFAULT_CONFIG_FILENAME, target_dir=".")
print(result.project_root)

Generated files include:

• <project>/sovaria.config.json
• <project>/.env.example
• <project>/sovaria/functions/
• <project>/sovaria/migrations/
• <project>/sovaria/tests/

Raw API Escape Hatch

You can call any gateway endpoint directly:

await sov.request("GET", "/api/v1/system/engines/health", requires_auth=False)

Functions + Storage Pass-Through

# Forge deployment status
status = await sov.fn_status("deploy-123")

# Generic Vault path operations
await sov.store_put_json("kv/app/settings", {"theme": "light"})
settings = await sov.store_get("kv/app/settings")
await sov.store_delete("kv/app/settings")

Sovariq tools (MCP)

sovaria.sovariq is the tool catalog behind Sovariq, Sovaria's AI engineer, and exposes your project to any AI assistant through the Model Context Protocol. Tools act with your permissions: every call goes through the Sovaria gateway, so Guard policies, plan limits and the audit log apply exactly as they do in the console.

pip install 'sovaria-sdk[mcp]'      # or: pip install 'sovaria-cli[mcp]'
sov login && sov scope use           # pick org + project
sov mcp config                       # prints a client config entry ("sovariq")

Add the printed entry to Claude Code (.mcp.json), Claude Desktop or Cursor; it runs sov mcp serve over stdio. Without the CLI, run sovariq-mcp (alias sovaria-mcp) with SOVARIA_API_URL, SOVARIA_ACCESS_TOKEN (or SOVARIA_API_KEY), SOVARIA_ORG_ID and SOVARIA_PROJECT_ID set.

Every tool declares a risk level (read, write, destructive, execute), a category and a timeout (MCP _meta). Only read tools are marked read-only, so clients ask before anything else runs. Mutating tools check their own postconditions and return a verification block, and record a version in Chrono like the console does.

sovariq-mcp --transport http serves streamable HTTP for internal services (the Sovariq agent uses it). It holds no credentials and rejects requests without a bearer token; restrict accepted Host headers with SOVARIA_MCP_ALLOWED_HOSTS. sovaria.ai remains as an alias of sovaria.sovariq for older imports.