IBC module¶

The IBC (Inter-Blockchain Communication) module provides cross-chain transfer, client management, and query operations for the Akash Network.

Transaction parameters

All transaction functions support optional parameters like fee_amount, gas_limit, and gas_adjustment. See Transaction parameters for details.

IBCClient class¶

from akash import AkashClient

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

# Access IBC operations
ibc = client.ibc

Transactions¶

transfer()¶

Execute an IBC token transfer to another blockchain.

def transfer(wallet, source_channel: str, token_amount: str, token_denom: str, receiver: str,
            source_port: str = "transfer", timeout_height: Optional[Dict[str, int]] = None,
            timeout_timestamp: Optional[int] = None, memo: str = "",
            fee_amount: Optional[str] = None, gas_limit: Optional[int] = None,
            gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet for signing the transaction
source_channel (str): Source channel ID (e.g., "channel-17")
token_amount (str): Amount to transfer in base units
token_denom (str): Token denomination (e.g., "uakt")
receiver (str): Recipient address on destination chain

Optional parameters:

source_port (str): Source port (default: "transfer")
timeout_height (Dict): Timeout height with revision_number and revision_height
- revision_number (str): Revision number
- revision_height (str): Revision height
timeout_timestamp (int): Timeout timestamp in nanoseconds (default: 10 minutes)
memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in base units
gas_limit (int): Gas limit override
gas_adjustment (float): Gas estimation multiplier (default: 1.2)
use_simulation (bool): Use 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("your mnemonic here")

result = client.ibc.transfer(
    wallet=wallet,
    source_channel="channel-17",  # Akash -> Cosmos Hub
    token_amount="1000000",  # 1 AKT
    token_denom="uakt",
    receiver="cosmos1abc...",
    memo=""
)

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

create_client()¶

Create a new IBC client for a target chain.

def create_client(wallet, target_chain_id: str, target_rpc_url: str,
                           trusting_period_seconds: int = 1209600,
                           unbonding_period_seconds: int = 1814400,
                           max_clock_drift_seconds: int = 600,
                           memo: str = "", fee_amount: Optional[str] = None,
                           gas_limit: Optional[int] = None, gas_adjustment: float = 1.2,
                           use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet for signing
target_chain_id (str): Target chain ID (e.g., "cosmoshub-4")
target_rpc_url (str): Target chain RPC URL

Optional parameters:

trusting_period_seconds (int): Trusting period (default: 14 days)
unbonding_period_seconds (int): Unbonding period (default: 21 days)
max_clock_drift_seconds (int): Max clock drift (default: 10 minutes)
memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount
gas_limit (int): Gas limit
gas_adjustment (float): Gas multiplier (default: 1.2)
use_simulation (bool): Use 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("your mnemonic here")

result = client.ibc.create_client(
    wallet=wallet,
    target_chain_id="osmosis-1",
    target_rpc_url="https://rpc.osmosis.zone:443"
)

if result.success:
    client_id = result.get_event_attribute('create_client', 'client_id')
    print(f"Created client: {client_id}")

update_client()¶

Update an existing IBC client with the latest block header.

def update_client(wallet, client_id: str, target_rpc_url: str, memo: str = "",
                 fee_amount: Optional[str] = None, gas_limit: Optional[int] = None,
                 gas_adjustment: float = 1.2, use_simulation: bool = True,
                 akash_api: str = "https://akash-api.polkachu.com") -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet for signing
client_id (str): Client ID to update (e.g., "07-tendermint-53")
target_rpc_url (str): Target chain RPC URL

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount
gas_limit (int): Gas limit
gas_adjustment (float): Gas multiplier (default: 1.2)
use_simulation (bool): Use simulation (default: True)
akash_api (str): Akash API endpoint for queries (default: "https://akash-api.polkachu.com")

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("your mnemonic here")

result = client.ibc.update_client(
    wallet=wallet,
    client_id="07-tendermint-53",
    target_rpc_url="https://cosmos-rpc.polkachu.com:443",
    gas_limit=1000000,
    fee_amount="25000"
)

if result.success:
    print(f"Client updated: {result.tx_hash}")

Queries¶

get_client_state()¶

Query the state of a specific IBC client.

def get_client_state(client_id: str) -> Optional[Dict[str, Any]]

Required parameters:

client_id (str): Client identifier (e.g., "07-tendermint-53")

Returns: Dict or None - Client state with the following structure:

client_state (Dict): Client state information
- @type (str): Type identifier
- chain_id (str): Target chain identifier
- trust_level (Dict): Trust level configuration
  - numerator (str): Trust numerator
  - denominator (str): Trust denominator
- trusting_period (str): Trusting period duration
- unbonding_period (str): Unbonding period duration
- max_clock_drift (str): Maximum clock drift allowed
- latest_height (Dict): Latest known height
  - revision_number (str): Revision number
  - revision_height (str): Revision height
proof (str): Merkle proof for the client state
proof_height (Dict): Height at which proof was generated
- revision_number (str): Revision number
- revision_height (str): Revision height

Example:

from akash import AkashClient

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

client_state = client.ibc.get_client_state("07-tendermint-53")
if client_state:
    print(f"Client state found for {client_state['client_state']['chain_id']}")

get_client_status()¶

Query the status of an IBC client.

def get_client_status(client_id: str) -> str

Required parameters:

client_id (str): Client identifier

Returns: str - Status: "Active", "Expired", "Frozen", or "Unknown"

Example:

from akash import AkashClient

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

status = client.ibc.get_client_status("07-tendermint-53")
print(f"Client status: {status}")

get_client_states()¶

Query all IBC client states with pagination.

def get_client_states(limit: Optional[int] = None, next_key: Optional[str] = None,
                       timeout_minutes: int = 3, rest_endpoint: str = "https://rest.cosmos.directory/akash") -> Dict[str, Any]

Optional parameters:

limit (int): Maximum results per page (default: 5000)
next_key (str): Pagination key for next page
timeout_minutes (int): Query timeout (default: 3)
rest_endpoint (str): REST API endpoint (default: "https://rest.cosmos.directory/akash")

Returns: Dict[str, Any] - Response with the following structure:

client_states (List[Dict]): List of client states
- client_id (str): Client identifier
- client_state (Dict): Client state data
  - @type (str): Type identifier
  - chain_id (str): Target chain identifier
  - trust_level (Dict): Trust level configuration
    - numerator (str): Trust numerator
    - denominator (str): Trust denominator
  - trusting_period (str): Trusting period duration
  - unbonding_period (str): Unbonding period duration
  - latest_height (Dict): Latest known height
    - revision_number (str): Revision number
    - revision_height (str): Revision height
pagination (Dict): Pagination information
- next_key (str): Key for next page
- total (str): Total number of items

Example:

from akash import AkashClient

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

result = client.ibc.get_client_states(limit=10)
clients = result['client_states']

for client_info in clients:
    print(f"Client: {client_info['client_id']}")

get_connection()¶

Query a specific IBC connection.

def get_connection(connection_id: str) -> Optional[Dict[str, Any]]

Required parameters:

connection_id (str): Connection identifier (e.g., "connection-29")

Returns: Dict or None - Connection with the following structure:

connection (Dict): Connection information
- client_id (str): Associated client identifier
- versions (List[Dict]): Supported protocol versions
  - identifier (str): Version identifier
  - features (List[str]): Supported features
- state (int): Connection state (0=uninitialized, 1=init, 2=tryopen, 3=open)
- counterparty (Dict): Counterparty connection details
  - client_id (str): Counterparty client ID
  - connection_id (str): Counterparty connection ID
  - prefix (Dict): Key prefix for proofs
    - key_prefix (str): Prefix for merkle proof keys
- delay_period (str): Delay period for packet verification
proof (str): Merkle proof for the connection
proof_height (Dict): Height at which proof was generated
- revision_number (str): Revision number
- revision_height (str): Revision height

Example:

from akash import AkashClient

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

conn = client.ibc.get_connection("connection-29")
if conn:
    print(f"Connection uses client: {conn['connection']['client_id']}")

get_connections()¶

Query all IBC connections with pagination.

def get_connections(limit: Optional[int] = None, next_key: Optional[str] = None,
                     timeout_minutes: int = 3, rest_endpoint: str = "https://rest.cosmos.directory/akash") -> Dict[str, Any]

Optional parameters:

limit (int): Maximum results per page (default: 5000)
next_key (str): Pagination key
timeout_minutes (int): Query timeout (default: 3)
rest_endpoint (str): REST API endpoint (default: "https://rest.cosmos.directory/akash")

Returns: Dict[str, Any] - Response with the following structure:

connections (List[Dict]): List of connections
- id (str): Connection identifier
- client_id (str): Associated client identifier
- versions (List[Dict]): Supported protocol versions
- state (int): Connection state (0=uninitialized, 1=init, 2=tryopen, 3=open)
- counterparty (Dict): Counterparty connection details
  - client_id (str): Counterparty client ID
  - connection_id (str): Counterparty connection ID
  - prefix (Dict): Key prefix for proofs
    - key_prefix (str): Prefix for merkle proof keys
- delay_period (str): Delay period for packet verification
pagination (Dict): Pagination information
- next_key (str): Key for next page
- total (str): Total number of items
height (Dict): Query height information
- revision_number (str): Revision number
- revision_height (str): Revision height

Example:

from akash import AkashClient

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

result = client.ibc.get_connections(limit=20)
connections = result['connections']

for conn in connections:
    if conn['state'] == 3:
        print(f"Connection {conn['id']}: {conn['client_id']}")

get_channel()¶

Query a specific IBC channel.

def get_channel(port_id: str, channel_id: str) -> Optional[Dict[str, Any]]

Required parameters:

port_id (str): Port identifier (e.g., "transfer")
channel_id (str): Channel identifier (e.g., "channel-17")

Returns: Dict or None - Channel with the following structure:

channel (Dict): Channel information
- state (int): Channel state (0=uninitialized, 1=init, 2=tryopen, 3=open, 4=closed)
- ordering (int): Channel ordering (0=none, 1=unordered, 2=ordered)
- counterparty (Dict): Counterparty channel details
  - port_id (str): Counterparty port identifier
  - channel_id (str): Counterparty channel identifier
- connection_hops (List[str]): List of connection IDs in the path
- version (str): Channel version
proof (str): Merkle proof for the channel
proof_height (Dict): Height at which proof was generated
- revision_number (str): Revision number
- revision_height (str): Revision height

Example:

from akash import AkashClient

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

channel = client.ibc.get_channel("transfer", "channel-17")
if channel:
    print(f"Channel state: {channel['channel']['state']}")
    print(f"Connection hops: {channel['channel']['connection_hops']}")

get_channels()¶

Query all IBC channels with pagination.

def get_channels(limit: Optional[int] = None, next_key: Optional[str] = None,
                  timeout_minutes: int = 3, rest_endpoint: str = "https://rest.cosmos.directory/akash") -> Dict[str, Any]

Optional parameters:

limit (int): Maximum results per page (default: 5000)
next_key (str): Pagination key
timeout_minutes (int): Query timeout (default: 3)
rest_endpoint (str): REST API endpoint (default: "https://rest.cosmos.directory/akash")

Returns: Dict[str, Any] - Response with the following structure:

channels (List[Dict]): List of channels
- state (int): Channel state (0=uninitialized, 1=init, 2=tryopen, 3=open, 4=closed)
- ordering (int): Channel ordering (0=none, 1=unordered, 2=ordered)
- counterparty (Dict): Counterparty channel details
  - port_id (str): Counterparty port identifier
  - channel_id (str): Counterparty channel identifier
- connection_hops (List[str]): List of connection IDs in the path
- version (str): Channel version
- port_id (str): Port identifier
- channel_id (str): Channel identifier
pagination (Dict): Pagination information
- next_key (str): Key for next page
- total (str): Total number of items
height (Dict): Query height information
- revision_number (str): Revision number
- revision_height (str): Revision height

Example:

from akash import AkashClient

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

result = client.ibc.get_channels(limit=50)
channels = result['channels']

transfer_channels = [
    ch for ch in channels 
    if ch['port_id'] == 'transfer' and ch['state'] == 3
]

print(f"Found {len(transfer_channels)} active transfer channels")

get_denom_trace()¶

Query IBC denomination trace information.

def get_denom_trace(hash: str) -> Optional[Dict[str, Any]]

Required parameters:

hash (str): IBC denomination hash (without 'ibc/' prefix)

Returns: Dict or None - Denomination trace with the following structure:

denom_trace (Dict): Trace information
- path (str): Transfer path (e.g., "transfer/channel-17")
- base_denom (str): Original denomination (e.g., "uatom")
proof (str): Merkle proof for the denomination trace
proof_height (Dict): Height at which proof was generated
- revision_number (str): Revision number
- revision_height (str): Revision height

Example:

from akash import AkashClient

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

hash = "27394FB092D2ECCD56123C74F36E4C1F926001CEADA9CA97EA622B25F41E5EB2"
trace = client.ibc.get_denom_trace(hash)

if trace:
    info = trace['denom_trace']
    print(f"Base denom: {info['base_denom']}")  # uatom
    print(f"Path: {info['path']}")  # transfer/channel-0

get_transfer_params()¶

Query IBC transfer module parameters.

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

Returns: Dict with transfer module configuration:

send_enabled (bool): Whether IBC transfers can be sent from this chain
receive_enabled (bool): Whether IBC transfers can be received on this chain

Example:

from akash import AkashClient

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

params = client.ibc.get_transfer_params()
print(f"Send enabled: {params['send_enabled']}")
print(f"Receive enabled: {params['receive_enabled']}")

trace_ibc_token()¶

Resolve IBC token denomination to its original base denomination and transfer path.

def trace_ibc_token(ibc_denom: str) -> Optional[Dict[str, Any]]

Required parameters:

ibc_denom (str): Token denomination ("ibc/ABC123..." or "uakt")

Returns: Dict with:

is_ibc (bool): Whether it's an IBC token
base_denom (str): Original denomination
path (str): Transfer path (IBC tokens only)

Example:

from akash import AkashClient

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

# Trace IBC ATOM
trace = client.ibc.trace_ibc_token("ibc/27394FB092D2ECCD56123C74F36E4C1F926001CEADA9CA97EA622B25F41E5EB2")
print(f"{trace['base_denom']} via {trace['path']}")  # "uatom via transfer/channel-17"

# Trace native token
trace = client.ibc.trace_ibc_token("uakt")
print(f"Native: {trace['base_denom']}")  # "Native: uakt"

find_active_clients_for_chain()¶

Find active IBC clients for a specific chain.

def find_active_clients_for_chain(chain_id: str, max_results: int = 10) -> List[str]

Required parameters:

chain_id (str): Target chain ID (e.g., "cosmoshub-4", "osmosis-1")

Optional parameters:

max_results (int): Maximum results to return (default: 10)

Returns: List[str] - List of active client IDs

Example:

from akash import AkashClient

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

active_clients = client.ibc.find_active_clients_for_chain("cosmoshub-4")
print(f"Active Cosmos clients: {active_clients}")

find_active_channels_for_chain()¶

Find active IBC transfer channels for a specific chain.

def find_active_channels_for_chain(chain_id: str, max_results: int = 10) -> List[Dict[str, Any]]

Required parameters:

chain_id (str): Target chain ID

Optional parameters:

max_results (int): Maximum results (default: 10)

Returns: List[Dict[str, Any]] - List of active channels with the following structure:

channel_id (str): Channel identifier (e.g., "channel-17")
port_id (str): Port identifier (usually "transfer")
connection_id (str): Associated connection identifier
state (int): Channel state (3=open)
counterparty (Dict): Counterparty channel details
- port_id (str): Counterparty port identifier
- channel_id (str): Counterparty channel identifier

Example:

from akash import AkashClient

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

channels = client.ibc.find_active_channels_for_chain("cosmoshub-4")

if channels:
    channel = channels[0]
    print(f"Use channel {channel['channel_id']} for Cosmos transfers")
else:
    print("No active channels found")