Cosmos client

class xchainpy2_cosmos.cosm_client.CosmosGaiaClient(network=NetworkType.MAINNET, phrase: str | None = None, private_key: str | bytes | callable | None = None, root_derivation_paths: Dict[NetworkType, str] | None = None, client_urls={NetworkType.MAINNET: 'https://lcd-cosmoshub.keplr.app', NetworkType.STAGENET: 'https://lcd-cosmoshub.keplr.app', NetworkType.TESTNET: 'https://rest.sentry-02.theta-testnet.polypore.xyz'}, chain_ids={NetworkType.MAINNET: 'cosmoshub-4', NetworkType.STAGENET: 'cosmoshub-4', NetworkType.TESTNET: 'theta-testnet-001'}, explorer_providers={NetworkType.MAINNET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.STAGENET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.TESTNET: ('https://explorer.theta-testnet.polypore.xyz/', 'https://explorer.theta-testnet.polypore.xyz/accounts/{address}', 'https://explorer.theta-testnet.polypore.xyz/transactions/{tx_id}')}, wallet_index=0)

Bases: XChainClient

async broadcast_tx(tx_hex: bytes | str) → str

Broadcast a transaction to the Cosmos-based network.

Parameters:: tx_hex – Either a hex string or bytes representing the transaction.
Returns:: str Transaction hash (tx digest).

build_transfer_tx(what: CryptoAmount, recipient: str) → Transaction

property chain_id: str

Get the current chain id for the Cosmos network.

Returns:: str Chain ID

property client_urls

convert_coin_to_amount(c: Coin)

async estimate_gas_of_transfer(what: CryptoAmount, recipient: str, memo: str | None = None, gas: Gas | None = None) → CryptoAmount

Since “simulate_tx” is broken in cosmpy, we use the standard transaction fee as an estimate. No simulation is performed.

Parameters:

what – CryptoAmount to transfer
recipient – Recipient address
memo – Optional memo for the transaction
gas – Optional Gas options. If not provided, the default gas limit will be used.

Returns:

CryptoAmount representing the estimated gas

async fetch_chain_id(server='') → str

Helper to get Cosmos’ chain id.

Returns:: str Chain ID

async get_account(address: str | None = None) → Account | None: Get the account information for the given address: number and sequence. It there is no account, it will return None. It will throw an exception for special addresses (like Reserve) :param address: By default, it will return the account of the current wallet. (optional) :return:

get_address() → str: Get the address for the given wallet index. :return: string address

get_amount_string(amount)

Builds a string representation of the amount in the format required by Cosmos. For instance, “123atom”

Parameters:: amount – Amount to convert, should be an instance of Amount or CryptoAmount, or just int
Returns:: str

async get_balance(address: str = '', **kwargs) → List[CryptoAmount]

Get the balances of a given address.

Parameters:: address – By default, it will return the balance of the current wallet. (optional)
Returns:

get_client() → LedgerClient: Please use this getter to obtain LedgerClient for specific Cosmos calls like delegation, staking etc. Underlying library is cosmpy (https://github.com/fetchai/cosmpy) :return: LedgerClient

get_denom(asset: Asset) → str

Converts an Asset to its corresponding denomination string. If the asset is AssetATOM, it returns the constant “uatom”.

Parameters:: asset – Asset instance (e.g., AssetATOM or any other asset)
Returns:: str Denomination string for the asset

async get_fees() → FlatFee

get_network()

Get the network type for this client.

Returns:: NetworkType

get_private_key_cosmos() → PrivateKey

get_public_key() → PublicKey

Get the public key for the given wallet index. # todo: make it consistent for all clients, some return hex, some return bytes!

Returns:: str Public key as hex string.

async get_transaction_data(tx_id: str, our_address: str = '') → XcTx | None

Get the transaction data for the given transaction id.

Parameters:

tx_id
our_address

Returns:

async get_transaction_data_cosmos(tx_id: str) → dict: Get the transaction data for the given transaction id. :param tx_id: :return: raw dict object

async get_transaction_data_raw(tx_id: str) → dict

async get_transactions(address: str = '', offset: int = 0, limit: int = 10, start_time: datetime | None = None, end_time: datetime | None = None, asset: Asset | None = None, filter_function: Callable[[object], bool] | None = None, batch_size=10, batch_delay_sec=2) → TxPage

parse_denom_to_asset(denom: str) → Asset

patch_client(user_agent='Mozilla/5.0 (Windows NT 10.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36', identifier9r='xchainpy-client')

Patch the REST client with custom user agent and client id header (x-client-id).

Parameters:

user_agent – User agent string to set in the headers.
identifier9r – Client identifier for Nine Realms servers (x-client-id).

property prefix: str

purge_client(): Purge the client by clearing the phrase and private key.

property rest_session: ClientSession

property rpc_url

property sdk_client: LedgerClient

async search_tx(message_action=None, message_sender=None, offset=0, limit=50)

async search_tx_from_rpc(message_action=None, message_sender=None, transfer_sender=None, transfer_recipient=None, page=None, limit=None, tx_min_height=None, tx_max_height=None, rpc_endpoint=None)

property server_url

set_chain_id(chain_id: str): Set/update the current chain id. :param chain_id: :return: None

set_network(network: NetworkType): Set/update the current network. :param network: :return: None

set_phrase(phrase: str, wallet_index: int = 0)

Set the seed phrase for the client. It will also set the wallet index.

Parameters:

phrase – Mnemonic phrase (12-24 words)
wallet_index – Wallet index (default is 0)

async transfer(what: CryptoAmount, recipient: str, memo: str | None = None, gas: Gas | None = None, check_balance: bool = True, **kwargs) → str

Transfer coins to another address with an optional memo.

Parameters:

what – CryptoAmount (amount and asset to transfer)
recipient – str recipient address
memo – str
gas – Gas options. If not provided, the default gas limit will be used.
check_balance – Check balance before transfer. Default is True.

Returns:

str tx hash

url_to_fetch_tx_data(tx_id, server_url=None)

validate_address(address: str) → bool: Validates a Cosmos address. :param address: Address string :return: True if valid, False otherwise.

Cosmos constants

xchainpy2_cosmos.const.BIG_DIPPER_EXPLORER_PROVIDER = {NetworkType.MAINNET: ('https://bigdipper.live/cosmos/', 'https://bigdipper.live/cosmos/accounts/{address}', 'https://bigdipper.live/cosmos/transactions/{tx_id}'), NetworkType.STAGENET: ('https://bigdipper.live/cosmos/', 'https://bigdipper.live/cosmos/accounts/{address}', 'https://bigdipper.live/cosmos/transactions/{tx_id}'), NetworkType.TESTNET: ('https://explorer.theta-testnet.polypore.xyz/', 'https://explorer.theta-testnet.polypore.xyz/accounts/{address}', 'https://explorer.theta-testnet.polypore.xyz/transactions/{tx_id}')}: Big Dipper explorer provider for Cosmos networks.

xchainpy2_cosmos.const.CLIENT_URL_COSMOSTATION = 'https://lcd-cosmos.cosmostation.io/': Cosmostation client URL for Cosmos Hub.

xchainpy2_cosmos.const.CLIENT_URL_COSMOS_DIRECTORY = 'https://rest.cosmos.directory/cosmoshub': Cosmos Directory client URL for Cosmos Hub.

xchainpy2_cosmos.const.CLIENT_URL_KEPLR = 'https://lcd-cosmoshub.keplr.app': Keplr client URL for Cosmos Hub.

xchainpy2_cosmos.const.COSMOS_ADDR_PREFIX = 'cosmos': Cosmos address prefix.

xchainpy2_cosmos.const.COSMOS_CHAIN_IDS = {NetworkType.MAINNET: 'cosmoshub-4', NetworkType.STAGENET: 'cosmoshub-4', NetworkType.TESTNET: 'theta-testnet-001'}: Default chain IDs for Cosmos networks.

xchainpy2_cosmos.const.COSMOS_DECIMAL = 6: Cosmos Atom decimal places.

xchainpy2_cosmos.const.COSMOS_DENOM = 'uatom': Cosmos Atom denomination.

xchainpy2_cosmos.const.COSMOS_ROOT_DERIVATION_PATHS = {NetworkType.MAINNET: "44'/118'/0'/0/", NetworkType.STAGENET: "44'/118'/0'/0/", NetworkType.TESTNET: "44'/118'/0'/0/"}: Default root derivation paths for Cosmos accounts.

xchainpy2_cosmos.const.DEFAULT_CLIENT_URLS = {NetworkType.MAINNET: 'https://lcd-cosmoshub.keplr.app', NetworkType.STAGENET: 'https://lcd-cosmoshub.keplr.app', NetworkType.TESTNET: 'https://rest.sentry-02.theta-testnet.polypore.xyz'}: Default client URLs for Cosmos networks. Now it uses Keplr for Mainnet and Stagenet, and Theta Testnet for Testnet.

xchainpy2_cosmos.const.DEFAULT_DERIVATION_PATH = "44'/118'/0'/0/": Default derivation path for Cosmos accounts.

xchainpy2_cosmos.const.DEFAULT_EXPLORER_PROVIDER = {NetworkType.MAINNET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.STAGENET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.TESTNET: ('https://explorer.theta-testnet.polypore.xyz/', 'https://explorer.theta-testnet.polypore.xyz/accounts/{address}', 'https://explorer.theta-testnet.polypore.xyz/transactions/{tx_id}')}: Default explorer provider for Cosmos networks. Now it uses Mint Scan for Mainnet and Stagenet.

xchainpy2_cosmos.const.DEFAULT_FEE = ((1000, 6), ('GAIA', 'ATOM', '', AssetKind.NATIVE)): Default Max fee for Cosmos transactions. Type: CryptoAmount It is calculated as DEFAULT_GAS_LIMIT * FEE_MINIMUM_GAS_PRICE.

xchainpy2_cosmos.const.DEFAULT_GAS_LIMIT = 200000: Default gas limit for Cosmos transactions. It should be enough for most of the transfer transactions with MEMO.

xchainpy2_cosmos.const.DEFAULT_REST_USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36': Default user agent for REST requests.

xchainpy2_cosmos.const.FEE_MINIMUM_GAS_PRICE = 0.005: Minimum gas price for Cosmos transactions. On Cosmos Hub (chain ID cosmoshub-4), the default recommended value is 0.025 uatom per gas unit.

xchainpy2_cosmos.const.MAX_PAGES_PER_FUNCTION_CALL = 15: Maximum number of pages to fetch Txs per function call.

xchainpy2_cosmos.const.MAX_TX_COUNT_PER_FUNCTION_CALL = 500: Maximum number of transactions to fetch per function call.

xchainpy2_cosmos.const.MAX_TX_COUNT_PER_PAGE = 100: Maximum number of transactions to fetch per page.

xchainpy2_cosmos.const.MINT_SCAN_EXPLORER_PROVIDER = {NetworkType.MAINNET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.STAGENET: ('https://www.mintscan.io/cosmos/', 'https://www.mintscan.io/cosmos/account/{address}', 'https://www.mintscan.io/cosmos/txs/{tx_id}'), NetworkType.TESTNET: ('https://explorer.theta-testnet.polypore.xyz/', 'https://explorer.theta-testnet.polypore.xyz/accounts/{address}', 'https://explorer.theta-testnet.polypore.xyz/transactions/{tx_id}')}: Mint Scan explorer provider for Cosmos networks.

xchainpy2_cosmos.const.TEST_EXPLORER_URL = 'https://explorer.theta-testnet.polypore.xyz/{path}': Testnet explorer URL for Cosmos.

xchainpy2_cosmos.const.make_explorer(path, acc_subpath, tx_subpath)

xchainpy2_cosmos.const.make_explorer_mint_scan()

xchainpy2_cosmos.const.make_explorer_std(path)

Cosmos Utils

xchainpy2_cosmos.utils.convert_address_for_msg(address, prefix: str) → bytes

xchainpy2_cosmos.utils.get_asset(denom: str, native_denom='uatom', native_asset=('GAIA', 'ATOM', '', AssetKind.NATIVE)) → Asset | None: Get Asset from denomination - currently ATOM supported only :param denom: :param native_asset: :param native_denom: :return: The asset of the given denomination.

xchainpy2_cosmos.utils.get_coin_amount(coins, decimals) → Amount

xchainpy2_cosmos.utils.get_coins_by_asset(coins, search_asset: Asset, native_denom: str, native_asset: Asset)

xchainpy2_cosmos.utils.get_denom(asset: Asset) → str: Get denomination from Asset - currently ATOM supported only :param asset: :return: The denomination of the given asset.

xchainpy2_cosmos.utils.is_msg_multi_send(msg)

xchainpy2_cosmos.utils.is_msg_send(msg)

xchainpy2_cosmos.utils.parse_cosmos_amount(amount: str) → Tuple[int, str]: Parse COSMOS amount into integer amount and string denomination :param amount: a string with format “123asset”, like “100rune” :return: (int, str)

xchainpy2_cosmos.utils.parse_cosmos_amounts(value: str) → List[Tuple[int, str]]: Parse COSMOS amounts from string like “1234rune,1234bnb” :param value: Cosmos SDK amount string :return: List of (value, asset) pairs

xchainpy2_cosmos.utils.parse_transfer_log(log: TxLog, decimals, filter_address, native_denom: str, native_asset: Asset) → List[TokenTransfer]

xchainpy2_cosmos.utils.parse_tx_response_json(j: dict, tx_id: str, address: str, decimals: int, native_denom: str, native_asset: Asset) → XcTx

Cosmos models

class xchainpy2_cosmos.models.RPCResponse(jsonrpc, id, result)

Bases: NamedTuple

classmethod from_rpc_json(j)

id: int: Alias for field number 1

jsonrpc: str: Alias for field number 0

result: dict: Alias for field number 2

class xchainpy2_cosmos.models.TxEvent(type, attributes)

Bases: NamedTuple

property as_dict

property as_dict_of_list

attributes: List[TxEventAttribute]: Alias for field number 1

find_attr_value_first(key, default=None) → str | None

find_attributes(key)

classmethod from_rpc_json(j)

type: str: Alias for field number 0

class xchainpy2_cosmos.models.TxEventAttribute(key, value)

Bases: NamedTuple

classmethod from_rpc_json(j)

key: str: Alias for field number 0

value: str: Alias for field number 1

class xchainpy2_cosmos.models.TxHistoryResponse(page_number, page_total, limit, total, tx_responses, txs)

Bases: NamedTuple

classmethod from_rpc_json(j)

limit: int: Alias for field number 2

page_number: int: Alias for field number 0

page_total: int: Alias for field number 1

total: int: Alias for field number 3

tx_responses: List[TxResponse]: Alias for field number 4

txs: List[dict]: Alias for field number 5

exception xchainpy2_cosmos.models.TxInternalException: Bases: Exception

exception xchainpy2_cosmos.models.TxLoadException: Bases: Exception

class xchainpy2_cosmos.models.TxLog(msg_index, log, events)

Bases: NamedTuple

events: List[TxEvent]: Alias for field number 2

find_event(event_name)

find_events(event_name) → List[TxEvent]

classmethod from_rpc_json(j)

log: str: Alias for field number 1

msg_index: int: Alias for field number 0

class xchainpy2_cosmos.models.TxResponse(height, txhash, data, raw_log, logs, gas_wanted, gas_used, tx, timestamp)

Bases: NamedTuple

data: str: Alias for field number 2

classmethod from_rpc_json(j)

gas_used: str: Alias for field number 6

gas_wanted: str: Alias for field number 5

height: int: Alias for field number 0

logs: list: Alias for field number 4

raw_log: str: Alias for field number 3

timestamp: str: Alias for field number 8

tx: dict | None: Alias for field number 7

txhash: str: Alias for field number 1

xchainpy2_cosmos.models.load_logs(j_logs)