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 collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

from ault_sdk import Manifest, 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
    manifest_override: Manifest | Mapping[str, Any] | None = None  # see "SDK manifest"

SDK manifest

create_client resolves network endpoints and gas/EIP-712 defaults from three layers, in increasing precedence:

1. The embedded baseline shipped with the SDK (NetworkConfig plus GAS_CONSTANTS).
2. The manifest fetched from <dex_api_url>/api/v1/sdk-manifest at bootstrap and refreshed every five minutes in the background. The operator rotates URLs, EIP-712 fee defaults, and per-module gas constants by editing the dex-api SDK_MANIFEST_JSON env var; older SDK versions pick up the new values within one refresh window without a package bump. The refresh runs on the caller's event loop via loop.call_later, so any custom httpx.AsyncClient transport, auth, or proxy configuration carries over to every tick.
3. An optional manifest_override you pass to create_client. The override sits on top of both other layers, is frozen for the lifetime of the client, and is not touched by background refreshes.

The override layer covers cases the fetched manifest cannot: pointing the client at a private RPC during local development, pinning a deterministic manifest in tests, surviving a URL rotation cutover by keeping the old endpoint until you redeploy, or tuning gas constants without waiting on a manifest deploy.

from ault_sdk import ClientOptions, Manifest, create_client, get_network_config, PrivateKeySigner

client = await create_client(
    ClientOptions(
        network=get_network_config("ault_10904-1"),
        signer=PrivateKeySigner("0x..."),
        manifest_override=Manifest(
            schema_version=1,
            urls={"rpc": "https://rpc.internal.example.com"},
            gas={"LIMIT_ORDER": 250_000},
        ),
    )
)

A plain mapping in the on-the-wire shape (camelCase keys including schemaVersion) works too:

client = await create_client(
    ClientOptions(
        network=get_network_config("ault_10904-1"),
        signer=PrivateKeySigner("0x..."),
        manifest_override={
            "schemaVersion": 1,
            "urls": {"rpc": "https://rpc.internal.example.com"},
            "gas": {"LIMIT_ORDER": 250_000},
        },
    )
)

The override is sparse — every field is optional, only the keys you set take effect, and disjoint keys flow through from the fetched manifest. Passing {"gas": {"LIMIT_ORDER": 250_000}} does not erase a MARKET_ORDER value the dex-api emits. Unknown top-level keys pass through untouched for forward compatibility with future manifest fields.

Supported fields

The Manifest dataclass has four top-level fields. schema_version is required; the three sub-mappings are optional and each one is itself sparse. On the wire-shape dict, the same keys appear as camelCase (schemaVersion).

@dataclass(frozen=True, slots=True)
class Manifest:
    schema_version: int                       # required, integer >= 1
    gas: Mapping[str, Any] | None = None
    eip712: Mapping[str, Any] | None = None
    urls: Mapping[str, Any] | None = None

urls — endpoint overrides. Each value is a non-empty string. The dex_api_url itself is not overridable here (it is the trust root for the fetched manifest).

URL rotations propagate to the tx executor on the next call but not to the REST/WS contexts captured at create_client time. Re-create the client to pick up new endpoints across the board.

eip712 — fee defaults applied to every transaction unless per-call options override them.

gas — per-message-type gas constants and the fee-free multiplier. Every value is a positive integer. Keys outside the supported list are dropped silently and the embedded baseline keeps applying.

The keys most consumers will touch are the CLOB ones plus the fee-free multiplier:

The schema also accepts per-message gas constants for the license, miner, agent, reward, and accountx modules (PER_LICENSE, PER_DELEGATE_MINING, APPROVE_AGENT, SET_REFERRAL_CODE, OPEN_ACCOUNT, and the rest). These exist for operator-side deploy-cycle bypass and are rarely needed in consumer code. The full list lives in GasConstants in src/ault_sdk/core/network.py.

The gas mapping never accepts EIP712_FEE_AMOUNT, EIP712_GAS_LIMIT, or DENOM — those belong on eip712 as feeAmount, gasLimit, and feeDenom. Smuggling them through gas is dropped because the broadcast layer never reads them off the gas mapping.

Both shapes route through parse_manifest, which validates the schema version and EIP-712 numeric fields synchronously at create_client time. A malformed override raises ValueError straight away instead of falling back to the baseline the way a broken fetched manifest does, since override data comes from your code and a typo is a bug worth surfacing loudly. The override is deep-cloned at construction, so mutating the dict or the dataclass's sub-mappings after create_client returns does not affect resolver state.

When a non-empty override is present, the resolver emits one logger.info line at boot listing the override key paths (never values, since urls may carry private endpoints).

The package also re-exports the low-level helpers if you need them: Manifest, merge_manifests, fetch_manifest, apply_network_overrides, apply_gas_overrides, create_manifest_resolver, create_static_resolver, parse_manifest.

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)

