Transaction broadcasting module

The transaction module provides RPC-based transaction broadcasting functionality for the Akash network, including transaction simulation, signing, and confirmation waiting.

When to use this module

Most users should use the high-level module methods instead:

• For sending tokens: client.bank.send()
• For staking: client.staking.delegate()
• For deployments: client.deployment.create_deployment()
• See individual module documentation for all available methods

Use broadcast_transaction_rpc() for:

• Custom message types not covered by module methods
• Advanced transaction composition with multiple message types
• Low-level transaction control and debugging

from akash.tx import broadcast_transaction_rpc, simulate_transaction

result = broadcast_transaction_rpc(client, wallet, messages)

estimated_gas = simulate_transaction(client, wallet, messages, memo, fee_amount)

Functions

broadcast_transaction_rpc()

Broadcast a transaction via RPC with full confirmation support.

def broadcast_transaction_rpc(client, wallet, messages, memo="", fee_amount=None, 
                             gas_limit=None, gas_adjustment=1.2, use_simulation=True,
                             wait_for_confirmation=True, confirmation_timeout=30) -> BroadcastResult

Parameters:

• client (AkashClient): AkashClient instance
• wallet (AkashWallet): AkashWallet instance
• messages (List[dict]): List of message dictionaries with the following fields:
  • @type (str): Message type (e.g., "/cosmos.bank.v1beta1.MsgSend")
  • Additional fields specific to message type
• memo (str): Transaction memo
• fee_amount (str): Fee amount in uakt (default: "5000")
• gas_limit (int): Gas limit override
• gas_adjustment (float): Multiplier for simulated gas (default: 1.2)
• use_simulation (bool): Enable gas simulation (default: True)
• wait_for_confirmation (bool): Wait for confirmation (default: True)
• confirmation_timeout (int): Timeout in seconds (default: 30)

Returns: BroadcastResult - Transaction broadcast result

Example:

from akash import AkashClient, AkashWallet
from akash.tx import broadcast_transaction_rpc

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

messages = [{
    "@type": "/cosmos.bank.v1beta1.MsgSend",
    "from_address": wallet.address,
    "to_address": recipient_address,
    "amount": [{"denom": "uakt", "amount": "1000000"}]
}]

result = broadcast_transaction_rpc(
    client=client,
    wallet=wallet,
    messages=messages,
    memo="",
    fee_amount="5000",
    use_simulation=True
)

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

simulate_transaction()

Simulate a transaction to estimate gas needed.

def simulate_transaction(client, wallet, messages, memo, fee_amount, 
                        default_gas=200000) -> int

Parameters:

• client (AkashClient): AkashClient instance
• wallet (AkashWallet): AkashWallet instance
• messages (List[dict]): List of message dictionaries with the following fields:
  • @type (str): Message type (e.g., "/cosmos.bank.v1beta1.MsgSend")
  • Additional fields specific to message type
• memo (str): Transaction memo
• fee_amount (str): Fee amount for simulation
• default_gas (int): Default gas if simulation fails

Returns: int - Estimated gas needed

Example:

from akash import AkashClient, AkashWallet
from akash.tx import simulate_transaction

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

messages = [{
    "@type": "/cosmos.bank.v1beta1.MsgSend",
    "from_address": wallet.address,
    "to_address": recipient_address,
    "amount": [{"denom": "uakt", "amount": "1000000"}]
}]

estimated_gas = simulate_transaction(
    client, wallet, messages, "Test memo", "5000"
)
print(f"Estimated gas: {estimated_gas}")

BroadcastResult Class

The BroadcastResult class represents the result returned by broadcast_transaction_rpc(). Users receive this object and use its properties and methods to check transaction status and extract information.

Properties

tx_hash

Transaction hash from blockchain.

Returns: str - Transaction hash

code

Transaction result code.

Returns: int - Result code (0 = success, >0 = error)

raw_log

Raw transaction log information.

Returns: str - Raw transaction log

success

Transaction success status.

Returns: bool - True if transaction succeeded

Example:

from akash import AkashClient, AkashWallet
from akash.tx import broadcast_transaction_rpc

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

result = broadcast_transaction_rpc(client, wallet, messages)
if result.success:
    print("Transaction succeeded")

events

Transaction events from blockchain.

Returns: List[Dict] - List of transaction events with the following fields:

• type (str): Event type
• attributes (List[Dict]): Event attributes
  • key (str): Attribute key
  • value (str): Attribute value

Event extraction methods

get_event_attribute()

Extract attribute value from transaction events.

def get_event_attribute(event_type: str, attribute_key: str) -> str

Parameters:

• event_type (str): Event type to search for (e.g., 'akash.v1')
• attribute_key (str): Attribute key to find (e.g., 'dseq')

Returns: str - Attribute value or None if not found

Example:

from akash import AkashClient, AkashWallet
from akash.tx import broadcast_transaction_rpc

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

result = broadcast_transaction_rpc(client, wallet, messages)
dseq = result.get_event_attribute('akash.v1', 'dseq')
print(f"Deployment sequence: {dseq}")

get_dseq()

Extract deployment sequence number from deployment creation transaction.

Returns: int - DSEQ or None if not found

Note: This method parses the akash.v1 event for the dseq attribute and converts it to an integer.

Example:

from akash import AkashClient, AkashWallet

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

result = client.deployment.create_deployment(wallet, groups)
if result.success:
    dseq = result.get_dseq()
    print(f"Created deployment: {dseq}")

get_order_info()

Extract order information from deployment creation transaction.

Returns: dict - Order info dictionary with the following fields, or None if not found:

• dseq (int): Deployment sequence number
• gseq (int): Group sequence number
• oseq (int): Order sequence number

Note: Extracts dseq, gseq, and oseq from akash.v1 events and converts numeric values to integers.

Example:

from akash import AkashClient, AkashWallet

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

result = client.deployment.create_deployment(wallet, groups)
if result.success:
    order_info = result.get_order_info()
    print(f"Order: {order_info['dseq']}/{order_info['gseq']}/{order_info['oseq']}")

get_provider_address()

Extract provider address from provider creation transaction.

Returns: str - Provider address or None if not found

Note: Extracts the owner attribute from akash.v1 events, which represents the provider address in provider creation transactions.

Example:

from akash import AkashClient, AkashWallet

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

result = client.provider.create_provider(wallet, config)
if result.success:
    address = result.get_provider_address()
    print(f"Provider address: {address}")

get_bid_info()

Extract bid information from bid creation transaction.

Returns: dict - Bid info dictionary with the following fields, or None if not found:

• provider (str): Provider address
• dseq (int): Deployment sequence number
• gseq (int): Group sequence number
• oseq (int): Order sequence number

Note: Extracts provider, dseq, gseq, and oseq from akash.v1 events and converts numeric values to integers.

Example:

from akash import AkashClient, AkashWallet

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

result = client.market.create_bid(wallet, order, price)
if result.success:
    bid_info = result.get_bid_info()
    print(f"Bid: {bid_info['provider']} on {bid_info['dseq']}")

Message type mapping

The module supports all Cosmos and Akash message types. Here are the commonly used ones:

Cosmos modules

• Bank: /cosmos.bank.v1beta1.MsgSend
• Staking: /cosmos.staking.v1beta1.MsgDelegate, /cosmos.staking.v1beta1.MsgUndelegate, /cosmos.staking.v1beta1.MsgBeginRedelegate, /cosmos.staking.v1beta1.MsgCreateValidator, /cosmos.staking.v1beta1.MsgEditValidator
• Distribution: /cosmos.distribution.v1beta1.MsgWithdrawDelegatorReward, /cosmos.distribution.v1beta1.MsgWithdrawValidatorCommission, /cosmos.distribution.v1beta1.MsgSetWithdrawAddress, /cosmos.distribution.v1beta1.MsgFundCommunityPool
• Governance: /cosmos.gov.v1beta1.MsgSubmitProposal, /cosmos.gov.v1beta1.MsgVote, /cosmos.gov.v1beta1.MsgDeposit
• Authz: /cosmos.authz.v1beta1.MsgGrant, /cosmos.authz.v1beta1.MsgRevoke, /cosmos.authz.v1beta1.MsgExec
• Feegrant: /cosmos.feegrant.v1beta1.MsgGrantAllowance, /cosmos.feegrant.v1beta1.MsgRevokeAllowance
• Slashing: /cosmos.slashing.v1beta1.MsgUnjail

