Provider¶

The Provider module manages provider lifecycle including registration, updates, and query operations 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.

ProviderClient class¶

from akash import AkashClient

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

Transactions¶

create_provider()¶

Register a new provider on Akash Network.

def create_provider(wallet, host_uri: str, email: str = "", website: str = "",
                    attributes: List[Dict[str, Any]] = None, memo: str = "",
                    fee_amount: str = None, gas_limit: int = None, gas_adjustment: float = 1.2,
                    use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (Wallet): Provider operator wallet instance
host_uri (str): Provider host URI

Optional parameters:

email (str): Provider email (default: "")
website (str): Provider website (default: "")
attributes (List[Dict]): Provider attributes (default: []) with the following fields:
- key (str, required): Attribute key
- value (str, required): Attribute value
memo (str): Transaction memo (default: "")
fee_amount (str): Transaction fee amount (default: auto-calculated)
gas_limit (int): Gas limit (default: auto-estimated)
gas_adjustment (float): Gas estimation multiplier (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult 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("provider mnemonic here")

result = client.provider.create_provider(
    wallet=wallet,
    host_uri="https://provider.example.com",
    email="admin@provider.com",
    website="https://provider.com",
    attributes=[
        {"key": "location-region", "value": "West Coast"},
        {"key": "country", "value": "United States"},
        {"key": "tier", "value": "dcloud"}
    ]
)

if result.success:
    print(f"Provider created: {result.tx_hash}")
else:
    print(f"Provider creation failed: {result.raw_log}")

update_provider()¶

Update existing provider information.

def update_provider(wallet, host_uri: str, email: str = "", website: str = "",
                    attributes: List[Dict[str, Any]] = None, memo: str = "",
                    fee_amount: str = None, gas_limit: int = None, gas_adjustment: float = 1.2,
                    use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (Wallet): Provider operator wallet instance
host_uri (str): Provider host URI

Optional parameters:

email (str): Provider email (default: "")
website (str): Provider website (default: "")
attributes (List[Dict]): Provider attributes (default: []) with the following fields:
- key (str, required): Attribute key
- value (str, required): Attribute value
memo (str): Transaction memo (default: "")
fee_amount (str): Transaction fee amount (default: auto-calculated)
gas_limit (int): Gas limit (default: auto-estimated)
gas_adjustment (float): Gas estimation multiplier (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult 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("provider mnemonic here")

result = client.provider.update_provider(
    wallet=wallet,
    host_uri="https://newhost.provider.com",
    email="newemail@provider.com",
    attributes=[
        {"key": "location-region", "value": "East Coast"},
        {"key": "country", "value": "United States"},
        {"key": "tier", "value": "premium"}
    ]
)

Queries¶

get_providers()¶

List all registered providers via RPC query.

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

Optional parameters:

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

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

owner (str): Provider owner address
host_uri (str): Provider host URI
attributes (List[Dict]): Provider attributes
- key (str): Attribute key
- value (str): Attribute value
info (Dict): Provider info with email and website
- email (str): Provider contact email
- website (str): Provider website URL

Example:

from akash import AkashClient

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

providers = client.provider.get_providers(limit=10)
print(f"Found {len(providers)} providers")

for provider in providers:
    print(f"Provider: {provider['owner']}")
    print(f"Host: {provider['host_uri']}")
    print(f"Email: {provider['info']['email']}")

get_provider()¶

Get specific provider information using proper ABCI query.

def get_provider(owner_address: str) -> Dict[str, Any]

Required parameters:

owner_address (str): Provider owner address

Returns: Dict - Provider information dictionary with the following fields:

owner (str): Provider owner address
host_uri (str): Provider host URI
attributes (List[Dict]): Provider attributes
- key (str): Attribute key
- value (str): Attribute value
info (Dict): Provider info
- email (str): Provider email
- website (str): Provider website

Example:

from akash import AkashClient

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

provider = client.provider.get_provider("akash1...")

if provider:
    print(f"Provider: {provider['owner']}")
    print(f"Host: {provider['host_uri']}")
    print("Attributes:")
    for attr in provider['attributes']:
        print(f"{attr['key']}: {attr['value']}")

get_provider_leases()¶

Get active leases for a provider.

def get_provider_leases(owner_address: str) -> List[Dict[str, Any]]

Required parameters:

owner_address (str): Provider owner address

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

lease_id (str): Formatted lease identifier
tenant (str): Lease tenant address
provider (str): Lease provider address
state (str): Lease state
price (Dict): Lease price information
- amount (str): Price amount
- denom (str): Price denomination
created_at (int): Lease creation timestamp
closed_on (int): Lease close timestamp
escrow_payment (Dict): Escrow payment information
- account_id (Dict): Payment account identifier
  - scope (str): Account scope
  - xid (str): Account transaction ID
- payment_id (str): Payment ID
- owner (str): Payment owner address
- state (str): Payment state
- rate (Dict): Payment rate
  - denom (str): Rate denomination
  - amount (str): Rate amount
- balance (Dict): Current balance
  - denom (str): Balance denomination
  - amount (str): Balance amount
- withdrawn (Dict): Amount withdrawn
  - denom (str): Withdrawn denomination
  - amount (str): Withdrawn amount

Example:

from akash import AkashClient

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

leases = client.provider.get_provider_leases("akash1provider...")
print(f"Provider has {len(leases)} active leases")

for lease in leases:
    print(f"Lease: {lease['lease_id']}")
    print(f"Tenant: {lease['tenant']}")
    print(f"Price: {lease['price']['amount']} {lease['price']['denom']}")

get_provider_attributes()¶

Get provider attributes.

def get_provider_attributes(owner_address: str) -> List[Dict[str, str]]

Required parameters:

owner_address (str): Provider owner address

Returns: List[Dict] - List of attribute dictionaries with the following fields:

key (str): Attribute key
value (str): Attribute value

Example:

from akash import AkashClient

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

attributes = client.provider.get_provider_attributes("akash1provider...")

for attr in attributes:
    print(f"{attr['key']}: {attr['value']}")

query_providers_by_attributes()¶

Query providers that have specific attributes.

def query_providers_by_attributes(required_attributes: List[Dict[str, str]]) -> List[Dict[str, Any]]

Required parameters:

required_attributes (List[Dict]): List of required attribute dictionaries
- key (str): Attribute key
- value (str): Attribute value

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

owner (str): Provider owner address
host_uri (str): Provider host URI
attributes (List[Dict]): Provider attributes
- key (str): Attribute key
- value (str): Attribute value
info (Dict): Provider info
- email (str): Provider email
- website (str): Provider website

Example:

from akash import AkashClient

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

storage_providers = client.provider.query_providers_by_attributes([
    {"key": "host", "value": "akash"},
    {"key": "tier", "value": "community"}
])

print(f"Found {len(storage_providers)} providers with specified attributes")

validate_provider_config()¶

Validate provider configuration before registration.

def validate_provider_config(provider_info: Dict[str, Any]) -> Dict[str, Any]

Required parameters:

provider_info (Dict): Provider configuration to validate with the following fields:
- host_uri (str): Provider host URI
- email (str): Provider email
- website (str): Provider website
- attributes (List[Dict]): Provider attributes
  - key (str): Attribute key
  - value (str): Attribute value

Returns: Dict - Validation result with the following fields:

valid (bool): Whether configuration is valid
errors (List[str]): List of validation errors
warnings (List[str]): List of validation warnings

Example:

from akash import AkashClient

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

provider_config = {
    "host_uri": "https://provider.com",
    "email": "admin@provider.com",
    "attributes": [
        {"key": "location-region", "value": "West Coast"},
        {"key": "country", "value": "United States"}
    ]
}

validation = client.provider.validate_provider_config(provider_config)

if validation["valid"]:
    print("Configuration is valid")
    if validation["warnings"]:
        for warning in validation["warnings"]:
            print(f"Warning: {warning}")
else:
    print("Configuration errors:")
    for error in validation["errors"]:
        print(f"- {error}")

get_provider_status()¶

Get detailed provider status via off-chain gRPC query.

def get_provider_status(owner_address: str) -> Dict[str, Any]

Required parameters:

owner_address (str): Provider owner address

Returns: Dict - Provider status information with the following fields:

inventory (Dict): Available/pending/allocated resources per node
- cluster (Dict): Cluster inventory
  - nodes (List[Dict]): List of node resources
    - cpu (Dict): CPU resources
      - quantity (Dict): CPU quantity
        
        value (int): CPU value in millicores
    - memory (Dict): Memory resources
      - quantity (Dict): Memory quantity
        
        value (int): Memory value in bytes
    - storage (Dict): Storage resources
      - quantity (Dict): Storage quantity
        
        value (int): Storage value in bytes
available_leases (int): Number of available leases
orders (Dict): Map of order states to counts
- open (int): Number of open orders
- matched (int): Number of matched orders
- lost (int): Number of lost orders
bids (Dict): Map of bid states to counts
- open (int): Number of open bids
- matched (int): Number of matched bids
- lost (int): Number of lost bids
- closed (int): Number of closed bids
leases (Dict): Map of lease states to counts
- active (int): Number of active leases
- available (int): Number of available lease slots
- pending (int): Number of pending leases
- closed (int): Number of closed leases
cluster (Dict): Cluster status information
- leases (Dict): Lease information
  - active (int): Active leases count
  - available (int): Available lease slots
- inventory (Dict): Cluster inventory
  - active (List[Dict]): Active resources
  - pending (List[Dict]): Pending resources
  - available (List[Dict]): Available resources
bid_engine (Dict): Bid engine status
- orders (int): Number of orders being processed
manifest (Dict): Manifest processing status
- deployments (int): Number of deployments with manifests
active_leases (int): Number of active leases

Example:

from akash import AkashClient

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

status = client.provider.get_provider_status("akash1provider...")
print(f"Active leases: {status.get('active_leases', 0)}")
print(f"Available leases: {status.get('available_leases', 0)}")
print(f"Cluster status: {status.get('cluster', {})}")

get_provider_version()¶

Get provider version information via REST API.

def get_provider_version(owner_address: str, timeout: int = 5000) -> Optional[Dict[str, str]]

Required parameters:

owner_address (str): Provider owner address

Optional parameters:

timeout (int): Timeout in milliseconds (default: 5000)

Returns: Dict or None - Provider version information with the following fields, or None if query fails:

version (str): Provider version string
cosmos_sdk_version (str): Cosmos SDK version string

Example:

from akash import AkashClient

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

version = client.provider.get_provider_version("akash1provider...")

if version:
    print(f"Provider version: {version.get('version')}")
    print(f"Cosmos SDK version: {version.get('cosmos_sdk_version')}")
else:
    print("Failed to retrieve provider version")

Performance Note

The filtering functions below (get_providers_by_region() and get_providers_by_capabilities()) fetch all providers from the chain across all pages before applying filters. Additionally, both functions support an include_status parameter which makes individual gRPC calls to each matching provider. These operations can take a significant amount of time, especially on mainnet with many providers.

get_providers_by_region()¶

Get providers filtered by region attribute.

def get_providers_by_region(region: str, include_status: bool = False) -> List[Dict[str, Any]]

Required parameters:

region (str): Region to filter by (e.g., 'United States', 'Europe', 'Asia', 'East Coast') Performs case-insensitive substring matching on region-related attributes

Optional parameters:

include_status (bool): Whether to include off-chain status for each provider (default: False)

Returns: List[Dict] - List of providers in the specified region with the following fields:

owner (str): Provider owner address
host_uri (str): Provider host URI
attributes (List[Dict]): Provider attributes
- key (str): Attribute key
- value (str): Attribute value
info (Dict): Provider info
- email (str): Provider email
- website (str): Provider website

Example:

from akash import AkashClient

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

us_providers = client.provider.get_providers_by_region("United States")
print(f"Found {len(us_providers)} providers in United States")

for provider in us_providers:
    print(f"Provider: {provider['owner']}")
    print(f"Host: {provider['host_uri']}")

get_providers_by_capabilities()¶

Get providers filtered by required capabilities.

def get_providers_by_capabilities(capabilities: List[str], include_status: bool = False) -> List[Dict[str, Any]]

Required parameters:

capabilities (List[str]): List of required capabilities (e.g., ['gpu', 'nvidia', '4090'])
Performs case-insensitive substring matching on provider capability attributes
Checks attributes with feat-* prefix (e.g., feat-persistent-storage, feat-endpoint-ip)
Checks attributes with capabilities/* prefix (e.g., capabilities/gpu/vendor/nvidia/model/4090)
All specified capabilities must be present (AND logic)

Optional parameters:

include_status (bool): Whether to include off-chain status for each provider (default: False)

Returns: List[Dict] - List of providers that have all specified capabilities with the following fields:

owner (str): Provider owner address
host_uri (str): Provider host URI
attributes (List[Dict]): Provider attributes
- key (str): Attribute key
- value (str): Attribute value
info (Dict): Provider info
- email (str): Provider email
- website (str): Provider website
matched_capabilities (List[str]): List of matching capability attributes

Example:

from akash import AkashClient

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

# Find all GPU providers
gpu_providers = client.provider.get_providers_by_capabilities(["gpu"])
print(f"Found {len(gpu_providers)} GPU providers")

for provider in gpu_providers:
    print(f"Provider: {provider['owner']}")
    print(f"Host: {provider['host_uri']}")
    print(f"Matched capabilities: {provider.get('matched_capabilities', [])}")

nvidia_providers = client.provider.get_providers_by_capabilities(["nvidia"])
print(f"\nFound {len(nvidia_providers)} NVIDIA GPU providers")

rtx4090_providers = client.provider.get_providers_by_capabilities(["4090"])
print(f"Found {len(rtx4090_providers)} providers with RTX 4090")

gpu_storage_providers = client.provider.get_providers_by_capabilities(
    ["gpu", "persistent-storage"]
)
print(f"\nFound {len(gpu_storage_providers)} providers with GPU and persistent storage")

storage_providers = client.provider.get_providers_by_capabilities(
    ["persistent-storage"]
)
print(f"Found {len(storage_providers)} providers with persistent storage")

endpoint_providers = client.provider.get_providers_by_capabilities(
    ["endpoint"]
)
print(f"Found {len(endpoint_providers)} providers with custom endpoint support")

validate_provider_endpoint()¶

Validate provider endpoint connectivity and DNS resolution.

def validate_provider_endpoint(provider_address: str) -> bool

Required parameters:

provider_address (str): Provider address to validate

Returns: bool - True if provider endpoint is accessible, False otherwise

Example:

from akash import AkashClient

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

provider_address = "akash1provider123..."
is_valid = client.provider.validate_provider_endpoint(provider_address)

if is_valid:
    print(f"Provider {provider_address} has a valid endpoint")
else:
    print(f"Provider {provider_address} endpoint is not accessible")

filter_valid_providers()¶

Filter provider addresses to only include those with valid endpoints.

def filter_valid_providers(provider_addresses: List[str]) -> List[str]

Required parameters:

provider_addresses (List[str]): List of provider addresses to filter

Returns: List[str] - List of provider addresses with valid endpoints

Example:

from akash import AkashClient

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

all_providers = ["akash1provider1...", "akash1provider2...", "akash1provider3..."]
valid_providers = client.provider.filter_valid_providers(all_providers)

print(f"Filtered {len(all_providers)} providers down to {len(valid_providers)} valid providers")
for provider in valid_providers:
    print(f"Valid provider: {provider}")