Deploying applications

Complete examples for creating and managing deployments on Akash Network using the Python SDK.

Overview

Deployments on Akash Network follow a specific lifecycle:

Deployment Creation:

graph LR
    A[Create SDL] --> B[Validate SDL]
    B --> C[Create Deployment]

Market Interaction:

graph LR
    D[Wait for Bids] --> E["Create Lease (Accept Bid)"]

Deployment Management:

graph LR
    G[Upload Manifest] --> H[Monitor Deployment]
    H --> J[Update/Close]

Basic deployments

Simple web server

from akash import AkashClient, AkashWallet

def deploy_nginx():
    """Deploy a basic Nginx web server"""

    wallet = AkashWallet.from_mnemonic("your mnemonic here")
    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.5
        memory:
          size: 512Mi
        storage:
          size: 1Gi
  placement:
    akash:
      attributes:
        host: akash
      pricing:
        web:
          denom: uakt
          amount: 1000
deployment:
  web:
    akash:
      profile: web
      count: 1
'''

    result = client.deployment.create_deployment(
        sdl_yaml=sdl_content,
        wallet=wallet,
        deposit="500000"
    )

    if result.success:
        print(f"Deployment created: DSEQ {result.dseq}")
        print(f"Transaction: {result.tx_hash}")
        return result
    else:
        print(f"Deployment failed: {result.raw_log}")
        return None

deployment = deploy_nginx()

API reference

See Deployment API for SDL validation, deployment creation, and lifecycle management

Database deployment

from akash import AkashClient, AkashWallet

def deploy_database():
    """Deploy a PostgreSQL database"""

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

    sdl_content = '''
version: "2.0"
services:
  database:
    image: postgres:14
    env:
      - POSTGRES_DB=myapp
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=secure123
      - PGDATA=/var/lib/postgresql/data/pgdata
    expose:
      - port: 5432
        as: 5432
        to:
          - global: true
profiles:
  compute:
    database:
      resources:
        cpu:
          units: 1.0
        memory:
          size: 2Gi
        storage:
          - size: 10Gi
            attributes:
              persistent: true
              class: beta2
  placement:
    akash:
      attributes:
        host: akash
      pricing:
        database:
          denom: uakt
          amount: 5000
deployment:
  database:
    akash:
      profile: database
      count: 1
'''

    result = client.deployment.create_deployment(
        sdl_yaml=sdl_content,
        wallet=wallet,
        deposit="5000000"
    )

    if result.success:
        print(f"Database deployment created: DSEQ {result.dseq}")
        print(f"Transaction: {result.tx_hash}")
        return result
    else:
        print(f"Database deployment failed: {result.raw_log}")
        return None

deployment = deploy_database()

USDC deployment

from akash import AkashClient, AkashWallet

def deploy_with_usdc():
    """Deploy using USDC denom"""

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

    USDC_DENOM = "ibc/170C677610AC31DF0904FFE09CD3B5C657492170E7E52372E48756B71E56F2F1"

    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.5
        memory:
          size: 512Mi
        storage:
          size: 1Gi
  placement:
    akash:
      attributes:
        host: akash
      pricing:
        web:
          denom: ibc/170C677610AC31DF0904FFE09CD3B5C657492170E7E52372E48756B71E56F2F1
          amount: 500000  
deployment:
  web:
    akash:
      profile: web
      count: 1
'''

    result = client.deployment.create_deployment(
        sdl_yaml=sdl_content,
        wallet=wallet,
        deposit="5000000",           
        deposit_denom=USDC_DENOM,   
        fee_amount="7000"           
    )

    if result.success:
        print(f"USDC deployment created: DSEQ {result.dseq}")
        print(f"Transaction: {result.tx_hash}")
        return result
    else:
        print(f"Deployment failed: {result.raw_log}")
        return None

deployment = deploy_with_usdc()

Denomination matching

The deposit_denom parameter must exactly match the denom in your SDL pricing section.

Market interaction

Waiting for bids

from akash import AkashClient, AkashWallet
import time

def wait_for_bids(client, owner, dseq, timeout=300):
    """Wait for bids on a deployment"""
    start_time = time.time()

    while time.time() - start_time < timeout:
        bids = client.market.get_bids(owner=owner, dseq=dseq)

        if bids:
            print(f"Received {len(bids)} bids!")
            for bid in bids:
                print(f"Provider: {bid['bid_id']['provider']}")
                print(f"Price: {bid['price']['amount']} {bid['price']['denom']}")
                print(f"State: {bid['state']}")
            return bids

        print("Waiting for bids...")
        time.sleep(10)

    raise TimeoutError("No bids received within timeout period")