# Perpetual queries (chain v3)
collateral_denom = await client.perpetual.get_collateral_denom()
cross_account = await client.perpetual.get_account(client.address)
risk = await client.perpetual.get_account_risk(client.address)
position = await client.perpetual.get_position(client.address, market_id, cross=True)
funding = await client.perpetual.get_funding(market_id)
prices = await client.perpetual.get_prices(market_id)
impact_prices = await client.perpetual.get_impact_prices(market_id)

Perpetual markets

Perpetual markets (chain v3) trade on the same order book as spot. An order on a perpetual market is a regular place_limit_order; the only perp-specific flag is reduceOnly. The chain accepts limit orders only, so a "market order" is an immediate-or-cancel limit priced through the book:

from ault_sdk import ioc_limit_price

dex = client._low_level.rest.dex
book = await dex.get_perp_orderbook(market_id_hex, depth=5)
quote = ioc_limit_price(
    "buy",
    best_ask=book.asks[0].price if book.asks else None,
    best_bid=book.bids[0].price if book.bids else None,
    slippage="0.005",
    price_decimals=2,
)
await client.clob_tx.place_limit_order(
    marketId=market_id,
    isBuy=True,
    price=quote.price,
    quantity="1",
    deadline=deadline,
    timeInForce="ioc",
    reduceOnly=False,
)

Collateral lives in a per-owner cross account in the chain's single collateral denom; isolated markets get their own derived account. Margin mode and leverage are set per market and cannot change while a position or resting order exists:

await client.perpetual_tx.deposit_perp_collateral(amount="1000000000")  # base units
await client.perpetual_tx.set_margin_config(market_id=market_id, leverage=10, cross=True)
await client.perpetual_tx.modify_position_margin(market_id=market_id, margin_delta="-500000")
await client.perpetual_tx.withdraw_perp_collateral(amount="250000000")

A perpetual market carries the config x/perpetual trades it under: margin tiers, funding caps, the mark price bound and the oracle feed ids. The chain requires that config on MsgCreateMarket, rejects it on a spot market, and never lets a market change type afterwards. So create_market raises on a perpetual market type, and create_perpetual_market takes the config. The deployer can replace the whole config later with set_perpetual_market_config. Decimal fields take the same readable strings as every other SDK input ("0.0001"); packing the config scales them to the chain's 18-place integers.

from dataclasses import replace

from ault_sdk.proto.gen.ault.perpetual.v1 import MarginTier, PerpetualMarketConfig

config = PerpetualMarketConfig(
    isolated_only=False,
    funding_interest_rate="0.0001",  # per eight-hour interval
    funding_interest_adjustment_cap="0.0005",
    funding_hourly_cap="0.005",
    oracle_max_age_seconds=120,
    margin_tiers=[
        MarginTier(notional_cap="100000000000000000000000", max_leverage=20),
        MarginTier(notional_cap="0", max_leverage=10),  # "0" = unbounded last tier
    ],
    mark_price_max_change_rate="0.05",
    max_open_interest_notional="0",  # "0" = uncapped
    max_position_notional="0",
    impact_notional="10000",
    max_impact_levels=64,
    index_feed_id="BTC-USD",
    external_perpetual_feed_id="BTC-PERP-USD",
)

await client.market_tx.create_perpetual_market(
    base_denom="ubtc",
    quote_denom="uusdc",  # the module's collateral denom
    tick_precision=6,
    config=config,
)

await client.market_tx.set_perpetual_market_config(
    market_id=market_id,
    config=replace(config, oracle_max_age_seconds=60),
)

To build the config Any by hand, for a raw msg.market.create_market or an agent tx, use pack_perpetual_market_config. It applies the same scaling.

The same four collateral and margin methods exist on the agent client. Read models (markets, tickers with mark and funding, positions, fills, ledger) come from the DEX API's /api/v1/perp endpoints, list_perp_markets through list_perp_closeouts on DexApi (*_detailed variants feed paginate). ault_sdk.perp ports the chain's margin-tier, funding and position math (maintenance_margin, initial_margin, equity, funding_rate, ...) with the chain's rounding for client-side previews.

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)
trade = await dex.get_trade("12345-00010000000000003039000000000002-999")

# 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):
    # v2 l2Book: {coin, levels: [bids, asks]}, each level {px, sz, n}.
    print(data["levels"][0][0], data["levels"][1][0])

