Manifest¶

The manifest module provides manifest functionality for the Akash Network, including SDL parsing, manifest validation, and off-chain provider communication.

Off-chain module

The manifest module handles off-chain operations with providers. For deployment creation, use the deployment module.

ManifestClient class¶

from akash import AkashClient

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

manifest = client.manifest

Provider communication¶

submit_manifest()¶

Submit manifest to provider (automatically detects provider version).

def submit_manifest(provider_endpoint: str, lease_id: Dict[str, Any], sdl_content: str,
                   cert_pem: str, key_pem: str, timeout: int = 60) -> Dict[str, Any]

Required parameters:

provider_endpoint (str): Provider endpoint URL
lease_id (Dict[str, Any]): Lease identifier
- owner (str): Deployment owner address
- dseq (str): Deployment sequence
- gseq (int): Group sequence
- oseq (int): Order sequence
- provider (str): Provider address
sdl_content (str): SDL manifest content
cert_pem (str): Client certificate PEM
key_pem (str): Client private key PEM

Optional parameters:

timeout (int): Request timeout in seconds (default: 60)

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

Success case:

status (str): Submission status ("success")
provider (str): Provider endpoint
lease_id (Dict): Lease identifier
- owner (str): Deployment owner address
- dseq (str): Deployment sequence
- gseq (int): Group sequence
- oseq (int): Order sequence
- provider (str): Provider address
method (str): Submission method used ("HTTP" or "HTTP (socket-based)")
status_code (int): HTTP status code
provider_version (str): Detected provider version

Error case:

status (str): Submission status ("error")
error (str): Error message
provider_version (str, optional): Detected provider version (if detection succeeded)

Example:

from akash import AkashClient, AkashWallet

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

sdl_content = '''
version: "2.0"
services:
  web:
    image: nginx:latest
    expose:
      - port: 80
        as: 80
        to:
          - global: true
profiles:
  compute:
    web:
      resources:
        cpu:
          units: 0.1
        memory:
          size: 128Mi
        storage:
          size: 512Mi
  placement:
    akash:
      pricing:
        web:
          denom: uakt
          amount: 1000
deployment:
  web:
    akash:
      profile: web
      count: 1
'''

lease_id = {
    "dseq": "12345",
    "gseq": 1,
    "oseq": 1
}

success, cert_pem, key_pem = client.cert.ensure_certificate(wallet)

result = client.manifest.submit_manifest(
    provider_endpoint="https://provider.example.com:8443",
    lease_id=lease_id,
    sdl_content=sdl_content,
    cert_pem=cert_pem,
    key_pem=key_pem
)

if result["status"] == "success":
    print(f"Manifest submitted via {result['method']}")
else:
    print(f"Submission failed: {result['error']}")

Queries¶

get_deployment_manifest()¶

Query deployment manifest from provider's cluster (automatically detects provider version).

Provider-stored manifests

Manifests are stored in the provider's Kubernetes cluster after submission. The blockchain only stores resource requirements (CPU, memory, storage, GPU). This function retrieves the full manifest from the provider via HTTP API.

The function detects the provider version and adapts the authentication method:

Legacy providers (v0.6.x and below): Uses standard client certificate authentication, with automatic fallback to mtls. approach if needed
Modern providers (v0.7+): Uses socket-based mTLS with mtls. prefix for proper authentication
Unknown version: Tries mtls. approach first, with automatic fallback to standard client certificate if needed

def get_deployment_manifest(provider_endpoint: str, lease_id: Dict[str, Any],
                            cert_pem: str, key_pem: str, timeout: int = 30) -> Dict[str, Any]

Required parameters:

provider_endpoint (str): Provider endpoint URL
lease_id (Dict[str, Any]): Lease identifier
- owner (str): Deployment owner address
- dseq (str): Deployment sequence
- gseq (int): Group sequence
- oseq (int): Order sequence
- provider (str): Provider address
cert_pem (str): Client certificate PEM
key_pem (str): Client private key PEM

Optional parameters:

timeout (int): Request timeout in seconds (default: 30)

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