bids = wait_for_bids(client, wallet.address, dseq)

API reference

See Market API for bid listing, lease creation, and market operations

Selecting and accepting bids

from akash import AkashClient, AkashWallet

def select_best_bid(client, bids):
    """Select the best bid based on price and provider validation"""
    if not bids:
        return None

    print("Validating provider endpoints...")
    provider_addresses = [b['bid_id']['provider'] for b in bids]
    valid_providers = client.provider.filter_valid_providers(provider_addresses)

    valid_bids = [b for b in bids if b['bid_id']['provider'] in valid_providers]
    if not valid_bids:
        print("No valid providers found")
        return None

    bids_by_price = sorted(valid_bids, key=lambda b: float(b['price']['amount']))

    print("Valid bid comparison:")
    for i, bid in enumerate(bids_by_price, 1):
        print(f"{i}. Provider: {bid['bid_id']['provider']}")
        print(f"Price: {bid['price']['amount']} {bid['price']['denom']}")
        print(f"State: {bid['state']}")

    return bids_by_price[0] if bids_by_price else None

def accept_bid(client, bid, wallet):
    """Accept a bid and create a lease"""
    print(f"Accepting bid from {bid['bid_id']['provider']}")

    lease_result = client.market.create_lease_from_bid(wallet, bid)

    if lease_result["success"]:
        print(f"Lease created: TX {lease_result['tx_hash']}")
        print(f"Provider endpoint: {lease_result['provider_endpoint']}")
        return lease_result
    else:
        print(f"Lease creation failed: {lease_result['error']}")
        return None

best_bid = select_best_bid(client, bids)
if best_bid:
    lease = accept_bid(client, best_bid, wallet)

Deployment monitoring

Monitor status and get URIs

from akash import AkashClient, AkashWallet

def monitor_deployment(client, owner, dseq):
    """Monitor deployment status and get service information"""
    deployments = client.deployment.get_deployments(owner=owner)
    deployment = next((d for d in deployments if d['deployment']['deployment_id']['dseq'] == str(dseq)), None)

    if deployment:
        print(f"Deployment status: {deployment['deployment']['state']}")
        print(f"DSEQ: {deployment['deployment']['deployment_id']['dseq']}")
        print(f"Owner: {deployment['deployment']['deployment_id']['owner']}")

    leases = client.market.get_leases(owner=owner, dseq=dseq)

    for lease in leases:
        print(f"\nLease: {lease['lease_id']['provider']}")
        print(f"State: {lease['state']}")
        print(f"Price: {lease['price']['amount']} {lease['price']['denom']}")

        if lease['state'] == "active":
            try:
                provider_info = client.provider.get_providers()
                provider = next((p for p in provider_info if p['owner'] == lease['lease_id']['provider']), None)
                if provider:
                    print(f"Provider endpoint: {provider['host_uri']}")
            except Exception as e:
                print(f"Could not get provider info: {e}")

monitor_deployment(client, wallet.address, dseq)

Access deployment logs

from akash import AkashClient, AkashWallet

def get_deployment_logs(client, provider_endpoint, lease_id, service_name="web", follow=False):
    """Get logs from a deployment service"""
    try:
        if follow:
            print(f"Following logs for {service_name}...")
            for log_line in client.deployment.stream_service_logs(
                provider_endpoint=provider_endpoint,
                lease_id=lease_id,
                service_name=service_name,
                follow=True
            ):
                print(f"[{service_name}] {log_line}")
        else:
            logs = client.deployment.get_service_logs(
                provider_endpoint=provider_endpoint,
                lease_id=lease_id,
                service_name=service_name,
                tail=100
            )
            print(f"Recent logs for {service_name}:")
            for log_line in logs:
                print(f"[{service_name}] {log_line}")

    except Exception as e:
        print(f"Error getting logs: {e}")

get_deployment_logs(client, provider_endpoint, lease_id, "web")

get_deployment_logs(client, provider_endpoint, lease_id, "web", follow=True)

Complete deployment workflow

Deploy nginx with full lifecycle

from akash import AkashClient, AkashWallet
import time

def complete_deployment_example():
    """Complete deployment workflow from start to running application"""

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

    print("Step 1: Ensuring certificate")
    success, cert_pem, key_pem = client.cert.ensure_certificate(wallet)
    if not success:
        print("Certificate setup failed")
        return None

    print("Step 2: Creating deployment from SDL...")
    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:
      attributes:
        host: akash
      pricing:
        web:
          denom: uakt
          amount: 1000
