Client module¶

The AkashClient is the main entry point for interacting with Akash Network using RPC endpoints. It provides access to all blockchain modules and convenience methods for common operations.

Class definition¶

class AkashClient:
    def __init__(
        self,
        rpc_endpoint: str = "https://akash-rpc.polkachu.com:443",
        chain_id: str = "akashnet-2"
    )

Constructor parameters¶

Parameter	Type	Default	Description
`rpc_endpoint`	`str`	`"https://akash-rpc.polkachu.com:443"`	RPC server endpoint URL
`chain_id`	`str`	`"akashnet-2"`	Blockchain chain ID for transactions

Properties¶

rpc_endpoint¶

The RPC endpoint this client is connected to.

Returns: str - The RPC endpoint URL

chain_id¶

The blockchain chain ID this client is configured for.

Returns: str - The chain ID

Module clients¶

The AkashClient provides access to specialized module clients for different blockchain operations:

Distribution ¶

Distribution operations (rewards and commission)

Governance ¶

Governance operations (proposals, voting)

Market ¶

Marketplace operations (orders, bids, leases)

Staking ¶

Staking operations (delegation, validators, rewards)

Core methods¶

rpc_query()¶

def rpc_query(method: str, params: List[Any] = None) -> Dict[str, Any]

Perform an RPC query.

Parameters:

method (str): RPC method name
params (List[Any]): RPC parameters

Returns: Dict[str, Any] - RPC response with the following fields:

jsonrpc (str): JSON-RPC version
id (int): Request ID
result (Any): Method-specific result data
error (Dict, optional): Error information if request failed
- code (int): Error code
- message (str): Error message

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
result = client.rpc_query("status")
print(f"Latest block: {result['sync_info']['latest_block_height']}")

abci_query()¶

def abci_query(path: str, data: str = "", prove: bool = False) -> Dict[str, Any]

Perform an ABCI query via RPC.

Parameters:

path (str): Query path
data (str): Query data (hex encoded)
prove (bool): Include proof in response

Returns: Dict[str, Any] - ABCI query response with the following fields:

response (Dict): ABCI response
- code (int): Response code (0 for success)
- log (str): Response log
- info (str): Additional info
- index (str): Response index
- key (str): Query key (base64)
- value (str): Query value (base64)
- proof_ops (Dict, optional): Proof operations if requested
- height (str): Block height
- codespace (str): Code namespace

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
result = client.abci_query("/cosmos.staking.v1beta1.Query/Validators")
print(f"Query response code: {result['response']['code']}")

broadcast_tx_async()¶

def broadcast_tx_async(tx_bytes: bytes) -> Dict[str, Any]

Broadcast transaction using broadcast_tx_async.

⚠️ Getting code 0 and a transaction hash does NOT guarantee the transaction succeeded. The transaction is only queued for mempool inclusion. Use broadcast_tx_sync() or wait for blockchain confirmation to verify actual success.

Parameters:

tx_bytes (bytes): Serialized transaction bytes

Returns: Dict[str, Any] - Broadcast result with the following fields:

jsonrpc (str): JSON-RPC version
id (int): Request ID
result (Dict): Broadcast result
- code (int): Response code (0 for success)
- data (str): Response data
- log (str): Transaction log
- hash (str): Transaction hash

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
result = client.broadcast_tx_async(tx_bytes)
print(f"Transaction hash: {result.get('hash')}")

broadcast_tx_sync()¶

def broadcast_tx_sync(tx_bytes: bytes) -> Dict[str, Any]

Broadcast transaction using broadcast_tx_sync.

Parameters:

tx_bytes (bytes): Serialized transaction bytes

Returns: Dict[str, Any] - Broadcast result with the following fields:

jsonrpc (str): JSON-RPC version
id (int): Request ID
result (Dict): Broadcast result
- code (int): Response code (0 for success)
- data (str): Response data
- log (str): Transaction log
- hash (str): Transaction hash

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
result = client.broadcast_tx_sync(tx_bytes)
print(f"Transaction hash: {result.get('hash')}")

health_check()¶

def health_check(timeout: float = 10.0) -> bool

Check if the RPC endpoint is healthy.

Parameters:

timeout (float): Timeout in seconds

Returns: bool - True if healthy, False otherwise

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
if client.health_check():
    print("RPC endpoint is healthy")
else:
    print("RPC endpoint is not responding")

get_network_status()¶

def get_network_status() -> Dict[str, Any]