Akash modules

• Deployment: /akash.deployment.v1beta3.MsgCreateDeployment, /akash.deployment.v1beta3.MsgUpdateDeployment, /akash.deployment.v1beta3.MsgCloseDeployment, /akash.deployment.v1beta3.MsgDepositDeployment, /akash.deployment.v1beta3.MsgCloseGroup, /akash.deployment.v1beta3.MsgPauseGroup, /akash.deployment.v1beta3.MsgStartGroup
• Market: /akash.market.v1beta4.MsgCreateBid, /akash.market.v1beta4.MsgCloseBid, /akash.market.v1beta4.MsgCreateLease, /akash.market.v1beta4.MsgCloseLease, /akash.market.v1beta4.MsgWithdrawLease
• Provider: /akash.provider.v1beta3.MsgCreateProvider, /akash.provider.v1beta3.MsgUpdateProvider
• Cert: /akash.cert.v1beta3.MsgCreateCertificate, /akash.cert.v1beta3.MsgRevokeCertificate
• Audit: /akash.audit.v1beta3.MsgSignProviderAttributes, /akash.audit.v1beta3.MsgDeleteProviderAttributes

IBC modules

• Transfer: /ibc.applications.transfer.v1.MsgTransfer
• Client: /ibc.core.client.v1.MsgCreateClient, /ibc.core.client.v1.MsgUpdateClient

Usage examples

Basic transaction broadcasting

from akash import AkashClient
from akash import AkashWallet
from akash.tx import broadcast_transaction_rpc

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

messages = [{
    "@type": "/cosmos.bank.v1beta1.MsgSend",
    "from_address": wallet.address,
    "to_address": "akash1recipient...",
    "amount": [{"denom": "uakt", "amount": "1000000"}]
}]

result = broadcast_transaction_rpc(
    client=client,
    wallet=wallet,
    messages=messages,
    memo="",
    use_simulation=True
)

print(f"Success: {result.success}")
print(f"TX hash: {result.tx_hash}")

Multiple message transaction

messages = [
    {
        "@type": "/cosmos.staking.v1beta1.MsgDelegate",
        "delegator_address": wallet.address,
        "validator_address": "akashvaloper1...",
        "amount": {"denom": "uakt", "amount": "500000"}
    },
    {
        "@type": "/cosmos.distribution.v1beta1.MsgWithdrawDelegatorReward",
        "delegator_address": wallet.address,
        "validator_address": "akashvaloper1..."
    }
]

result = broadcast_transaction_rpc(
    client=client,
    wallet=wallet,
    messages=messages,
    memo=""
)

Custom gas and fee

result = broadcast_transaction_rpc(
    client=client,
    wallet=wallet,
    messages=messages,
    memo="",
    fee_amount="10000",
    gas_limit=300000,
    use_simulation=False
)

Extract deployment information

from akash import AkashClient, AkashWallet

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

groups = [...]
result = client.deployment.create_deployment(wallet, groups)

if result.success:
    dseq = result.get_dseq()
    print(f"Created deployment {dseq}")

    order_info = result.get_order_info()
    if order_info:
        print(f"Order: {order_info['dseq']}/{order_info['gseq']}/{order_info['oseq']}")

Transaction confirmation handling

result = broadcast_transaction_rpc(
    client=client,
    wallet=wallet,
    messages=messages,
    wait_for_confirmation=True,
    confirmation_timeout=60
)

if result.success:
    print("Transaction confirmed on blockchain")
else:
    if "timeout" in result.raw_log:
        print("Transaction may still be processing")
    else:
        print(f"Transaction failed: {result.raw_log}")