Success case:

status (str): "success"
provider_version (str): Detected provider version (e.g., "v0.8.3-rc10" or "unknown")
method (str): Query method used ("standard", "mtls-socket", "mtls-socket (fallback)", or "standard (fallback)")
manifest (List[Dict]): Manifest groups
- name (str): Group name
- services (List[Dict]): Services in group
  - name (str): Service name
  - image (str): Container image
  - command (List[str], optional): Command override
  - args (List[str], optional): Arguments
  - env (List[str], optional): Environment variables
  - resources (Dict): Resource requirements
    - cpu (Dict): CPU allocation
      - units (Dict): CPU units
        
        val (str): CPU millicores as string (e.g., "100" for 0.1 CPU)
    - memory (Dict): Memory allocation
      - size (Dict): Memory size
        
        val (str): Memory in bytes as string (e.g., "134217728" for 128Mi)
    - storage (List[Dict], optional): Storage volumes
      - name (str): Storage volume name (e.g., "default")
      - size (Dict): Storage size
        
        val (str): Storage in bytes as string (e.g., "536870912" for 512Mi)
    - gpu (Dict, optional): GPU allocation
      - units (Dict): GPU units
        
        val (str): Number of GPUs as string
      - attributes (List[Dict], optional): GPU attributes/requirements
        
        key (str): Attribute key (e.g., "vendor/nvidia/model/rtx4090")
  - expose (List[Dict]): Port exposures
    - port (int): Container port
    - as (int): External port
    - proto (str): Protocol
    - global (bool): Global exposure
  - params (Dict, optional): Service parameters for storage mounts
    - storage (List[Dict]): Storage mount configurations
      - name (str): Storage volume name (must match a storage resource)
      - mount (str): Container mount path
      - readOnly (bool): Whether mount is read-only
  - credentials (Dict, optional): Private container registry credentials
    - host (str): Registry hostname (e.g., "docker.io", "ghcr.io")
    - username (str): Registry username
    - password (str): Registry password/token
    - email (str): Registry email (optional)
  - count (int): Number of replicas

Error case:

status (str): "error"
provider_version (str): Detected provider version (if detection succeeded)
method (str): Query method attempted (if applicable)
error (str): Error message

Example:

from akash import AkashClient, AkashWallet

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

success, cert_pem, key_pem = client.cert.ensure_certificate(wallet)

lease_id = {
    "dseq": "12345",
    "gseq": 1,
    "oseq": 1
}

result = client.manifest.get_deployment_manifest(
    provider_endpoint="https://provider.example.com:8443",
    lease_id=lease_id,
    cert_pem=cert_pem,
    key_pem=key_pem
)

if result["status"] == "success":
    print(f"Provider version: {result['provider_version']}")
    print(f"Method: {result['method']}")

    manifest = result["manifest"]
    print(f"Retrieved manifest with {len(manifest)} groups")

    for group in manifest:
        print(f"\nGroup: {group['name']}")
        for service in group['services']:
            print(f" Service: {service['name']}")
            print(f" Image: {service['image']}")
            print(f" Count: {service['count']}")

            if 'command' in service:
                print(f" Command: {service['command']}")
            if 'args' in service:
                print(f" Args: {service['args']}")
            if 'env' in service:
                print(f" Environment: {service['env']}")

            if 'resources' in service:
                resources = service['resources']
                print(f" Resources:")

                cpu_val = resources['cpu']['units']['val']
                print(f" CPU: {cpu_val} millicores")

                mem_val = resources['memory']['size']['val']
                mem_mb = int(mem_val) / (1024**2)
                print(f" Memory: {mem_mb:.0f}Mi")

                for storage in resources.get('storage', []):
                    storage_val = storage['size']['val']
                    storage_gb = int(storage_val) / (1024**3)
                    print(f" Storage ({storage['name']}): {storage_gb:.1f}Gi")

                if 'gpu' in resources:
                    gpu_val = resources['gpu']['units']['val']
                    print(f" GPU: {gpu_val} units")
                    if 'attributes' in resources['gpu']:
                        for attr in resources['gpu']['attributes']:
                            print(f" {attr['key']}")

            if 'expose' in service:
                print(f" Exposed ports:")
                for expose in service['expose']:
                    proto = expose.get('proto', 'TCP')
                    port = expose['port']
                    ext_port = expose.get('externalPort', port)
                    global_str = " (global)" if expose.get('global') else ""
                    print(f" {port} -> {ext_port} ({proto}){global_str}")

            if 'params' in service and service['params']:
                params = service['params']
                if 'storage' in params:
                    print(f" Storage mounts:")
                    for mount in params['storage']:
                        readonly = " (read-only)" if mount.get('readOnly') else ""
                        print(f" {mount['name']} -> {mount['mount']}{readonly}")

            if 'credentials' in service and service['credentials']:
                creds = service['credentials']
                print(f" Registry: {creds['host']}")
                print(f" Registry user: {creds['username']}")