Get network status information.

Returns: Dict[str, Any] - Network status with the following fields:

network (str): Network/chain ID
latest_block_height (str): Latest block height
latest_block_time (str): Latest block timestamp
latest_block_hash (str): Latest block hash
earliest_block_height (str): Earliest available block height
earliest_block_time (str): Earliest block timestamp
catching_up (bool): Whether node is catching up
node_info (Dict): Node information
- protocol_version (Dict): Protocol versions
  - p2p (str): P2P protocol version
  - block (str): Block protocol version
  - app (str): Application protocol version
- id (str): Node ID
- listen_addr (str): Node listen address
- network (str): Network name
- version (str): Node version
- channels (str): Supported channels
- moniker (str): Node moniker
- other (Dict): Other node info
  - tx_index (str): Transaction indexing mode
  - rpc_address (str): RPC server address

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
status = client.get_network_status()
print(f"Network: {status['network']}")
print(f"Block height: {status['latest_block_height']}")

Convenience methods¶

get_balance()¶

def get_balance(address: str, denom: str = "uakt") -> str

Get account balance - delegates to BankClient.

Parameters:

address (str): Account address
denom (str): Token denomination

Returns: str - Balance amount

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
balance = client.get_balance("akash1qnnhhgzxj24f2kld5yhy4v4h4s9r295ak5gjw4")
print(f"Balance: {int(balance) / 1_000_000:.6f} AKT")

get_balances()¶

def get_balances(address: str) -> Dict[str, str]

Get all balances for an address - delegates to BankClient.

Parameters:

address (str): Account address

Returns: Dict[str, str] - Dictionary of balances by denomination with:

key (str): Token denomination (e.g., "uakt", "usdc")
value (str): Balance amount as string

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
balances = client.get_balances("akash1qnnhhgzxj24f2kld5yhy4v4h4s9r295ak5gjw4")
for denom, amount in balances.items():
    print(f"{denom}: {amount}")

get_validators()¶

def get_validators() -> List[Dict[str, Any]]

Get validators list.

Returns: List[Dict[str, Any]] - List of validator dictionaries with the following fields:

operator_address (str): Validator operator address
consensus_pubkey (Dict): Consensus public key
- @type (str): Key type URL
- key (str): Base64-encoded public key
jailed (bool): Whether validator is jailed
status (int): Validator status
tokens (str): Bonded tokens
delegator_shares (str): Delegator shares
description (Dict): Validator description
- moniker (str): Validator moniker
- identity (str): Identity signature
- website (str): Website URL
- security_contact (str): Security contact
- details (str): Description details
unbonding_height (str): Unbonding height
unbonding_time (str): Unbonding time
commission (Dict): Commission information
- commission_rates (Dict): Commission rates
  - rate (str): Current commission rate
  - max_rate (str): Maximum commission rate
  - max_change_rate (str): Maximum commission change rate
- update_time (str): Last commission update time
min_self_delegation (str): Minimum self delegation

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
validators = client.get_validators()
print(f"Found {len(validators)} active validators")

get_proposals()¶

def get_proposals() -> List[Dict[str, Any]]

Get governance proposals.

Returns: List[Dict[str, Any]] - List of proposal dictionaries with the following fields:

proposal_id (str): Proposal ID
content (Dict): Proposal content
- @type (str): Proposal type
- title (str): Proposal title
- description (str): Proposal description
status (int): Proposal status
final_tally_result (Dict): Final vote tally
- yes (str): Yes votes
- abstain (str): Abstain votes
- no (str): No votes
- no_with_veto (str): No with veto votes
submit_time (str): Submission timestamp
deposit_end_time (str): Deposit period end time
total_deposit (List[Dict]): Total deposit
- denom (str): Token denomination
- amount (str): Deposit amount
voting_start_time (str): Voting start time
voting_end_time (str): Voting end time

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
proposals = client.get_proposals()
print(f"Found {len(proposals)} governance proposals")

get_staking_params()¶

def get_staking_params() -> Dict[str, Any]

Get staking module parameters.

Returns: Dict[str, Any] - Staking parameters with the following fields:

unbonding_time (str): Unbonding period duration
max_validators (int): Maximum number of validators
max_entries (int): Maximum unbonding/redelegation entries
historical_entries (int): Number of historical entries
bond_denom (str): Bonding token denomination
min_commission_rate (str): Minimum commission rate

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
params = client.get_staking_params()
print(f"Unbonding time: {params.get('unbonding_time', 'N/A')}")

