Python client for Proxmox: building robust automation

For Python automation, the standard library is proxmoxer. It wraps the REST API with Python objects, handles authentication, and provides both sync and async clients.

Installing proxmoxer

pip install proxmoxer requests

# Or in a virtualenv
python3 -m venv proxmox-venv
source proxmox-venv/bin/activate
pip install proxmoxer

Basic usage

import proxmoxer

# Connect with API token
prox = proxmoxer.ProxmoxAPI(
    host='pve-01.cluster.example.com',
    user='automation@pve',
    token_name='automation',
    token_value='12345678-1234-1234-1234-123456789abc',
    verify_ssl=False,  # Set True in production
)

# List all nodes
for node in prox.nodes.get():
    print(node['node'], node['status'])

# List VMs on a specific node
for vm in prox.nodes('pve-01').qemu.get():
    print(vm['vmid'], vm['name'], vm['status'])

# Get a specific VM
vm = prox.nodes('pve-01').qemu(100)
config = vm.config.get()
print(config['memory'], config['cores'])

The library uses proxmoxer.ProxmoxAPI as the entry point, with a hierarchical structure that mirrors the API paths. nodes('pve-01').qemu(100) is shorthand for /nodes/pve-01/qemu/100.

Common operations

Create a VM from a template

import time

# Clone from template
new_vmid = prox.nodes('pve-01').qemu.post(
    newid=200,
    name='web-01',
    clone='debian-12-web-template',
    full=False,
    storage='local-zfs',
)

# Wait for the clone to complete (task UPID returned)
task = prox.nodes('pve-01').qemu(new_vmid).status.current.get()
print(f"Clone started: {task}")

# Configure network
prox.nodes('pve-01').qemu(new_vmid).config.post(
    net0='virtio,bridge=vmbr0,tag=100',
    ipconfig0='ip=dhcp',
)

# Start the VM
prox.nodes('pve-01').qemu(new_vmid).status.start.post()

# Wait for the VM to be running
while True:
    status = prox.nodes('pve-01').qemu(new_vmid).status.current.get()
    if status['status'] == 'running':
        print(f"VM {new_vmid} is running")
        break
    time.sleep(2)

Bulk operations

# Start all VMs tagged 'production'
for vm in prox.cluster.resources.get(type='vm'):
    if 'production' in (vm.get('tags') or \').split(';'):
        try:
            prox.nodes(vm['node']).qemu(vm['vmid']).status.start.post()
            print(f"Started {vm['name']}")
        except proxmoxer.core.ResourceException as e:
            print(f"Failed to start {vm['name']}: {e}")

Async client

For high-throughput operations:

import asyncio
from proxmoxer import ProxmoxAPI

async def list_vms_async(prox):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, prox.cluster.resources.get)

async def main():
    prox = proxmoxer.ProxmoxAPI(
        host='pve-01.cluster.example.com',
        user='automation@pve',
        token_name='automation',
        token_value='12345678-1234-1234-1234-123456789abc',
    )
    vms = await list_vms_async(prox)
    print(f"Found {len(vms)} VMs")

For true async, use the aiohttp library directly with async session managers.

Error handling

from proxmoxer.core import ResourceException
import requests.exceptions

try:
    prox.nodes('pve-01').qemu(999).config.get()
except proxmoxer.core.ResourceException as e:
    # HTTP errors from PVE
    print(f"PVE error: {e.status_code} {e.message}")
    # Inspect details
    print(f"Errors object: {e.errors}")
except requests.exceptions.ConnectionError as e:
    # Network errors
    print(f"Connection failed: {e}")
except requests.exceptions.Timeout as e:
    # Timeout
    print(f"Request timed out: {e}")

The library raises ResourceException for HTTP error responses and lets standard requests exceptions propagate for transport-level errors.

Retries with exponential backoff

For transient errors, wrap API calls with retry logic:

import time
import requests.exceptions
from proxmoxer.core import ResourceException