deployment:
  web:
    akash:
      profile: web
      count: 1
'''

    deployment_result = client.deployment.create_deployment(
        sdl_yaml=sdl_content,
        wallet=wallet,
        deposit="500000"
    )

    if not deployment_result.success:
        print(f"Deployment failed: {deployment_result.raw_log}")
        return None

    dseq = deployment_result.dseq
    print(f"Deployment created: DSEQ {dseq}")
    print(f"Transaction: {deployment_result.tx_hash}")

    print("Step 3: Waiting for bids...")
    time.sleep(30)

    bids = client.market.get_bids(owner=wallet.address, dseq=dseq)
    if not bids:
        print("No bids received")
        return None

    print("Step 4: Selecting best validated provider...")
    provider_addresses = [b['bid_id']['provider'] for b in bids]
    valid_providers = client.provider.filter_valid_providers(provider_addresses)

    if not valid_providers:
        print("No valid providers found")
        return None

    valid_bids = [b for b in bids if b['bid_id']['provider'] in valid_providers]
    best_bid = min(valid_bids, key=lambda b: float(b['price']['amount']))

    provider = best_bid['bid_id']['provider']
    print(f"Best bid from provider: {provider}")

    print("Step 5: Creating lease...")
    lease_result = client.market.create_lease_from_bid(wallet, best_bid)

    if not lease_result["success"]:
        print(f"Lease creation failed: {lease_result['error']}")
        return None

    print(f"Lease created: TX {lease_result['tx_hash']}")
    provider_endpoint = lease_result['provider_endpoint']

    print("Step 6: Submitting manifest to provider...")
    lease_id = lease_result['lease_id']

    manifest_result = client.manifest.submit_manifest(
        provider_endpoint=provider_endpoint,
        lease_id=lease_id,
        sdl_content=sdl_content,
        cert_pem=cert_pem,
        key_pem=key_pem
    )

    if manifest_result["status"] != "success":
        print(f"Manifest submission failed: {manifest_result['error']}")
        return None

    print(f"Manifest submitted via {manifest_result.get('method', 'HTTP')}")
    print("Your nginx application is now starting...")

    print("Step 7: Waiting for application to start...")
    time.sleep(60)  # Give nginx time to boot

    print("Step 8: Retrieving application logs...")
    try:
        logs = client.deployment.get_service_logs(
            provider_endpoint=provider_endpoint,
            lease_id=lease_id,
            tail=10
        )
        print(f"Retrieved {len(logs)} log lines:")
        for log in logs:
            print(f"{log}")
    except Exception as e:
        print(f"Log retrieval note: {e}")

    print("Step 9: Streaming live logs for 60 seconds...")
    try:
        log_count = 0
        for log_line in client.deployment.stream_service_logs(
            provider_endpoint=provider_endpoint,
            lease_id=lease_id,
            follow=True,
            timeout=60
        ):
            print(f"[LIVE] {log_line}")
            log_count += 1
            if log_count >= 10:
                print("   (stopping after 10 lines for demo)")
                break
    except Exception as e:
        print(f"Live log streaming note: {e}")

    print("\n" + "="*60)
    print("Deployment complete! Your nginx server is running.")
    print(f"DSEQ: {dseq}")
    print(f"Provider: {provider}")
    print(f"Provider endpoint: {provider_endpoint}")
    print("="*60)

    return {
        "dseq": dseq,
        "provider": provider,
        "provider_endpoint": provider_endpoint,
        "lease_id": lease_id
    }

if __name__ == "__main__":
    complete_deployment_example()

Deployment management

Update existing deployment

from akash import AkashClient, AkashWallet
import time

def update_deployment_example():
    """Update an existing deployment with new configuration"""

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

    dseq = 12345
    provider_endpoint = "https://provider.europlots.com:8443"
    lease_id = {
        "dseq": str(dseq),
        "gseq": 1,
        "oseq": 1
    }

    success, cert_pem, key_pem = client.cert.ensure_certificate(wallet)
    if not success:
        print("Certificate setup failed")
        return None

    updated_sdl_content = '''
version: "2.0"
services:
  web:
    image: nginx:1.21
    command: ["/bin/sh", "-c", "echo 'Updated container' && nginx -g 'daemon off;'"]
    env:
      - NGINX_WORKER_PROCESSES=auto
      - UPDATE_VERSION=v2
    expose:
      - port: 80
        as: 80
        to:
          - global: true