get_governance_params()¶

def get_governance_params() -> Dict[str, Any]

Get governance module parameters.

Returns: Dict[str, Any] - Governance parameters with the following fields:

voting_params (Dict): Voting parameters
- voting_period (str): Voting period duration
deposit_params (Dict): Deposit parameters
- min_deposit (List[Dict]): Minimum deposit
  - denom (str): Token denomination
  - amount (str): Minimum amount
- max_deposit_period (str): Maximum deposit period
tally_params (Dict): Tally parameters
- quorum (str): Quorum threshold
- threshold (str): Pass threshold
- veto_threshold (str): Veto threshold

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
params = client.get_governance_params()
print(f"Voting period: {params.get('voting_period', 'N/A')}")

get_account_info()¶

def get_account_info(address: str) -> Dict[str, Any]

Get account information - delegates to AuthClient.

Parameters:

address (str): Account address

Returns: Dict[str, Any] - Account information with the following fields:

@type (str): Account type
address (str): Account address
pub_key (Dict, optional): Public key information
- @type (str): Public key type
- key (str): Public key (base64)
account_number (str): Account number
sequence (str): Account sequence number

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
info = client.get_account_info("akash1qnnhhgzxj24f2kld5yhy4v4h4s9r295ak5gjw4")
print(f"Account number: {info.get('account_number', 0)}")
print(f"Sequence: {info.get('sequence', 0)}")

get_delegations()¶

def get_delegations(delegator_address: str) -> List[Dict[str, Any]]

Get delegations for an address.

Parameters:

delegator_address (str): Delegator address

Returns: List[Dict[str, Any]] - List of delegation dictionaries with the following fields:

delegation (Dict): Delegation information
- delegator_address (str): Delegator address
- validator_address (str): Validator address
- shares (str): Delegation shares
balance (Dict): Delegation balance
- denom (str): Token denomination
- amount (str): Delegated amount

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
delegations = client.get_delegations("akash1qnnhhgzxj24f2kld5yhy4v4h4s9r295ak5gjw4")
print(f"Active delegations: {len(delegations)}")

get_provider_endpoint()¶

def get_provider_endpoint(provider_address: str) -> str

Fetch provider's gRPC endpoint from on-chain data.

Parameters:

provider_address (str): Provider's Akash address

Returns: str - Provider's gRPC endpoint

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
endpoint = client.get_provider_endpoint("akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63")
print(f"Provider endpoint: {endpoint}")

Context manager support¶

The AkashClient supports context manager usage for automatic resource cleanup:

from akash import AkashClient

with AkashClient("https://akash-rpc.polkachu.com:443") as client:
    balance = client.bank.get_balance("akash1...")
    validators = client.staking.get_validators()

Usage examples¶

Basic setup¶

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

client = AkashClient(
    rpc_endpoint="https://rpc.sandbox-01.aksh.pw:443",
    chain_id="sandbox-01"
)

Health check¶

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

if client.health_check():
    print("RPC endpoint is healthy")

    status = client.get_network_status()
    print(f"Network: {status.get('network', '')}")
    print(f"Latest block: {status.get('latest_block_height', '0')}")
else:
    print("RPC endpoint not responding")

Account operations¶

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")
address = "akash1qnnhhgzxj24f2kld5yhy4v4h4s9r295ak5gjw4"

account_info = client.get_account_info(address)
print(f"Account number: {account_info.get('account_number', 0)}")
print(f"Sequence: {account_info.get('sequence', 0)}")

balance = client.get_balance(address)
print(f"AKT balance: {int(balance) / 1_000_000:.6f}")

balances = client.get_balances(address)
for denom, amount in balances.items():
    print(f"{denom}: {amount}")

Network information¶

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

validators = client.get_validators()
print(f"Active validators: {len(validators)}")

staking_params = client.get_staking_params()
print(f"Unbonding time: {staking_params.get('unbonding_time', 'N/A')}")

proposals = client.get_proposals()
print(f"Active proposals: {len(proposals)}")

gov_params = client.get_governance_params()
print(f"Voting period: {gov_params.get('voting_period', 'N/A')}")

Provider operations¶

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

provider_address = "akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63"
try:
    endpoint = client.get_provider_endpoint(provider_address)
    print(f"Provider endpoint: {endpoint}")
except Exception as e:
    print(f"Failed to get provider endpoint: {e}")