def with_retry(func, max_attempts=5, base_delay=1):
    for attempt in range(max_attempts):
        try:
            return func()
        except (requests.exceptions.ConnectionError,
                requests.exceptions.Timeout) as e:
            if attempt == max_attempts - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"Attempt {attempt+1} failed: {e}. Retrying in {delay}s...")
            time.sleep(delay)
        except ResourceException as e:
            # 5xx errors are transient; 4xx are not
            if e.status_code >= 500 and attempt < max_attempts - 1:
                delay = base_delay * (2 ** attempt)
                print(f"PVE &#123;e.status_code&#125;: &#123;e.message&#125;. Retrying in &#123;delay&#125;s...")
                time.sleep(delay)
            else:
                raise

# Usage
vm = with_retry(lambda: prox.nodes('pve-01').qemu(100).config.get())

Connection pooling and timeouts

import requests

# Use a session for connection pooling
session = requests.Session()
session.headers.update({'User-Agent': 'MyApp/1.0'})
session.mount('https://', requests.adapters.HTTPAdapter(
    pool_connections=10,
    pool_maxsize=10,
    max_retries=3,
))

# Pass the session to proxmoxer (requires a custom transport)
prox = proxmoxer.ProxmoxAPI(
    host='pve-01.cluster.example.com',
    user='automation@pve',
    token_name='automation',
    token_value='12345678-...',
    session=session,
    timeout=30,
)

Packaging automation as a CLI

For team use, package your automation as a CLI tool with click or argparse:

#!/usr/bin/env python3
# cluster-tool.py — CLI for cluster operations
import click
import proxmoxer

@click.group()
@click.option('--host', default='pve-01.cluster.example.com')
@click.option('--user', default='automation@pve')
@click.option('--token-name', default='automation')
@click.option('--token-value', envvar='PROXMOX_TOKEN')
@click.pass_context
def cli(ctx, host, user, token_name, token_value):
    ctx.ensure_object(dict)
    ctx.obj['prox'] = proxmoxer.ProxmoxAPI(
        host=host,
        user=user,
        token_name=token_name,
        token_value=token_value,
        verify_ssl=False,
    )

@cli.command()
@click.pass_context
def list_running(ctx):
    """List all running VMs."""
    prox = ctx.obj['prox']
    running = [vm for vm in prox.cluster.resources.get(type='vm')
               if vm['status'] == 'running']
    for vm in running:
        click.echo(f"{vm['vmid']:>4}  {vm['node']:<12}  {vm['name']}")

@cli.command()
@click.argument('vmid', type=int)
@click.argument('target_node')
@click.pass_context
def migrate(ctx, vmid, target_node):
    """Migrate a VM to another node."""
    prox = ctx.obj['prox']
    current = prox.cluster.resources.get(type='vm', vmid=vmid)
    if not current:
        click.echo(f"VM {vmid} not found", err=True)
        return
    node = current[0]['node']
    prox.nodes(node).qemu(vmid).migrate.post(target=target_node, online=1)
    click.echo(f"Migrating VM {vmid} from {node} to {target_node}")

if __name__ == '__main__':
    cli(obj={})

# Install in venv
pip install click

# Use
export PROXMOX_TOKEN="12345678-..."
./cluster-tool.py list-running
./cluster-tool.py migrate 100 pve-02

Production considerations

• Connection pooling. For high-throughput, use a session with a connection pool. Default proxmoxer creates a new connection per request, which is slow at scale.
• Async where it matters. For one-shot scripts, sync is fine. For services that handle many API calls per second, async with aiohttp is necessary.
• Type hints and dataclasses. Wrap API responses in dataclasses for type safety:

  from dataclasses import dataclass
  
  @dataclass
  class VMInfo:
      vmid: int
      name: str
      status: str
  
  def parse_vm(data: dict) -> VMInfo:
      return VMInfo(
          vmid=data['vmid'],
          name=data['name'],
          status=data['status'],
      )

• Error budgets. When automating, every operation should have a timeout and retry policy. Operations that hang forever are the hardest bugs to debug.

Common mistakes

• No connection pooling. Creating a new ProxmoxAPI per request is slow. Reuse one instance.
• No timeouts. A hung API call blocks forever. Set timeouts.
• No error context. Just catching ResourceException without parsing e.errors makes debugging hard.
• Logging tokens. Never log API tokens. Mask them or use a secret manager.

Key takeaways

• proxmoxer is the standard Python library for PVE.
• Use API tokens, not passwords, for automation.
• Add retries with exponential backoff for transient errors.
• Use connection pooling for high-throughput scripts.

Knowledge check