ault-sdk 0.0.1a2

Python SDK for the Ault blockchain — high-level transaction methods and REST/WS clients.

Ault SDK for Python

A Python SDK for the Ault blockchain — high-level transaction methods and REST/WebSocket query clients for the current Ault modules (License, Miner, Market, CLOB, Agent, Authority, RateLimit, Reward, TokenRegistry, Staking, Bank).

Features

• High-level async client with simple transaction methods (no manual message building).
• Flexible signer support: private key, EIP-1193 providers, or any object implementing the TypedDataSigner protocol.
• Automatic address format conversion (0x ⇄ ault1).
• DEX API client for market data, orders, trades, and UDF charting.
• DEX WebSocket client for real-time streaming with auto-reconnect and topic subscriptions.
• REST query clients for every Ault module.
• EIP-712 typed-data signing, pinned by a byte-exact signature regression test against a known-good fixture.
• Agent (SIGN_MODE_DIRECT) transactions for bot/delegated signing.
• EVM client for ERC-20, native AULT, and the bridge precompile on top of web3.py.
• Automatic retry with exponential backoff on REST calls.

Installation

PyPI publishing is wired (Trusted Publishing on tag push, see .github/workflows/release.yml) but no version has been tagged yet. For now, install from source with uv:

# From the repo root
uv sync

Requirements

• Python >=3.12.
• uv for dependency management.
• just + ty for the dev workflow (optional, but just check drives lint/format/typecheck/test).

Quick Start

import asyncio
from ault_sdk import create_client, get_network_config, PrivateKeySigner


async def main() -> None:
    client = await create_client(
        ClientOptions(
            network=get_network_config("ault_10904-1"),
            signer=PrivateKeySigner("0x..."),
        )
    )

    # Query data.
    licenses = await client.license.get_owned_by(client.address)
    epoch = await client.miner.get_current_epoch()

    # Execute a transaction (no manual message building required).
    result = await client.miner_tx.delegate_mining(
        license_ids=[1, 2, 3],   # accepts ints, strs, or decimal strings
        operator="0xOperator...",  # accepts 0x or ault1 format
    )

    print(
        f"TX Hash: {result.tx_hash}, Confirmed: {result.confirmed}, Success: {result.success}"
    )


asyncio.run(main())

Note: client.<module> returns the REST query object; transaction methods live under client.<module>_tx. The split keeps REST queries and tx methods in separate namespaces so they can't silently shadow each other.

Creating a Client

create_client() accepts several signer input shapes and resolves the signer's bech32 address from the on-chain public key.

With a private key

from ault_sdk import create_client, ClientOptions, PrivateKeySigner, get_network_config

client = await create_client(
    ClientOptions(
        network=get_network_config("ault_10904-1"),
        signer=PrivateKeySigner("0x..."),
    )
)

With an EIP-1193 provider

Adapter for any provider-style request({method, params}) callable — hosted signing services, browser-wallet bridges, custom RPC signers.

from ault_sdk import Eip1193Signer, create_client, ClientOptions, get_network_config


async def provider_request(args: dict) -> str:
    # Your provider transport — e.g. forward to an HSM or Trezor bridge.
    ...


client = await create_client(
    ClientOptions(
        network=get_network_config("ault_10904-1"),
        signer=Eip1193Signer(provider_request, "0xYourEvmAddress"),
    )
)

With a custom TypedDataSigner

Any object with an async sign_typed_data(typed_data) -> str method satisfies the protocol.

from ault_sdk import TypedDataSigner, Eip712TypedData


class MySigner:  # structurally satisfies TypedDataSigner
    address: str

    def __init__(self, address: str) -> None:
        self.address = address

    async def sign_typed_data(self, typed_data: Eip712TypedData) -> str:
        # Hand off to your signing backend; return a 0x-prefixed 65-byte hex signature.
        ...


client = await create_client(
    ClientOptions(
        network=get_network_config("ault_10904-1"),
        signer=MySigner("0xYourEvmAddress"),
    )
)

Client Options

from dataclasses import dataclass
from ault_sdk import NetworkConfig, FetchOptions
from ault_sdk.eip712.signers import SignerInput

@dataclass
class ClientOptions:
    network: NetworkConfig                    # required
    signer: SignerInput                       # required
    signer_address: str | None = None          # optional override
    client: httpx.AsyncClient | None = None    # optional shared client
    fetch_options: FetchOptions | None = None  # retries / timeout / backoff
    default_gas_limit: str | None = None
    default_memo: str = ""
    default_confirmation_timeout_ms: int | None = None

Query Methods

Every REST module is available as a namespace on the client. Methods are async and return Python dict objects with the chain's snake_case JSON keys.

# License queries
license_info = await client.license.get_license("1")
licenses = await client.license.get_owned_by(client.address)
balance = await client.license.get_balance(client.address)

# Miner queries
epoch = await client.miner.get_current_epoch()
operators = await client.miner.get_operators()
delegation = await client.miner.get_license_delegation("123")
emission = await client.miner.get_emission_info()

# Market + CLOB queries
markets = await client.market.get_markets()
market = await client.market.get_market(1)
orders = await client.clob.get_orders(market_id=1)

# Reward queries
reward_params = await client.reward.get_params()
referral_code = await client.reward.get_referral_code(client.address)
builder_allowances = await client.reward.get_builder_allowances(user=client.address)

# TokenRegistry queries
pair = await client.tokenregistry.get_token_pair("0xAb5801a7D398351b8bE11C439e05C5B3259aeC9B")
pairs = await client.tokenregistry.get_token_pairs()

# Bank (cosmos SDK)
balance = await client.bank.get_balance(client.address, "aault")
balances = await client.bank.get_all_balances(client.address)

DEX API

The SDK includes a client for the Ault DEX API — market data, order books, trades, balances, UDF charting.

from ault_sdk import create_ault_client, get_network_config
import dataclasses

network = get_network_config("ault_10904-1")
network = dataclasses.replace(network, dex_api_url="https://test-dex-api.cloud.aultblockchain.xyz")

client = create_ault_client(network=network)
dex = client.rest.dex

# Markets and currencies
markets = await dex.list_markets()
currencies = await dex.list_currencies()

# Ticker + order book + trades
ticker = await dex.get_ticker("AULT/USDT")
orderbook = await dex.get_orderbook("AULT/USDT", limit=50)
trades = await dex.list_trades("AULT/USDT", limit=100)

# OHLCV
candles = await dex.get_ohlcv("AULT/USDT", timeframe="1h", limit=100)

# User data (by address)
orders = await dex.list_orders(address="ault1...", status="open", symbol="AULT/USDT")
my_trades = await dex.list_my_trades(address="ault1...", symbol="AULT/USDT")
balances = await dex.get_balance(address="ault1...")

# TradingView UDF
config = await dex.get_udf_config()
history = await dex.get_udf_history(symbol="AULT/USDT", resolution="60", from_=1773500000, to=1773600000)

DEX WebSocket

Real-time streaming. Auto-reconnects with exponential backoff; re-subscribes all active topics after reconnect.

from ault_sdk import create_dex_ws, SubscriptionHandlers

ws = create_dex_ws("wss://test-dex-api.cloud.aultblockchain.xyz/ws")
await ws.connect()

def on_orderbook(data, msg):
    print(data["bids"][0], data["asks"][0])

sub = ws.subscribe(
    "orderbook:AULT/USDT",
    SubscriptionHandlers(on_data=on_orderbook),
)

# Lifecycle listeners
ws.on("open", lambda: print("connected"))
ws.on("close", lambda code, reason: print("closed", code, reason))
ws.on("reconnecting", lambda attempt: print("reconnecting", attempt))

# TTL: auto-expire subscription after 300s
ws.subscribe(
    "trades:AULT/USDT",
    SubscriptionHandlers(
        on_data=lambda data, msg: print(data),
        on_subscribed=lambda: print("subscribed!"),
        on_expired=lambda: print("expired"),
    ),
    ttl=300,
)

# Unsubscribe
sub.unsubscribe()

# Teardown
await ws.disconnect()

Address topics (orders:ault1..., balances:ault1..., mytrades:ault1...) are read-only projections of public on-chain state — no authentication needed.

Transaction Methods

All transaction methods are async and return a TxResult:

@dataclass(slots=True)
class TxResult:
    tx_hash: str            # transaction hash
    code: int               # result code (0 = success)
    success: bool           # true only when DeliverTx success was confirmed
    raw_log: str | None     # raw log
    confirmed: bool | None  # true = definitive, false = timed out, None = lookup unavailable

License Transactions

# Mint a license
await client.license_tx.mint_license(
    to="0x...",                          # EVM or ault1 address
    uri="https://example.com/metadata.json",
    reason="Minted via SDK",
)

# Batch mint
await client.license_tx.batch_mint_license(
    recipients=[
        {"to": "0xAddr1...", "uri": "https://example.com/1.json"},
        {"to": "0xAddr2...", "uri": "https://example.com/2.json"},
    ],
)

# Transfer / burn / revoke
await client.license_tx.transfer_license(license_id=123, to="0x...")
await client.license_tx.burn_license(license_id=123)
await client.license_tx.revoke_license(license_id=123)

# KYC management
await client.license_tx.approve_member(member="0x...")
await client.license_tx.revoke_member(member="0x...")
await client.license_tx.batch_approve_member(members=["0x...", "0x..."])

# Admin
await client.license_tx.set_minters(add=["0x..."], remove=[])
await client.license_tx.set_kyc_approvers(add=["0x..."], remove=[])

Miner Transactions

from ault_sdk import base64_to_bytes

# Delegate licenses to an operator
await client.miner_tx.delegate_mining(
    license_ids=[1, 2, 3],
    operator="0xOperator...",
)

# Cancel / redelegate
await client.miner_tx.cancel_mining_delegation(license_ids=[1, 2, 3])
await client.miner_tx.redelegate_mining(license_ids=[1, 2, 3], new_operator="0xNew...")

# VRF key + work submission (bytes accepted as raw or base64 string)
await client.miner_tx.set_owner_vrf_key(
    vrf_pubkey=base64_to_bytes("..."),
    possession_proof=base64_to_bytes("..."),
    nonce=1,
)

await client.miner_tx.submit_work(
    license_id=123,
    epoch=456,
    y=base64_to_bytes("..."),
    proof=base64_to_bytes("..."),
)

await client.miner_tx.batch_submit_work(
    submissions=[
        {"licenseId": 1, "epoch": 100, "y": base64_to_bytes("..."), "proof": base64_to_bytes("...")},
        {"licenseId": 2, "epoch": 100, "y": base64_to_bytes("..."), "proof": base64_to_bytes("...")},
    ],
)

# Operator management
await client.miner_tx.register_operator(commission_rate=5, commission_recipient="0x...")
await client.miner_tx.unregister_operator()
await client.miner_tx.update_operator_info(new_commission_rate=6)

Market and CLOB Transactions

# Create a spot market (marketType=1 = SPOT)
await client.market_tx.create_market(
    market_type=1,
    base_denom="aault",
    quote_denom="ausdc",
    tick_precision=6,
)

# Limit order
await client.clob_tx.place_limit_order(
    marketId=bytes([1]),  # 16-byte market id
    isBuy=True,
    price="1.5",
    quantity="100",
    lifespan={"seconds": 3600, "nanos": 0},
)

# Market order
await client.clob_tx.place_market_order(
    marketId=bytes([1]),
    isBuy=True,
    deposit="100",
)

# Cancel
await client.clob_tx.cancel_orders(order_ids=[b"\x00" * 16])

Reward and TokenRegistry Transactions

# Referral codes
await client.reward_tx.set_referral_code(code="RW18A001")
await client.reward_tx.approve_builder(
    builder="0x0000000000000000000000000000000000000001",
    max_fee_rate="0.005",
)

# Token pairs
await client.tokenregistry_tx.register_token_pair(
    erc20_address="0xAb5801a7D398351b8bE11C439e05C5B3259aeC9B",
    enabled=True,
)
await client.tokenregistry_tx.disable_token_registry_transfers()
await client.tokenregistry_tx.enable_token_registry_transfers()

Staking

# Delegate / undelegate / redelegate
await client.staking_tx.delegate(
    validator_address="aultvaloper1...",
    amount={"denom": "aault", "amount": "1000000"},
)
await client.staking_tx.undelegate(
    validator_address="aultvaloper1...",
    amount={"denom": "aault", "amount": "500000"},
)
await client.staking_tx.withdraw_rewards(validator_addresses=["aultvaloper1..."])

Transaction Options

Every tx method accepts optional gas_limit, fee_amount, memo, and confirmation_timeout_ms:

await client.miner_tx.delegate_mining(
    license_ids=[1, 2, 3],
    operator="0x...",
    gas_limit="300000",
    memo="Delegating via SDK",
)

EVM Client

Every Client has an evm attribute — an async EVM client on top of web3.py.

from ault_sdk import EvmClient, create_evm_client, CreateEvmClientOptions

# Construct standalone
evm = create_evm_client(CreateEvmClientOptions(
    network=get_network_config("ault_10904-1"),
    account=PrivateKeySigner("0x..."),  # optional — reader-only if omitted
))

# ERC-20 reads
meta = await evm.erc20.metadata("0xTokenAddress...")
balance = await evm.erc20.balance_of("0xToken...", "0xOwner...")

# ERC-20 writes (requires signer)
tx_hash = await evm.erc20.transfer(
    Erc20TransferParams(token="0xToken...", to="0xRecipient...", amount="10.5")
)

# Native AULT transfer
tx_hash = await evm.transfer_native(to="0xRecipient...", amount="1.0")

# TokenRegistry bridge (ERC-20 ⇄ bank)
await evm.bridge.to_core(BridgeParams(token="0xToken...", amount="10.0"))
await evm.bridge.to_evm(BridgeParams(token="0xToken...", amount="10.0"))

# Resolve token by symbol
address = await evm.tokens.resolve_address("USDT")
tokens = await evm.tokens.list_tokens()

Parallel Query Helpers

For license-heavy addresses (hundreds or thousands of licenses), the parallel helpers batch lookups:

analysis = await client.parallel.analyze_licenses("ault1...")
print(f"Total: {analysis['total']}")
print(f"Active: {analysis['active']}")
print(f"Delegated: {analysis['delegated']}")

# Or individual parallel helpers
ids = await client.parallel.get_all_license_ids("ault1...")
details = await client.parallel.get_license_details_parallel(ids)
delegations = await client.parallel.get_license_delegations_parallel(ids)

Error Handling

from ault_sdk import ApiError, NetworkError, TimeoutError

try:
    result = await client.license_tx.mint_license(to="0x...", uri="...")
    if result.success:
        print(f"Transaction committed: {result.tx_hash}")
    elif result.confirmed is False:
        print(f"Transaction broadcast but not confirmed yet: {result.tx_hash}")
    elif result.confirmed is None:
        print(
            f"Transaction accepted by CheckTx, but this node can't confirm it: "
            f"{result.tx_hash}"
        )
    else:
        print(f"Transaction failed: {result.raw_log}")
except ApiError as e:
    print(f"API Error (status={e.status}): {e}")
except NetworkError as e:
    print(f"Network Error: {e}")
except TimeoutError:
    print("Request timed out")

Advanced Usage

Low-level access

createClient returns a high-level Client; use createAultClient directly for a lower-level composite without tx method bundles.

from ault_sdk import create_ault_client, msg, sign_and_broadcast_eip712, SignAndBroadcastParams

client = create_ault_client(network=get_network_config("ault_10904-1"))

# Call the REST modules directly.
licenses = await client.rest.license.get_licenses()

# Build + broadcast a message manually.
delegate_msg = msg.miner.delegate_mining({
    "owner": "ault1...",
    "licenseIds": [1, 2, 3],
    "operator": "ault1...",
})

result = await sign_and_broadcast_eip712(SignAndBroadcastParams(
    network=client.network,
    signer=PrivateKeySigner("0x..."),
    signer_address="ault1...",
    msgs=[delegate_msg],
))

Network Configuration

from ault_sdk import NETWORKS, get_network_config, NetworkConfig

# Predefined networks
testnet = get_network_config("ault_10904-1")
localnet = get_network_config("ault_30904-1")

# Or construct a custom one
custom = NetworkConfig(
    name="My Network",
    type="testnet",
    chain_id="ault_10904-1",
    evm_chain_id=10904,
    rpc_url="https://my-rpc.example.com",
    rest_url="https://my-rest.example.com",
    evm_rpc_url="https://my-evm.example.com",
    dex_api_url="https://my-dex-api.example.com",
    is_production=False,
)

Utility Functions

from ault_sdk import (
    evm_to_ault, ault_to_evm,
    is_valid_ault_address, is_valid_evm_address,
    parse_evm_chain_id_from_cosmos_chain_id,
)

# Address conversion
ault_addr = evm_to_ault("0x1234...abcd")  # → "ault1..."
evm_addr = ault_to_evm("ault1...")        # → "0x1234...abcd"

# Validation
is_valid_ault_address("ault1...")  # True / False
is_valid_evm_address("0x1234...")  # True / False

# Chain ID
evm_chain_id = parse_evm_chain_id_from_cosmos_chain_id("ault_10904-1")  # → 10904

Constants

from ault_sdk import GAS_CONSTANTS, TIMING_CONSTANTS

GAS_CONSTANTS.EIP712_FEE_AMOUNT    # '5000000000000000'
GAS_CONSTANTS.EIP712_GAS_LIMIT     # '200000'
GAS_CONSTANTS.DENOM                # 'aault'
GAS_CONSTANTS.PER_LICENSE          # 200000

TIMING_CONSTANTS.API_TIMEOUT_MS    # 30000
TIMING_CONSTANTS.API_RETRY_DELAY_MS # 1000
TIMING_CONSTANTS.API_MAX_BACKOFF_MS # 30000

Development

Toolchain: uv + ty + ruff + just.

uv sync                  # install deps
just check               # lint + format + typecheck + unit tests
just test                # unit tests only
just test-e2e            # E2E (requires Docker + ../ault-dev monorepo)
just gen-proto           # regenerate protobuf bindings
just gen-openapi         # regenerate DEX REST client
just gen-eip712          # regenerate EIP-712 registry (native Python)
just gen-all             # regenerate everything

Codegen

All three codegen pipelines are native Python and read directly from the chain's source of truth:

• just gen-proto — buf + protoc-gen-python_betterproto over ../ault/proto + the curated ../cosmos-sdk/proto subset.
• just gen-openapi — openapi-python-client over specs/dex-openapi.yaml.
• just gen-eip712 — parses chain .proto files directly (text-based) and emits the EIP-712 registry, msg builders, and proto encoders.

Correctness across regens is pinned by the byte-exact signature tests (tests/unit/eip712/test_signers.py, tests/unit/agent/test_agent_signing.py): same private key + same input produces the same bytes as a known-good fixture. If anything in the registry, encoding, hashing, or signing paths drifts during a regen, these tests trip.

Testing

Unit tests run fast and require no external services:

just test                # ~3 seconds, 149 tests

E2E tests require Docker + the ault-dev monorepo checked out alongside this repo. They stand up a containerized localnet (chain + DEX + MySQL) per test file:

export GITHUB_TOKEN=...  # for the chain Dockerfile private deps
just test-e2e

Project Layout

src/ault_sdk/
├── __init__.py              # public API re-exports
├── client.py                # AultClient (low-level composite)
├── core/                    # errors, http, network, pagination, base64, address, …
├── eip712/                  # builder, registry, signers, sign-and-broadcast
│   ├── registry_generated.py     # generated
│   ├── msg_generated.py          # generated
│   └── msg_encoders_generated.py # generated
├── proto/
│   ├── tx_encode.py         # BinaryWriter + Cosmos Tx encoders
│   ├── duration_json.py
│   └── gen/                 # generated betterproto dataclasses
├── rest/                    # per-module REST clients
│   └── dex_generated/       # generated openapi-python-client
├── ws/                      # DEX WebSocket client
├── evm/                     # ERC-20, bridge, native, registry, client
├── agent/                   # agent tx signing + client
├── high_level/              # createClient factory, tx method bundles
└── utils/                   # address, chain_id, cosmos_dec

scripts/
├── generate_protos.py
├── generate_openapi.py
├── generate_eip712_registry.py

tests/
├── unit/                    # 149 tests
└── e2e/                     # Docker-based, requires ault-dev/

License

MIT.