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://api.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/

Ergonomic clients: data, storage, functions, realtime

Dot-access facades over the gateway, matching the Dart/TS clients. Each wraps a SovariaClient; the same server-side checks (Guard rules, plan limits, tenant isolation, audit) apply as to hand-written requests.

from sovaria import (
    SovariaClient, DataClient, StorageClient, FunctionsClient, RealtimeClient,
)

async with SovariaClient("https://<ref>.sovaria.in") as sov:
    sov.set_access_token("<anon-or-user-token>")
    sov.set_scope(org_id="...", project_id="...", env_id="...")

    # Database (SQL + documents)
    data = DataClient(sov)
    rows = await data.table("todos").query("SELECT * FROM todos WHERE done = $1", [False])
    await data.table("todos").insert({"title": "ship it"})
    count = await data.rpc("post_count", {"u": user_id})           # POST /rest/v1/rpc/post_count

    # Storage (Vault object buckets)
    storage = StorageClient(sov)
    await storage.bucket("avatars").upload("me.png", png_bytes, content_type="image/png", upsert=True)
    blob = await storage.bucket("avatars").download("me.png")          # -> bytes
    url  = await storage.bucket("avatars").create_signed_url("me.png", expires_in=3600)
    pub  = storage.bucket("avatars").public_url("me.png")              # public buckets only

    # Functions (deployed Forge functions)
    fns = FunctionsClient(sov)
    result = await fns.invoke("send-welcome-email", {"to": "alice@example.com"})

    # Realtime (Pulse channels)
    rt = RealtimeClient(sov)
    room = rt.channel("room:42")
    await room.publish({"text": "hello"})
    who = await room.presence()
    async for message in room.listen():          # needs pip install 'sovaria-sdk[realtime]'
        print(message)
        if message.get("text") == "bye":
            break                                 # leaving the loop closes the socket

listen() reconnects with backoff and stops cleanly when you are signed out or not allowed on the channel. After a reconnect it first yields what was published while it was offline (from the channel's replay history, in order, once each), then live messages; on_gap= is called for messages that can't be recovered, and catch_up=False turns it off. Without the realtime extra, use await room.ws_url() to open the socket with your own WebSocket client, or poll await room.replay().

AI: knowledge bases (RAG) and agents

from sovaria import AgentsClient, KnowledgeClient

kb = KnowledgeClient(sov)
await kb.add_file("support_docs", "handbook.pdf")           # PDF, DOCX, HTML, Markdown, TXT, CSV, JSON
hits = await kb.search("support_docs", "refund window")      # results + citations, only what the caller may read

agents = AgentsClient(sov)
reply = await agents.run("ag_123", "Where is my order?")
reply = await agents.run("ag_123", "And the invoice?", thread_id=reply.thread_id)
async for event in agents.stream("ag_123", "Summarise my plan"):
    ...  # conversation, text_delta, tool_call, text, done

The embedding model (your own key or Sovaria's) and usage in tokens and dollars: sov.api.embeddings.get_settings(), set_settings(...), usage(months=3); knowledge bases are also under sov.api.knowledge.*.

Raw API Escape Hatch

You can call any gateway endpoint directly:

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

For the Forge-native call style (execution profiles, async/heavy queueing, simulation) and generic Vault KV paths, the lower-level methods remain:

# Forge deployment status + native invoke with an execution profile
status = await sov.fn_status("deploy-123")
await sov.fn_invoke("heavy-report", {"month": "2026-10"}, execution_profile="heavy")

# Generic Vault path operations (JSON KV, not object storage)
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.

Hosted (nothing to install): add https://api.sovaria.in/mcp as a remote MCP server (a "custom connector") in Claude, ChatGPT, Cursor or VS Code. The assistant signs you in with OAuth; in the console you pick one organization (optionally one project) and read-only or read and write. Add ?project=<project id> to the URL to pick a project for an organization-wide connection, ?read_only=true to see only read tools, or ?features=database,storage to load fewer tools. Disconnect any time in Account › Security. Details: https://sovaria.in/docs/mcp.

Local (stdio, with the CLI login):

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. It holds no credentials: each request's bearer token (and org/project headers) is forwarded to the gateway (--gateway-url or SOVARIA_GATEWAY_URL, default https://api.sovaria.in), and requests without one are rejected. Restrict accepted Host headers with SOVARIA_MCP_ALLOWED_HOSTS.

Manage connections from code: client.api.mcp.list_connections(), disconnect(grant_id), org_settings(), set_org_allowed(allowed) and org_disconnect(grant_id) (TypeScript: sov.api.mcp.listConnections() …). sovaria.ai remains as an alias of sovaria.sovariq for older imports.