gRPC client module¶

The gRPC client module provides infrastructure for communication with Akash providers using gRPC protocol. It handles SSL connections with self-signed certificates that providers typically use. This is primarily used internally by other modules but can be used directly for custom provider integrations.

ProviderGRPCClient Class¶

from akash import AkashClient
from akash.grpc_client import ProviderGRPCClient

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

result = grpc_client.get_provider_status(provider_address)

Constructor parameters¶

Parameter	Type	Default	Description
`akash_client`	`AkashClient`	Required	Parent AkashClient instance
`timeout`	`int`	`30`	Default timeout for gRPC calls in seconds
`retries`	`int`	`3`	Default number of retries for failed calls

Provider status methods¶

get_provider_status()¶

Get provider status information via gRPC or REST API with automatic protocol selection.

def get_provider_status(
    provider_address: str,
    retries: Optional[int] = None,
    timeout: Optional[int] = None,
    insecure: bool = True,
    check_version: bool = True
) -> Dict[str, Any]

Required parameters:

provider_address (str): Provider's blockchain address

Optional parameters:

retries (int): Number of retries (default: uses client default)
timeout (int): Timeout in seconds (default: uses client default)
insecure (bool): Skip SSL certificate verification for self-signed certificates (default: True)
check_version (bool): Enable provider version checking for protocol selection (default: True)

Protocol selection:

Checks provider version via REST /version endpoint if check_version=True
Uses REST /status endpoint for providers with version < 0.5.0
Uses gRPC for providers with version >= 0.5.0 or when version check is disabled
Tests multiple ports (8443, 8444) for gRPC connections
Uses DNS pre-filtering to avoid invalid hostnames
Bypasses SSL certificate verification to handle provider self-signed certificates

Returns: Dict[str, Any] with the following fields:

status (str): Operation status ("success" or "error")
response (Dict): Provider status data (if successful) with the following fields:
- cluster (Dict): Cluster status information
  - leases (Dict): Lease information
    - active (int): Number of active leases
    - available (int): Number of available lease slots
  - inventory (Dict): Resource inventory
    - cluster (Dict): Cluster resource information
      - nodes (List[Dict]): List of nodes with resources
        
        name (str): Node name
        
        resources (Dict): Node resources
        
        cpu (Dict): CPU resources
        
        units (str): CPU units available
        
        attributes (List[Dict]): CPU attributes
        
        key (str): Attribute key
        
        value (str): Attribute value
        
        memory (Dict): Memory resources
        
        quantity (str): Memory quantity available
        
        attributes (List[Dict]): Memory attributes
        
        key (str): Attribute key
        
        value (str): Attribute value
        
        gpu (Dict): GPU resources
        
        units (str): GPU units available
        
        attributes (List[Dict]): GPU attributes
        
        key (str): Attribute key
        
        value (str): Attribute value
        
        ephemeral_storage (Dict): Storage resources
        
        quantity (str): Storage quantity available
        
        attributes (List[Dict]): Storage attributes
        
        key (str): Attribute key
        
        value (str): Attribute value
- bid_engine (Dict): Bid engine status
  - orders (int): Number of orders processed
  - available (bool): Whether bid engine is available
- manifest (Dict): Manifest processing status
  - deployments (int): Number of deployments processed
  - available (bool): Whether manifest processing is available
- errors (List[str]): Any error messages
- timestamp (str): Status timestamp
attempts (int): Number of connection attempts made
error (str): Error message (if failed)

Example:

from akash import AkashClient
from akash.grpc_client import ProviderGRPCClient

akash_client = AkashClient("https://akash-rpc.polkachu.com:443")
grpc_client = ProviderGRPCClient(akash_client)

result = grpc_client.get_provider_status(
    provider_address="akash1provider123...",
    timeout=10,
    retries=2
)

if result["status"] == "success":
    provider_data = result["response"]
    print(f"Provider status retrieved in {result['attempts']} attempts")
    print(f"Cluster leases: {provider_data.get('cluster', {}).get('leases', 0)}")

    inventory = provider_data.get('cluster', {}).get('inventory', {})
    print(f"Available resources: {inventory}")
else:
    print(f"Failed to get provider status: {result['error']}")
    print(f"Attempted {result['attempts']} times")

Management methods¶

cleanup_connections()¶

Clean up all active gRPC connections.

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

Returns: Dict[str, Any] - Cleanup result with the following fields:

status (str): Cleanup status ("success" or "error")
closed_channels (int): Number of gRPC channels closed
error (str, optional): Error message if cleanup failed

Example:

from akash import AkashClient
from akash.grpc_client import ProviderGRPCClient

akash_client = AkashClient("https://akash-rpc.polkachu.com:443")
grpc_client = ProviderGRPCClient(akash_client)

result = grpc_client.cleanup_connections()
if result["status"] == "success":
    print(f"Closed {result['closed_channels']} connections")
else:
    print(f"Cleanup error: {result['error']}")

Context manager support¶

The ProviderGRPCClient supports context manager usage for automatic cleanup.

from akash import AkashClient
from akash.grpc_client import ProviderGRPCClient

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

with ProviderGRPCClient(akash_client) as grpc_client:
    result = grpc_client.get_provider_status("akash1provider123...")

Usage examples¶

Basic provider status query¶

from akash import AkashClient
from akash.grpc_client import ProviderGRPCClient

akash_client = AkashClient("https://akash-rpc.polkachu.com:443")
grpc_client = ProviderGRPCClient(akash_client, timeout=60, retries=5)

provider_address = "akash1provider123..."

try:
    result = grpc_client.get_provider_status(
        provider_address=provider_address,
        timeout=30,
        retries=3
    )

    if result["status"] == "success":
        provider_data = result["response"]
        print(f"Provider status retrieved successfully")
        print(f"Attempts made: {result['attempts']}")
    else:
        print(f"Failed to get provider status: {result['error']}")

finally:
    grpc_client.cleanup_connections()

Integration patterns¶

High-level module usage¶

Most users will not use ProviderGRPCClient directly. Instead, it's used internally by other modules:

from akash import AkashClient

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

# send_manifest uses gRPC via ProviderGRPCClient
result = client.manifest.send_manifest(
    provider_endpoint=endpoint,
    lease_id=lease_id,
    manifest_groups=groups
)

# Provider status queries use gRPC 
status = client.provider.get_provider_status("akash1provider123...")

Using with discovery module¶

The ProviderGRPCClient is designed to work with the discovery module:

from akash import AkashClient

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

provider_addresses = [
    "akash1provider123...",
    "akash1provider456..."
]

for provider_address in provider_addresses:
    result = client.discovery.get_provider_status(provider_address)

    if result["status"] == "success":
        provider_status = result["provider_status"]
        method = provider_status.get("method", "Unknown")
        print(f"Provider {provider_address}: {method} connection successful")

        leases = provider_status.get("cluster", {}).get("leases", {})
        print(f"Active leases: {leases.get('active', 0)}")
    else:
        print(f"Provider {provider_address}: Failed - {result.get('error', 'Unknown error')}")

print("Discovery complete")