Working with providers¶
Complete examples for discovering, evaluating, and interacting with providers on the Akash Network.
Overview¶
Providers are the compute hosts on Akash Network that execute your deployments. Understanding how to find, evaluate, and select the right providers is crucial for successful deployments.
Provider discovery process:
graph LR
A[Discover Providers] --> B[Filter by Capabilities]
B --> C[Check Availability]Provider Evaluation:
graph LR
D[Evaluate Performance] --> E[Score & Rank]
E --> F[Select Best Provider]Provider Management:
graph LR
G[Monitor Performance] --> H[Track Metrics]
H --> I[Switch if Needed]Provider discovery¶
List all providers¶
from akash import AkashClient
def list_all_providers():
"""List all providers on the network with pagination"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
all_providers = []
offset = 0
limit = 100
while True:
providers = client.provider.get_providers(limit=limit, offset=offset)
if not providers:
break
all_providers.extend(providers)
if len(providers) < limit:
break
offset += limit
print(f"Found {len(all_providers)} total providers (showing first 10)")
print("-" * 50)
for i, provider in enumerate(all_providers[:10], 1):
print(f"{i}. {provider['owner']}")
print(f"Host: {provider.get('host_uri', 'N/A')}")
print(f"Attributes: {len(provider.get('attributes', []))}")
info = provider.get('info', {})
if info and info.get('website'):
print(f"Website: {info['website']}")
print()
return all_providers
providers = list_all_providers()
API reference
See Provider API for provider discovery, filtering, and management operations
Paginated provider listing¶
from akash import AkashClient
def paginated_provider_listing(page_size=20):
"""List providers with pagination for large datasets"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
offset = 0
page = 1
all_providers = []
while True:
providers = client.provider.get_providers(
limit=page_size,
offset=offset
)
if not providers:
print("No more providers found")
break
all_providers.extend(providers)
print(f"\nPage {page} ({len(providers)} providers)")
print("=" * 50)
for provider in providers:
print(f"• {provider['owner'][:20]}... - {provider.get('host_uri', 'N/A')}")
if len(providers) < page_size:
break
offset += page_size
page += 1
print(f"\nTotal providers found: {len(all_providers)}")
return all_providers
providers = paginated_provider_listing()
Provider filtering¶
Filter by region¶
from akash import AkashClient
client = AkashClient("https://akash-rpc.polkachu.com:443")
us_providers = client.provider.get_providers_by_region("United States")
europe_providers = client.provider.get_providers_by_region("Europe")
asia_providers = client.provider.get_providers_by_region("Asia")
print(f"US providers: {len(us_providers)}")
print(f"Europe providers: {len(europe_providers)}")
print(f"Asia providers: {len(asia_providers)}")
print("\nUS Provider details:")
for provider in us_providers[:5]:
print(f"• {provider['owner']}")
print(f" Host: {provider.get('host_uri', 'N/A')}")
for attr in provider.get('attributes', []):
if attr['key'] in ['location-region', 'country', 'city']:
print(f" {attr['key']}: {attr['value']}")
print()
Filter by capabilities¶
from akash import AkashClient
client = AkashClient("https://akash-rpc.polkachu.com:443")
storage_providers = client.provider.get_providers_by_capabilities(["persistent-storage"])
ip_providers = client.provider.get_providers_by_capabilities(["endpoint-ip"])
full_featured = client.provider.get_providers_by_capabilities(["persistent-storage", "endpoint-ip"])
print(f"Storage providers: {len(storage_providers)}")
print(f"IP endpoint providers: {len(ip_providers)}")
print(f"Full-featured providers: {len(full_featured)}")
print("\nStorage provider capabilities:")
for provider in storage_providers[:5]:
print(f"• {provider['owner']}")
print(f" Host: {provider.get('host_uri', 'N/A')}")
matched_caps = provider.get('matched_capabilities', [])
print(f" Capabilities: {', '.join(matched_caps[:5])}")
print()
Advanced multi-criteria filtering¶
from akash import AkashClient
client = AkashClient("https://akash-rpc.polkachu.com:443")
region_providers = client.provider.get_providers_by_region("United States")
capable_providers = []
for provider in region_providers:
provider_caps = set()
for attr in provider.get('attributes', []):
if attr['key'].startswith("feat-") and attr['value'].lower() == "true":
provider_caps.add(attr['key'])
elif attr['key'].startswith("capabilities/") and attr['value'].lower() == "true":
provider_caps.add(attr['key'])
if any("persistent-storage" in cap for cap in provider_caps) and \
any("endpoint-ip" in cap for cap in provider_caps):
capable_providers.append(provider)
print(f"US providers with storage and IP: {len(capable_providers)}")
for provider in capable_providers[:5]:
print(f"• {provider['owner']}")
print(f" Host: {provider.get('host_uri', 'N/A')}")
Provider evaluation¶
Get provider details¶
from akash import AkashClient
def get_provider_details(provider_address):
"""Get detailed information about a specific provider"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
try:
provider = client.provider.get_provider(provider_address)
print(f"Provider details: {provider_address}")
print("=" * 60)
print(f"Owner: {provider['owner']}")
print(f"Host URI: {provider.get('host_uri', 'N/A')}")
info = provider.get('info', {})
if info:
print(f"Email: {info.get('email', 'N/A')}")
print(f"Website: {info.get('website', 'N/A')}")
print(f"\nAttributes ({len(provider.get('attributes', []))}):")
for attr in provider.get('attributes', []):
print(f"{attr['key']}: {attr['value']}")
return provider
except Exception as e:
print(f"Error getting provider details: {e}")
return None
provider_details = get_provider_details("akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63")
Check provider status¶
from akash import AkashClient
def check_provider_status(provider_address):
"""Check the current status and availability of a provider"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
try:
provider = client.provider.get_provider(provider_address)
leases = client.provider.get_provider_leases(provider_address)
print(f"Provider status: {provider_address}")
print("=" * 60)
print(f"Host URI: {provider.get('host_uri', 'N/A')}")
print(f"Active leases: {len(leases)}")
capabilities = []
resources = {}
for attr in provider.get('attributes', []):
key = attr['key']
value = attr['value']
if key.startswith("feat-") and value.lower() == "true":
capabilities.append(key)
elif key.startswith("capabilities/") and value.lower() == "true":
capabilities.append(key)
elif "cpu" in key or "memory" in key or "storage" in key or "gpu" in key:
resources[key] = value
print(f"Capabilities: {', '.join(capabilities) if capabilities else 'None'}")
print("\nResource availability:")
for key, value in resources.items():
print(f"{key}: {value}")
return {
"provider": provider,
"leases": leases,
"capabilities": capabilities,
"resources": resources
}
except Exception as e:
print(f"Error checking provider status: {e}")
return None
status = check_provider_status("akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63")
Check provider real-time status¶
from akash import AkashClient
def check_provider_realtime_status(provider_address):
"""Get real-time provider status via off-chain gRPC query"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
try:
status = client.provider.get_provider_status(provider_address)
print(f"Real-time provider status: {provider_address}")
print("=" * 60)
if status and 'cluster' in status:
cluster = status['cluster']
if 'leases' in cluster:
active = cluster['leases'].get('active', 0)
available = cluster['leases'].get('available', 0)
print(f"Active leases: {active}")
print(f"Available slots: {available}")
if 'inventory' in cluster and 'cluster' in cluster['inventory']:
nodes = cluster['inventory']['cluster'].get('nodes', [])
print(f"Cluster nodes: {len(nodes)}")
if nodes:
total_cpu = sum(node.get('cpu', {}).get('quantity', {}).get('value', 0) for node in nodes)
total_memory = sum(node.get('memory', {}).get('quantity', {}).get('value', 0) for node in nodes)
print(f"Total CPU: {total_cpu}")
print(f"Total memory: {total_memory}")
else:
print("No cluster data available")
return status
except Exception as e:
print(f"Cannot reach provider off-chain endpoint: {e}")
return None
realtime_status = check_provider_realtime_status("akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63")
Score and rank providers¶
from akash import AkashClient, AkashWallet
def score_providers(providers, weights=None):
"""Score and rank providers based on multiple criteria"""
if weights is None:
weights = {
"capabilities": 0.5,
"region": 0.3,
"features": 0.2
}
scored_providers = []
for provider in providers:
score = 0
details = {}
capabilities = 0
features = []
for attr in provider.get('attributes', []):
if attr['key'].startswith("capabilities/"):
if attr['value'].lower() == "true":
capabilities += 1
elif attr['key'].startswith("feat-"):
if attr['value'].lower() == "true":
features.append(attr['key'])
if capabilities > 0:
capability_score = min(capabilities / 10, 1.0)
score += weights["capabilities"] * capability_score
details["capabilities"] = capabilities
if features:
feature_score = min(len(features) / 5, 1.0)
score += weights["features"] * feature_score
details["features"] = len(features)
scored_providers.append({
"provider": provider,
"score": score,
"details": details
})
scored_providers.sort(key=lambda x: x["score"], reverse=True)
print("Provider ranking by score:")
print("=" * 60)
for i, sp in enumerate(scored_providers[:10], 1):
provider = sp["provider"]
score = sp["score"]
details = sp["details"]
print(f"{i}. {provider['owner'][:30]}...")
print(f"Score: {score:.2f}")
print(f"Details: {details}")
print()
return scored_providers
client = AkashClient("https://akash-rpc.polkachu.com:443")
all_providers = []
offset = 0
limit = 100
while True:
providers = client.provider.get_providers(limit=limit, offset=offset)
if not providers:
break
all_providers.extend(providers)
if len(providers) < limit:
break
offset += limit
ranked_providers = score_providers(all_providers)
Provider interaction¶
Get provider endpoint¶
from akash import AkashClient
def get_provider_endpoint(provider_address):
"""Get the gRPC endpoint for direct provider communication"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
try:
provider = client.provider.get_provider(provider_address)
host_uri = provider.get('host_uri')
if host_uri:
if not host_uri.startswith("http"):
endpoint = f"https://{host_uri}"
else:
endpoint = host_uri
if ":8443" not in endpoint and ":443" not in endpoint:
endpoint = f"{endpoint}:8443"
print(f"Provider endpoint: {endpoint}")
return endpoint
else:
print("No host URI found for provider")
return None
except Exception as e:
print(f"Error getting provider endpoint: {e}")
return None
endpoint = get_provider_endpoint("akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63")
Query provider attributes¶
from akash import AkashClient
def query_provider_attributes(provider_address):
"""Query and analyze provider attributes"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
try:
provider = client.provider.get_provider(provider_address)
attributes = provider.get('attributes', [])
categorized = {
"capabilities": [],
"resources": [],
"location": [],
"network": [],
"other": []
}
for attr in attributes:
key = attr['key']
value = attr['value']
if key.startswith("feat-") or key.startswith("capabilities/"):
categorized["capabilities"].append(f"{key}: {value}")
elif any(r in key for r in ["cpu", "memory", "storage", "gpu", "hardware"]):
categorized["resources"].append(f"{key}: {value}")
elif any(l in key for l in ["region", "location", "country", "city"]):
categorized["location"].append(f"{key}: {value}")
elif any(n in key for n in ["network", "bandwidth", "endpoint", "host"]):
categorized["network"].append(f"{key}: {value}")
else:
categorized["other"].append(f"{key}: {value}")
print(f"Provider attributes: {provider_address}")
print("=" * 60)
for category, attrs in categorized.items():
if attrs:
print(f"\n{category.upper()}:")
for attr in attrs:
print(f"{attr}")
return categorized
except Exception as e:
print(f"Error querying provider attributes: {e}")
return None
attributes = query_provider_attributes("akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63")
Provider monitoring¶
Monitor provider performance¶
from akash import AkashClient, AkashWallet
import time
def monitor_provider_performance(provider_address, duration=300):
"""Monitor provider performance over time"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
start_time = time.time()
measurements = []
print(f"Monitoring provider {provider_address} for {duration}s...")
print("=" * 60)
while time.time() - start_time < duration:
try:
leases = client.provider.get_provider_leases(provider_address)
measurement = {
"timestamp": time.time(),
"active_leases": len(leases)
}
measurements.append(measurement)
print(f"[{time.strftime('%H:%M:%S')}] Active leases: {len(leases)}")
time.sleep(30)
except Exception as e:
print(f"Monitoring error: {e}")
time.sleep(10)
if measurements:
avg_leases = sum(m["active_leases"] for m in measurements) / len(measurements)
print(f"\nMonitoring summary:")
print(f"Average active leases: {avg_leases:.1f}")
print(f"Total measurements: {len(measurements)}")
return measurements
# metrics = monitor_provider_performance("akash1365yvmc4s7awdyj3n2sav7xfx76adc6dnmlx63", 300)
Best practices¶
Provider selection strategy¶
from akash import AkashClient
def select_best_provider(requirements):
"""Select the best provider based on requirements"""
client = AkashClient("https://akash-rpc.polkachu.com:443")
region = requirements.get("region")
capabilities = requirements.get("capabilities", [])
if region:
providers = client.provider.get_providers_by_region(region)
else:
all_providers = []
offset = 0
limit = 100
while True:
batch = client.provider.get_providers(limit=limit, offset=offset)
if not batch:
break
all_providers.extend(batch)
if len(batch) < limit:
break
offset += limit
providers = all_providers
if capabilities:
filtered = []
for provider in providers:
provider_caps = set()
for attr in provider.get('attributes', []):
if attr['key'].startswith("feat-") and attr['value'].lower() == "true":
provider_caps.add(attr['key'])
elif attr['key'].startswith("capabilities/") and attr['value'].lower() == "true":
provider_caps.add(attr['key'])
has_all = True
for required_cap in capabilities:
found = False
for provider_cap in provider_caps:
if required_cap in provider_cap or provider_cap in required_cap:
found = True
break
if not found:
has_all = False
break
if has_all:
filtered.append(provider)
providers = filtered
if not providers:
print("No providers match the requirements")
return None
scored = score_providers(providers)
if scored:
best = scored[0]
provider = best["provider"]
print(f"Selected provider: {provider['owner']}")
print(f"Score: {best['score']:.2f}")
print(f"Host: {provider.get('host_uri', 'N/A')}")
return provider
return None
requirements = {
"region": "United States",
"capabilities": ["persistent-storage", "endpoint-ip"]
}
best_provider = select_best_provider(requirements)