sub = ws.subscribe(
    {"type": "l2Book", "coin": "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(
    {"type": "trades", "coin": "AULT/USDT"},
    SubscriptionHandlers(
        on_data=lambda data, msg: print(data),
        on_subscribed=lambda ack: print("subscribed!", ack),
        on_expired=lambda: print("expired"),
    ),
    ttl=300,
)

# Unsubscribe
sub.unsubscribe()

# Teardown
await ws.disconnect()

User channels ({"type": "orderUpdates", "user": "ault1..."}, spotState, userFills) are read-only projections of public on-chain state, so no authentication is needed.

Transaction methods

All transaction methods are async, and most return a single TxResult. The exception is staking_tx.withdraw_rewards, which sends one transaction per validator and returns a list[TxResult] (one entry per validator, in input order — see Staking):

@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). A perpetual market takes its
# config at creation, so create_market raises on MarketType.PERPETUAL — see
# create_perpetual_market under "Perpetual markets".
await client.market_tx.create_market(
    market_type=1,
    base_denom="aault",
    quote_denom="ausdc",
    tick_precision=6,
)

# Limit order with an absolute deadline and an 8-byte client order id (cloid).
# deadline accepts a datetime, an RFC 3339 string, or a {seconds, nanos} dict.
from datetime import UTC, datetime, timedelta

cloid = b"\x01\x02\x03\x04\x05\x06\x07\x08"
await client.clob_tx.place_limit_order(
    marketId=bytes([1]),  # 16-byte market id
    isBuy=True,
    price="1.5",
    quantity="100",
    deadline=datetime.now(UTC) + timedelta(hours=1),
    cloid=cloid,
)

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

# Cancel by cloid (each order id may be an 8-byte cloid or a 16-byte order_id).
await client.clob_tx.cancel_orders(order_ids=[cloid])

# Resolve an order by cloid: chain-first (live orders), dex-api fallback (history).
found = await client.find_order_by_cloid(client.address, cloid)

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"},
)
# Withdraw rewards. The chain only accepts one message per transaction, so
# the SDK sends one tx per validator and returns a TxResult for each, in
# input order.
results = await client.staking_tx.withdraw_rewards(
    validator_addresses=["aultvaloper1...", "aultvaloper2..."]
)

If a later transaction fails after an earlier one has already committed, withdraw_rewards raises WithdrawRewardsPartialError instead of returning. Reconcile from partial_results rather than retrying the whole call, or you will double-claim the validators that already succeeded:

from ault_sdk import WithdrawRewardsPartialError

try:
    results = await client.staking_tx.withdraw_rewards(validator_addresses=validators)
except WithdrawRewardsPartialError as err:
    # err.partial_results: [{"tx_hash": ..., "validator_address": ...}] for
    # each claim that landed on-chain before the failure.
    # err.failed_validator_address: where the fan-out stopped.
    # err.cause / err.failed_result: why it stopped. cause = the broadcast
    #   raised. failed_result = the executor returned a failed TxResult, which
    #   is either a definitive chain rejection (confirmed) or a tx whose
    #   finality is unknown (unconfirmed — it may still have landed, so check
    #   failed_result.confirmed rather than code, and don't blindly retry it).
    handle_partial_withdrawal(err.partial_results)

Transaction options

Every tx method accepts optional gas_limit, fee_denom, 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",
)

fee_denom overrides the fee coin for a single transaction. Without it the client falls back to network.eip712_fee_denom, then to aault. You need it when the network default can't pay for a particular call: on mainnet, order placement has to use the ERC-20 denom listed in accountx.allowed_fees, but a fee-bearing cancel settles in aault.

When neither fee_amount nor network.eip712_fee_amount is set, the client prices the fee from the chain's feemarket params (GET /cosmos/evm/feemarket/v1/params, cached for 3 seconds per client): it takes the larger of base_fee and min_gas_price, multiplies by the gas limit, adds a 20% margin for base-fee moves between signing and inclusion, and never goes below eip712_fee_amount. A chain that does not expose the gateway gets the static floor. A transient gateway error is raised rather than silently signing an underpriced tx.

Staking gas

Cosmos staking messages (delegate, undelegate, redelegate, withdraw_rewards) are metered on execution. What a redelegation costs depends on the delegator's state, for example whether a delegation to the destination already exists or how many redelegation entries are open, so no fixed number is right for every account. These methods simulate the tx on the chain first (POST /cosmos/tx/v1beta1/simulate) and sign gas_used * SIMULATED_GAS_MULTIPLIER / 100 as the gas limit (1.3x by default, overridable through the manifest). The fee is computed from that gas.

If the chain rejects the simulated tx, the call raises TxSimulationError with the chain's message. Nothing is signed or broadcast, so you don't pay a fee for a tx that would have failed in the block. If the simulate endpoint is down or a proxy doesn't route it, the method logs a warning and falls back to the embedded per-message constant (DELEGATE, BEGIN_REDELEGATE, ...) times FEE_FREE_GAS_MULTIPLIER. Passing an explicit gas_limit skips the simulation.

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_40904-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.