else:
    print(f"Failed to get manifest: {result['error']}")

Utilities¶

parse_sdl()¶

Parse SDL (YAML) into a manifest dictionary following Akash SDL specification.

def parse_sdl(sdl_content: str) -> Dict[str, Any]

Required parameters:

sdl_content (str): SDL content in YAML format

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

status (str): Parse status ("success", "error", or "failed")
manifest_data (List[Dict], optional): Parsed manifest groups if successful
- Name (str): Group name
- Services (List[Dict]): List of services in group
  - name (str): Service name
  - image (str): Container image
  - count (int): Number of replicas
  - resources (Dict): Resource requirements
    - cpu (Dict): CPU allocation
      - units (Dict): CPU units
        
        val (str): CPU millicores as string (e.g., "100" for 0.1 CPU)
    - memory (Dict): Memory allocation
      - size (Dict): Memory size
        
        val (str): Memory in bytes as string (e.g., "134217728" for 128Mi)
    - storage (List[Dict]): Storage allocation
      - name (str): Storage volume name (e.g., "default")
      - size (Dict): Storage size
        
        val (str): Storage in bytes as string (e.g., "536870912" for 512Mi)
    - gpu (Dict, optional): GPU allocation
      - units (Dict): GPU units
        
        val (str): Number of GPUs as string
      - attributes (List[Dict], optional): GPU attributes/requirements
        
        key (str): Attribute key (e.g., "vendor/nvidia/model/rtx4090")
  - expose (List[Dict], optional): Port exposures (one entry per "to" target)
    - port (int): Internal container port
    - externalPort (int): External port mapping (from SDL "as" field, defaults to 0)
    - proto (str): Protocol ("TCP" or "UDP")
    - service (str): Target service name (empty string for global exposure)
    - global (bool): Whether exposed globally
    - hosts (List[str]): Accepted hostnames (from SDL "accept" field, null if none)
    - httpOptions (Dict): HTTP configuration options
      - maxBodySize (int): Maximum body size in bytes
      - readTimeout (int): Read timeout in milliseconds
      - sendTimeout (int): Send timeout in milliseconds
      - nextTries (int): Number of retry attempts
      - nextTimeout (int): Timeout between retries in milliseconds
      - nextCases (List[str]): HTTP status codes/errors to retry on
    - ip (str): IP address for IP lease endpoints (empty string if none)
    - endpointSequenceNumber (int): Endpoint sequence number
  - env (List[str], optional): Environment variables
  - command (List[str], optional): Command override
  - params (Dict, optional): Service parameters for storage mounts
    - storage (List[Dict]): Storage mount configurations
      - name (str): Storage volume name
      - mount (str): Container mount path
      - readOnly (bool): Whether mount is read-only
  - credentials (Dict, optional): Private container registry credentials
    - host (str): Registry hostname
    - username (str): Registry username
    - password (str): Registry password/token
    - email (str): Registry email (optional)
error (str, optional): Error message if failed

Example:

from akash import AkashClient

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

