Created
May 25, 2025 15:04
-
-
Save edhull/39403081eb6d3ec5ee65b76a6e87db6a to your computer and use it in GitHub Desktop.
Python script which can be used as a dynamic inventory for Proxmox. This will group running VMs/LXC based on tags, and will attempt to connect to VMs via IPs exposed by qemu-guest-agent.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """ | |
| Proxmox Dynamic Inventory Script for Ansible | |
| This script connects to a Proxmox VE API using an API token and dynamically generates | |
| an inventory of running virtual machines and containers. Each VM is grouped by tags, | |
| and metadata is included for Ansible use (e.g., connection type, user, IP address). | |
| Example use: ansible-playbook main.yml -i proxmox_ansible_dynamic_inventory.py | |
| Author: edhull / https://gist.github.com/edhull | |
| """ | |
| import sys | |
| import json | |
| import requests | |
| from urllib3.exceptions import InsecureRequestWarning | |
| # Disable warnings for self-signed certs (optional and insecure, use with caution) | |
| requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) | |
| # === CONFIGURATION === | |
| PROXMOX_HOST = "https://pve.domain.local:8006" | |
| API_USER = "readonly@pam" | |
| TOKEN_NAME = "ansible" | |
| API_TOKEN = "xxxxxxxxxx" | |
| VERIFY_SSL = False # Set to True if using trusted SSL certificates | |
| # Headers for API authentication | |
| HEADERS = { | |
| "Authorization": f"PVEAPIToken={API_USER}!{TOKEN_NAME}={API_TOKEN}" | |
| } | |
| def get_vms(): | |
| """ | |
| Fetch all VM and container resources from the Proxmox cluster. | |
| Returns: | |
| list: Running VMs (excluding templates) with metadata. | |
| """ | |
| url = f"{PROXMOX_HOST}/api2/json/cluster/resources?type=vm" | |
| response = requests.get(url, headers=HEADERS, verify=VERIFY_SSL) | |
| response.raise_for_status() | |
| all_vms = response.json()["data"] | |
| # Filter to include only running, non-template VMs | |
| return [ | |
| vm for vm in all_vms | |
| if not vm.get("template") and vm.get("status") == "running" | |
| ] | |
| def get_vm_type(node, vmid): | |
| """ | |
| Determine if the given VM ID on a node is a QEMU VM or an LXC container. | |
| Args: | |
| node (str): Proxmox node name. | |
| vmid (int or str): Virtual machine ID. | |
| Returns: | |
| str or None: "qemu", "lxc", or None if not found. | |
| """ | |
| try: | |
| url = f"{PROXMOX_HOST}/api2/json/nodes/{node}/qemu" | |
| response = requests.get(url, headers=HEADERS, verify=VERIFY_SSL) | |
| response.raise_for_status() | |
| for vm in response.json().get("data", []): | |
| if str(vm["vmid"]) == str(vmid): | |
| return "qemu" | |
| except Exception: | |
| pass | |
| try: | |
| url = f"{PROXMOX_HOST}/api2/json/nodes/{node}/lxc" | |
| response = requests.get(url, headers=HEADERS, verify=VERIFY_SSL) | |
| response.raise_for_status() | |
| for container in response.json().get("data", []): | |
| if str(container["vmid"]) == str(vmid): | |
| return "lxc" | |
| except Exception: | |
| pass | |
| return None | |
| def get_vm_ip(node, vmid): | |
| """ | |
| Retrieve the first valid IPv4 address of a VM or container. | |
| Args: | |
| node (str): Proxmox node name. | |
| vmid (int or str): VM/container ID. | |
| Returns: | |
| str or None: IP address or None if not found. | |
| """ | |
| type = get_vm_type(node, vmid) | |
| if type is None: | |
| return None | |
| if type == "lxc": | |
| try: | |
| url_lxc = f"{PROXMOX_HOST}/api2/json/nodes/{node}/lxc/{vmid}/interfaces" | |
| response = requests.get(url_lxc, headers=HEADERS, verify=VERIFY_SSL) | |
| lxc_data = response.json().get("data") | |
| if not lxc_data: | |
| return None | |
| for iface in lxc_data: | |
| if iface.get("name") == "lo": | |
| continue | |
| for ip_info in iface.get("ip-addresses", []): | |
| if ip_info.get("ip-address-type") == "inet" and not ip_info.get("ip-address", "").startswith("127."): | |
| return ip_info["ip-address"] | |
| except requests.exceptions.RequestException: | |
| pass | |
| elif type == "qemu": | |
| try: | |
| url_qemu = f"{PROXMOX_HOST}/api2/json/nodes/{node}/qemu/{vmid}/agent/network-get-interfaces" | |
| response = requests.get(url_qemu, headers=HEADERS, verify=VERIFY_SSL) | |
| response.raise_for_status() | |
| interfaces = response.json().get("data", {}).get("result", []) | |
| for iface in interfaces: | |
| for ip_info in iface.get("ip-addresses", []): | |
| if ip_info.get("ip-address-type") == "ipv4" and not ip_info.get("ip-address", "").startswith("127."): | |
| return ip_info["ip-address"] | |
| except Exception: | |
| pass | |
| return None | |
| def get_ssh_user(node, vmid): | |
| """ | |
| Determine the SSH user to use for Ansible connections. | |
| Args: | |
| node (str): Proxmox node name. | |
| vmid (int or str): VM/container ID. | |
| Returns: | |
| str or None: SSH username. | |
| """ | |
| try: | |
| url = f"{PROXMOX_HOST}/api2/json/nodes/{node}/lxc/{vmid}/status/current" | |
| response = requests.get(url, headers=HEADERS, verify=VERIFY_SSL) | |
| if response.status_code == 200: | |
| return "root" | |
| except requests.exceptions.RequestException: | |
| pass | |
| try: | |
| url = f"{PROXMOX_HOST}/api2/json/nodes/{node}/qemu/{vmid}/config" | |
| response = requests.get(url, headers=HEADERS, verify=VERIFY_SSL) | |
| response.raise_for_status() | |
| return response.json().get("data", {}).get("ciuser") | |
| except Exception: | |
| return None | |
| def group_vms_by_tag(vms): | |
| """ | |
| Group VMs by their tags and build an Ansible-compatible inventory. | |
| Args: | |
| vms (list): List of VM metadata dictionaries. | |
| Returns: | |
| dict: Ansible inventory structure with groups and host variables. | |
| """ | |
| inventory = { | |
| "_meta": {"hostvars": {}} | |
| } | |
| for vm in vms: | |
| name = vm.get("name") | |
| if not name: | |
| continue | |
| tags = vm.get("tags", "") | |
| tag_list = [tag.strip() for tag in tags.split(";") if tag.strip()] or ["untagged"] | |
| for tag in tag_list: | |
| inventory.setdefault(tag, {"hosts": []})["hosts"].append(name) | |
| is_windows = "windows" in tag_list | |
| host_vars = { | |
| "nickname": name, | |
| "vmid": vm.get("vmid"), | |
| "proxmox_id": vm.get("vmid"), | |
| "type": vm.get("type"), | |
| "node": vm.get("node"), | |
| "status": vm.get("status"), | |
| "tags": tag_list, | |
| "ansible_become": "true", | |
| "ansible_host": get_vm_ip(vm.get("node"), vm.get("vmid")), | |
| } | |
| if is_windows: | |
| host_vars.update({ | |
| "ansible_connection": "winrm", | |
| "ansible_winrm_transport": "kerberos", | |
| "ansible_winrm_port": "5985", | |
| "ansible_winrm_kerberos_hostname_override": name, | |
| }) | |
| else: | |
| host_vars.update({ | |
| "ansible_connection": "ssh", | |
| "ansible_user": get_ssh_user(vm.get("node"), vm.get("vmid")), | |
| }) | |
| inventory["_meta"]["hostvars"][name] = host_vars | |
| return inventory | |
| def main(): | |
| """ | |
| Main entry point: Generates and prints the dynamic inventory. | |
| """ | |
| try: | |
| vms = get_vms() | |
| inventory = group_vms_by_tag(vms) | |
| print(json.dumps(inventory, indent=2)) | |
| except Exception as e: | |
| print("Error") | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment