ault-sdk 0.0.1a3

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

Ault SDK for Python

Python SDK for the Ault blockchain. Gives you high-level transaction methods and REST/WebSocket query clients for the Ault modules currently on chain: License, Miner, Market, CLOB, Agent, Authority, RateLimit, Reward, TokenRegistry, Staking, Bank.

Features

• High-level async client with simple transaction methods (no manual message building).
• Signer support for private keys, 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

pip install ault-sdk

Or with uv:

uv add ault-sdk

The distribution name on PyPI is ault-sdk; the import name is ault_sdk.

Requirements

• Python 3.12 or newer.

Quick start

import asyncio
from ault_sdk import create_client, ClientOptions, 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. Works for hosted signing services, browser-wallet bridges, or 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, and 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, so no authentication is 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

create_client returns a high-level Client. Use create_ault_client directly if you want the lower-level composite without the 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

Contributing

Bug reports, feature requests, and pull requests are welcome on the GitHub repository.

Local development uses uv, ruff, ty, and just:

uv sync            # install dependencies
just check         # lint + format + typecheck + unit tests
just test          # unit tests only

The EIP-712 signing paths are guarded by byte-exact signature tests against known-good fixtures, so any drift in the signing stack trips the test suite immediately.

License

Released under the MIT License. See the LICENSE file in the repository for the full text.