sdl_content = '''
version: "2.0"
services:
  web:
    image: nginx:latest
    expose:
      - port: 80
        as: 80
        to:
          - global: true
profiles:
  compute:
    web:
      resources:
        cpu:
          units: 0.1
        memory:
          size: 128Mi
        storage:
          size: 512Mi
  placement:
    akash:
      pricing:
        web:
          denom: uakt
          amount: 1000
deployment:
  web:
    akash:
      profile: web
      count: 1
'''

result = client.manifest.parse_sdl(sdl_content)

if result["status"] == "success":
    manifest = result["manifest_data"]
    print(f"Parsed {len(manifest)} groups")
    for group in manifest:
        print(f"Group {group['Name']}: {len(group['Services'])} services")
else:
    print(f"Parse failed: {result['error']}")

validate_manifest()¶

Basic manifest validation.

def validate_manifest(manifest_data: List[Dict[str, Any]]) -> Dict[str, Any]

Required parameters:

manifest_data (List[Dict]): List of manifest groups to validate
- Name (str): Group name
- Services (List[Dict]): List of services in group
  - name (str): Service name
  - image (str): Container image
  - count (int): Number of replicas
  - resources (Dict): Resource requirements
    - cpu (Dict): CPU allocation
      - units (Dict): CPU units
        
        val (str): CPU millicores as string (e.g., "100" for 0.1 CPU)
    - memory (Dict): Memory allocation
      - size (Dict): Memory size
        
        val (str): Memory in bytes as string (e.g., "134217728" for 128Mi)
    - storage (List[Dict]): Storage allocation
      - name (str): Storage volume name (e.g., "default")
      - size (Dict): Storage size
        
        val (str): Storage in bytes as string (e.g., "536870912" for 512Mi)
    - gpu (Dict, optional): GPU allocation
      - units (Dict): GPU units
        
        val (str): Number of GPUs as string
      - attributes (List[Dict], optional): GPU attributes/requirements
        
        key (str): Attribute key (e.g., "vendor/nvidia/model/rtx4090")
  - expose (List[Dict], optional): Port exposures (one entry per "to" target in SDL)
    - port (int): Internal container port
    - externalPort (int): External port mapping (from SDL "as" field, defaults to 0)
    - proto (str): Protocol ("TCP" or "UDP")
    - service (str): Target service name (empty string for global exposure)
    - global (bool): Whether exposed globally
    - hosts (List[str]): Accepted hostnames (from SDL "accept" field, null if none)
    - httpOptions (Dict): HTTP configuration options
      - maxBodySize (int): Maximum body size in bytes
      - readTimeout (int): Read timeout in milliseconds
      - sendTimeout (int): Send timeout in milliseconds
      - nextTries (int): Number of retry attempts
      - nextTimeout (int): Timeout between retries in milliseconds
      - nextCases (List[str]): HTTP status codes/errors to retry on
    - ip (str): IP address for IP lease endpoints (empty string if none)
    - endpointSequenceNumber (int): Endpoint sequence number
  - env (List[str], optional): Environment variables
  - command (List[str], optional): Command override
  - args (List[str], optional): Command arguments
  - params (Dict, optional): Service parameters for storage mounts
    - storage (List[Dict]): Storage mount configurations
      - name (str): Storage volume name
      - mount (str): Container mount path
      - readOnly (bool): Whether mount is read-only
  - credentials (Dict, optional): Private container registry credentials
    - host (str): Registry hostname
    - username (str): Registry username
    - password (str): Registry password/token
    - email (str): Registry email (optional)

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

valid (bool): Whether manifest is valid
error (str, optional): Error message if validation failed

Example:

from akash import AkashClient

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

manifest_data = [{
    "Name": "akash",
    "Services": [{
        "name": "web",
        "image": "nginx:latest",
        "resources": {
            "cpu": {"units": {"val": "100"}},
            "memory": {"size": {"val": "134217728"}},
            "storage": [{"size": {"val": "536870912"}}]
        },
        "count": 1
    }]
}]

result = client.manifest.validate_manifest(manifest_data)

if result["valid"]:
    print("Manifest is valid")
else:
    print(f"Manifest validation failed: {result['error']}")