profiles:
  compute:
    web:
      resources:
        cpu:
          units: 0.2
        memory:
          size: 256Mi
        storage:
          size: 1Gi
  placement:
    akash:
      pricing:
        web:
          denom: uakt
          amount: 1500
deployment:
  web:
    akash:
      profile: web
      count: 1
'''

    print("Step 1: Updating deployment on-chain...")

    update_result = client.deployment.update_deployment(
        wallet=wallet,
        sdl_yaml=updated_sdl_content,
        owner=wallet.address,
        dseq=dseq,
        fee_amount="5000"
    )

    if not update_result.success:
        print(f"Deployment update failed: {update_result.raw_log}")
        return None

    print(f"Deployment updated on-chain: tx {update_result.tx_hash}")
    time.sleep(10)

    print("\nStep 2: Submitting updated manifest to provider...")

    manifest_result = client.manifest.submit_manifest(
        provider_endpoint=provider_endpoint,
        lease_id=lease_id,
        sdl_content=updated_sdl_content,
        cert_pem=cert_pem,
        key_pem=key_pem
    )

    if manifest_result['status'] != 'success':
        print(f"Manifest submission failed: {manifest_result.get('error')}")
        return None

    print("Updated manifest submitted successfully!")
    print(f"Method: {manifest_result.get('method')}")

    return update_result

update_result = update_deployment_example()

Two-step update process

Updating a deployment requires two steps:

1. Update on-chain: Call update_deployment() to update the version hash on the blockchain
2. Submit manifest: Send the updated manifest to the provider(s) using submit_manifest()

The provider validates that the manifest version hash matches the on-chain deployment version before applying changes.

Query deployment manifest

from akash import AkashClient, AkashWallet
import time

def query_manifest_example():
    """Query manifest from provider to verify deployment configuration"""

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

    dseq = 12345
    provider_endpoint = "https://provider.europlots.com:8443"
    lease_id = {
        "dseq": str(dseq),
        "gseq": 1,
        "oseq": 1
    }

    success, cert_pem, key_pem = client.cert.ensure_certificate(wallet)
    if not success:
        print("Certificate setup failed")
        return None

    print("Querying manifest from provider...")
    time.sleep(10)  

    result = client.manifest.get_deployment_manifest(
        provider_endpoint=provider_endpoint,
        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"Query method: {result['method']}")

        manifest = result["manifest"]
        print(f"\nRetrieved 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']}")
                if 'command' in service:
                    print(f" Command: {service['command']}")
                if 'env' in service:
                    print(f" Environment: {service['env']}")

                resources = service['resources']
                cpu_val = resources['cpu']['units']['val']
                mem_val = resources['memory']['size']['val']
                print(f" CPU: {cpu_val} millicores")
                print(f" Memory: {int(mem_val) / (1024**2):.0f}Mi")

                if 'expose' in service:
                    print(f" Exposed ports:")
                    for expose in service['expose']:
                        print(f"  Port {expose['port']} -> {expose['externalPort']} ({expose['proto']})")
    else:
        print(f"Failed to get manifest: {result['error']}")

    return result

query_result = query_manifest_example()

API reference

See Manifest API for manifest parsing, validation, and provider operations

Provider-stored manifests

Manifests are stored in the provider's Kubernetes cluster, NOT on the blockchain. The blockchain only stores resource requirements (CPU, memory, storage, GPU). The get_deployment_manifest() function retrieves the full manifest from the provider via HTTP API, with automatic version detection for both modern and legacy providers.

Close deployment

from akash import AkashClient, AkashWallet

def close_deployment(client, owner, dseq, wallet):
    """Close a deployment and all associated leases"""
    print("Closing deployment...")

    leases = client.market.get_leases(owner=owner, dseq=dseq)

    for lease in leases:
        if lease['state'] == "active":
            print(f"Closing lease with {lease['lease_id']['provider']}")
            close_result = client.market.close_lease(
                wallet=wallet,
                lease_id=lease['lease_id']
            )
            if close_result.success:
                print(f"Lease closed: TX {close_result.tx_hash}")

    close_result = client.deployment.close_deployment(
        wallet=wallet,
        owner=owner,
        dseq=dseq
    )

    if close_result.success:
        print(f"Deployment closed successfully: TX {close_result.tx_hash}")
    else:
        print(f"Deployment closure failed: {close_result.raw_log}")

close_deployment(client, wallet.address, dseq, wallet)