Staking¶

The Staking module provides staking operations including delegation, undelegation, redelegation, and validator management for the Akash Network.

Transaction parameters

All transaction functions support flexible parameters. See Transaction parameters for details on available options like fee_amount, gas_limit, and gas_adjustment.

StakingClient class¶

from akash import AkashClient

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

staking = client.staking

Transactions¶

delegate()¶

Delegate tokens to a validator with enhanced gas simulation.

def delegate(wallet, validator_address: str, amount: str, denom: str = "uakt",
             memo: str = "", fee_amount: str = None, gas_limit: int = None,
             gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Delegator wallet instance
validator_address (str): Validator address to delegate to
amount (str): Amount to delegate

Optional parameters:

denom (str): Token denomination (default: "uakt")
memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt
gas_limit (int): Gas limit override
gas_adjustment (float): Gas estimation multiplier (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("delegator mnemonic here")

result = client.staking.delegate(
    wallet=wallet,
    validator_address="akashvaloper1...",
    amount="5000000",
    memo=""
)

if result.success:
    print(f"Delegation successful: {result.tx_hash}")
else:
    print(f"Delegation failed: {result.log}")

undelegate()¶

Undelegate tokens from a validator using RPC-only broadcasting.

def undelegate(wallet, validator_address: str, amount: str, denom: str = "uakt",
               memo: str = "", fee_amount: str = None, gas_limit: int = None,
               gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (Wallet): Delegator wallet instance
validator_address (str): Validator address to undelegate from
amount (str): Amount to undelegate

Optional parameters:

denom (str): Token denomination (default: "uakt")
memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt
gas_limit (int): Gas limit override
gas_adjustment (float): Gas estimation multiplier (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("delegator mnemonic here")

result = client.staking.undelegate(
    wallet=wallet,
    validator_address="akashvaloper1...",
    amount="2000000",
    memo=""
)

if result.success:
    print(f"Undelegation started: {result.tx_hash}")
    print("Note: 21-day unbonding period applies")

redelegate()¶

Redelegate tokens from one validator to another using RPC-only broadcasting.

def redelegate(wallet, src_validator_address: str, dst_validator_address: str, amount: str,
               denom: str = "uakt", memo: str = "", fee_amount: str = None, gas_limit: int = None,
               gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (Wallet): Delegator wallet instance
src_validator_address (str): Source validator address
dst_validator_address (str): Destination validator address
amount (str): Amount to redelegate

Optional parameters:

denom (str): Token denomination (default: "uakt")
memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt
gas_limit (int): Gas limit override (default: 300000)
gas_adjustment (float): Gas estimation multiplier (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("delegator mnemonic here")

result = client.staking.redelegate(
    wallet=wallet,
    src_validator_address="akashvaloper1src...",
    dst_validator_address="akashvaloper1dst...",
    amount="3000000",
    memo=""
)

if result.success:
    print(f"Redelegation successful: {result.tx_hash}")

withdraw_rewards()¶

Withdraw delegation rewards from a validator using RPC-only broadcasting.

def withdraw_rewards(wallet, validator_address: str, memo: str = "",
                     fee_amount: str = None, gas_limit: int = None,
                     gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (Wallet): Delegator wallet instance
validator_address (str): Validator address to withdraw rewards from

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt
gas_limit (int): Gas limit override
gas_adjustment (float): Gas estimation multiplier (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("delegator mnemonic here")

result = client.staking.withdraw_rewards(
    wallet=wallet,
    validator_address="akashvaloper1...",
    memo=""
)

if result.success:
    print(f"Rewards withdrawn: {result.tx_hash}")

withdraw_all_rewards()¶

Withdraw all delegation rewards from all validators using RPC-only broadcasting.

def withdraw_all_rewards(wallet, memo: str = "", fee_amount: str = None,
                         gas_limit: int = None) -> BroadcastResult

Required parameters:

wallet (Wallet): Delegator wallet instance

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt
gas_limit (int): Gas limit override

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("delegator mnemonic here")

result = client.staking.withdraw_all_rewards(
    wallet=wallet,
    memo=""
)

if result.success:
    print(f"All rewards withdrawn: {result.tx_hash}")

create_validator()¶

Create a new validator.

def create_validator(wallet, validator_info: Dict[str, Any], memo: str = "",
                     fee_amount: str = "5000", gas_limit: int = None,
                     use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (Wallet): Validator operator wallet instance
validator_info (Dict): Dictionary with validator information containing:
- description (Dict, required): Validator description with:
  - moniker (str, required): Validator display name
  - identity (str, optional): Keybase identity (default: "")
  - website (str, optional): Validator website URL (default: "")
  - security_contact (str, optional): Security contact email (default: "")
  - details (str, optional): Additional validator details (default: "")
- commission (Dict, required): Commission configuration with:
  - rate (str, required): Initial commission rate (e.g., "0.10")
  - max_rate (str, required): Maximum commission rate (e.g., "0.20")
  - max_change_rate (str, required): Maximum daily rate change (e.g., "0.01")
- min_self_delegation (str, required): Minimum self delegation in base units
- pubkey (str, required): Validator consensus public key
- value (Dict, required): Initial delegation amount with:
  - denom (str, required): Token denomination (typically "uakt")
  - amount (str, required): Delegation amount in base units

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt (default: "5000")
gas_limit (int): Gas limit
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("validator operator mnemonic here")

validator_info = {
    "description": {
        "moniker": "My Validator",
        "identity": "",
        "website": "https://myvalidator.com",
        "security_contact": "security@myvalidator.com",
        "details": "Professional Akash validator"
    },
    "commission": {
        "rate": "0.10",
        "max_rate": "0.20",
        "max_change_rate": "0.01"
    },
    "min_self_delegation": "1000000",
    "pubkey": validator_pubkey,
    "value": {"denom": "uakt", "amount": "1000000"}
}

result = client.staking.create_validator(
    wallet=wallet,
    validator_info=validator_info
)

edit_validator()¶

Edit an existing validator.

def edit_validator(wallet, validator_address: str = None, description: Dict[str, Any] = None,
                   commission_rate: str = None, min_self_delegation: str = None,
                   memo: str = "", fee_amount: str = "5000", gas_limit: int = 200000,
                   use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (Wallet): Validator operator wallet instance

At least one parameter must be provided

You must specify at least one of: description, commission_rate, or min_self_delegation

Optional parameters:

validator_address (str): Validator operator address (auto-derived if not provided)
description (Dict): Updated description fields
- moniker (str): Validator display name
- identity (str): Keybase identity
- website (str): Validator website URL
- security_contact (str): Security contact email
- details (str): Additional validator details
commission_rate (str): New commission rate
min_self_delegation (str): New minimum self delegation
memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt (default: "5000")
gas_limit (int): Gas limit (default: 200000)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("validator operator mnemonic here")

description = {
    "moniker": "Updated Validator Name",
    "website": "https://newwebsite.com",
    "details": "Updated validator description"
}

result = client.staking.edit_validator(
    wallet=wallet,
    description=description,
    commission_rate="0.05"
)

Queries¶

get_validators()¶

Get validators list with optional status filtering.

def get_validators(status: str = None, limit: int = None, offset: int = None, count_total: bool = False) -> List[Dict[str, Any]]

Optional parameters:

status (str): Validator status filter. Valid values:
- "BOND_STATUS_UNSPECIFIED" (or "0")
- "BOND_STATUS_UNBONDED" (or "1")
- "BOND_STATUS_UNBONDING" (or "2")
- "BOND_STATUS_BONDED" (or "3")
limit (int): Maximum number of results to return
offset (int): Number of results to skip (default: 0)
count_total (bool): Include total count in response (default: False)

Returns: List[Dict] - List of validator information with the following fields:

operator_address (str): Validator operator address
consensus_pubkey (str): Consensus public key
jailed (bool): Whether validator is jailed
status (str): Validator status
tokens (str): Total bonded tokens
delegator_shares (str): Total delegator shares
description (Dict): Validator description
- moniker (str): Validator display name
- identity (str): Keybase identity
- website (str): Validator website URL
- security_contact (str): Security contact email
- details (str): Additional description text
commission (Dict): Commission information
- commission_rates (Dict): Commission rate details
  - rate (str): Current commission rate
  - max_rate (str): Maximum commission rate
  - max_change_rate (str): Maximum daily rate change
min_self_delegation (str): Minimum self delegation amount

Example:

from akash import AkashClient

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

bonded_validators = client.staking.get_validators("BOND_STATUS_BONDED")
print(f"Found {len(bonded_validators)} bonded validators")

validators_page1 = client.staking.get_validators(
    status="BOND_STATUS_BONDED",
    limit=10,
    count_total=True
)
print(f"Page 1: {len(validators_page1)} validators")

validators_page2 = client.staking.get_validators(
    status="BOND_STATUS_BONDED",
    limit=10,
    offset=10
)

get_validator()¶

Get information about a specific validator.

def get_validator(validator_address: str) -> Optional[Dict[str, Any]]

Required parameters:

validator_address (str): Validator operator address

Returns: Optional[Dict] - Validator information with the following structure or None if not found:

operator_address (str): Validator operator address
consensus_pubkey (str): Consensus public key
jailed (bool): Whether validator is jailed
status (str): Validator status
tokens (str): Total bonded tokens
delegator_shares (str): Total delegator shares
description (Dict): Validator description
- moniker (str): Validator display name
- identity (str): Keybase identity
- website (str): Validator website URL
- security_contact (str): Security contact email
- details (str): Additional validator details
commission (Dict): Commission information
- commission_rates (Dict): Commission rate details
  - rate (str): Current commission rate
  - max_rate (str): Maximum commission rate
  - max_change_rate (str): Maximum daily rate change
min_self_delegation (str): Minimum self delegation amount

Example:

from akash import AkashClient

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

validator = client.staking.get_validator("akashvaloper1...")
if validator:
    print(f"Validator: {validator['description']['moniker']}")
    print(f"Jailed: {validator['jailed']}")

get_delegations()¶

Get delegations for a delegator address.

def get_delegations(delegator_address: str, limit: int = None, offset: int = None, count_total: bool = False) -> List[Dict[str, Any]]

Required parameters:

delegator_address (str): Delegator's address

Optional parameters:

limit (int): Maximum number of results to return
offset (int): Number of results to skip (default: 0)
count_total (bool): Include total count in response (default: False)

Returns: List[Dict] - List of delegations with the following structure:

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

Example:

from akash import AkashClient

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

delegations = client.staking.get_delegations("akash1...")
for delegation in delegations:
    validator = delegation['delegation']['validator_address']
    amount = delegation['balance']['amount']
    print(f"Delegated {amount} uakt to {validator}")

get_delegation()¶

Get specific delegation information.

def get_delegation(delegator_address: str, validator_address: str) -> Optional[Dict[str, Any]]

Required parameters:

delegator_address (str): Delegator's address
validator_address (str): Validator's operator address

Returns: Optional[Dict] - Delegation information with the following structure or None if not found:

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

Example:

from akash import AkashClient

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

delegation = client.staking.get_delegation("akash1...", "akashvaloper1...")
if delegation:
    amount = delegation['balance']['amount']
    print(f"Delegation amount: {amount} uakt")

get_staking_params()¶

Get staking parameters.

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

Returns: Dict - Staking parameters with the following fields:

unbonding_time (str): Time required for unbonding
max_validators (int): Maximum number of validators
max_entries (int): Maximum number of unbonding delegation entries
historical_entries (int): Number of historical entries to persist
bond_denom (str): Denomination used for staking

Example:

from akash import AkashClient

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

params = client.staking.get_staking_params()
print(f"Unbonding time: {params['unbonding_time']}")
print(f"Max validators: {params['max_validators']}")
print(f"Bond denom: {params['bond_denom']}")

get_unbonding_delegations()¶

Get unbonding delegations for a delegator.

def get_unbonding_delegations(delegator_address: str, limit: int = None, offset: int = None, count_total: bool = False) -> List[Dict[str, Any]]

Required parameters:

delegator_address (str): Delegator's address

Optional parameters:

limit (int): Maximum number of results to return
offset (int): Number of results to skip (default: 0)
count_total (bool): Include total count in response (default: False)

Returns: List[Dict] - List of unbonding delegations with the following structure:

delegator_address (str): Delegator's address
validator_address (str): Validator's address
entries (List[Dict]): List of unbonding entries
- creation_height (str): Block height when unbonding started
- completion_time (str): ISO timestamp when unbonding completes
- initial_balance (str): Initial unbonding amount
- balance (str): Current unbonding balance

Example:

from akash import AkashClient

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

unbonding = client.staking.get_unbonding_delegations("akash1...")
for unbond in unbonding:
    validator = unbond['validator_address']
    print(f"Unbonding from {validator}:")
    for entry in unbond['entries']:
        print(f"Amount: {entry['balance']} uakt")
        print(f"Completion: {entry['completion_time']}")

get_redelegations()¶

Get redelegations for a delegator.

def get_redelegations(delegator_address: str, src_validator_address: str = "", dst_validator_address: str = "") -> List[Dict[str, Any]]

Required parameters:

delegator_address (str): Delegator's address

Optional parameters:

src_validator_address (str): Source validator address filter
dst_validator_address (str): Destination validator address filter

Returns: List[Dict] - List of redelegations with the following structure:

delegator_address (str): Delegator's address
validator_src_address (str): Source validator address
validator_dst_address (str): Destination validator address
entries (List[Dict]): List of redelegation entries
- creation_height (str): Block height when redelegation started
- completion_time (str): ISO timestamp when redelegation completes
- initial_balance (str): Initial redelegation amount
- shares_dst (str): Destination validator shares
- balance (str): Current redelegation balance

Example:

from akash import AkashClient

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

redelegations = client.staking.get_redelegations("akash1...")
for redel in redelegations:
    src = redel['validator_src_address']
    dst = redel['validator_dst_address']
    print(f"Redelegating from {src} to {dst}")

get_delegations_to_validator()¶

Get all delegations to a validator.

def get_delegations_to_validator(validator_address: str) -> List[Dict[str, Any]]

Required parameters:

validator_address (str): Validator's operator address

Returns: List[Dict] - List of delegations to the validator with the following structure:

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

Example:

from akash import AkashClient

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

delegations = client.staking.get_delegations_to_validator("akashvaloper1...")
print(f"Found {len(delegations)} delegations to this validator")
for delegation in delegations:
    delegator = delegation['delegation']['delegator_address']
    amount = delegation['balance']['amount']
    print(f"{delegator}: {amount} uakt")

get_redelegations_from_validator()¶

Get redelegations from a validator.

def get_redelegations_from_validator(src_validator_address: str) -> List[Dict[str, Any]]

Required parameters:

src_validator_address (str): Source validator address

Returns: List[Dict] - List of redelegations from the validator with the following structure:

delegator_address (str): Delegator's address
validator_src_address (str): Source validator address
validator_dst_address (str): Destination validator address
entries (List[Dict]): List of redelegation entries
- creation_height (str): Block height when redelegation started
- completion_time (str): ISO timestamp when redelegation completes
- initial_balance (str): Initial redelegation amount
- shares_dst (str): Destination validator shares
- balance (str): Current redelegation balance

Example:

from akash import AkashClient

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

redelegations = client.staking.get_redelegations_from_validator("akashvaloper1...")
print(f"Found {len(redelegations)} outgoing redelegations")
for redel in redelegations:
    dst = redel['validator_dst_address']
    delegator = redel['delegator_address']
    print(f"{delegator} -> {dst}")

get_unbonding_delegations_from_validator()¶

Get unbonding delegations from a validator.

def get_unbonding_delegations_from_validator(validator_address: str) -> List[Dict[str, Any]]

Required parameters:

validator_address (str): Validator's operator address

Returns: List[Dict] - List of unbonding delegations from the validator with the following structure:

delegator_address (str): Delegator's address
validator_address (str): Validator's address
entries (List[Dict]): List of unbonding entries
- creation_height (str): Block height when unbonding started
- completion_time (str): ISO timestamp when unbonding completes
- initial_balance (str): Initial unbonding amount
- balance (str): Current unbonding balance

Example:

from akash import AkashClient

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

unbonding = client.staking.get_unbonding_delegations_from_validator("akashvaloper1...")
print(f"Found {len(unbonding)} unbonding delegations from validator")
for unbond in unbonding:
    delegator = unbond['delegator_address']
    print(f"{delegator} unbonding:")
    for entry in unbond['entries']:
        print(f"Amount: {entry['balance']} uakt")

get_historical_info()¶

Get historical info at a specific height.

def get_historical_info(height: int) -> Optional[Dict[str, Any]]

Required parameters:

height (int): Block height

Returns: Optional[Dict] - Historical info with the following structure or None if not found:

header (Dict): Block header information
- version (Dict): Block version
- chain_id (str): Chain identifier
- height (str): Block height
- time (str): Block timestamp
- last_block_id (Dict): Previous block identifier
- last_commit_hash (str): Last commit hash
- data_hash (str): Data hash
- validators_hash (str): Validators hash
- next_validators_hash (str): Next validators hash
- consensus_hash (str): Consensus hash
- app_hash (str): Application hash
- last_results_hash (str): Last results hash
- evidence_hash (str): Evidence hash
- proposer_address (str): Block proposer address
validators_count (int): Number of validators in the set

Example:

from akash import AkashClient

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

history = client.staking.get_historical_info(1000000)
if history:
    print(f"Block height: {history['header']['height']}")
    print(f"Chain ID: {history['header']['chain_id']}")
    print(f"Validators: {history['validators_count']}")

get_pool()¶

Get staking pool information.

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

Returns: Dict - Staking pool information with the following fields:

bonded_tokens (str): Total amount of bonded tokens
not_bonded_tokens (str): Total amount of not bonded tokens

Example:

from akash import AkashClient

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

pool = client.staking.get_pool()
bonded_tokens = int(pool['bonded_tokens'])
not_bonded_tokens = int(pool['not_bonded_tokens'])

print(f"Bonded tokens: {bonded_tokens / 1_000_000:,.0f} AKT")
print(f"Not bonded tokens: {not_bonded_tokens / 1_000_000:,.0f} AKT")