Ansible Dynamic Inventory Plugins

Q: How Inventory Plugins Work?

1. User creates a YAML file that declares plugin: myplugin

Introduction

Ansible inventory plugins pull host data from external sources — cloud APIs, CMDBs, spreadsheets, databases, or any system that knows about your infrastructure. While Ansible ships with plugins for AWS, Azure, GCP, NetBox, and others, you can write your own plugin for any data source. This guide covers the inventory plugin API from scratch, with a complete working example.

How Inventory Plugins Work

┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│  Inventory File  │────►│  Inventory Plugin │────►│  Ansible          │
│  (*.yml)         │     │  (Python class)   │     │  (hosts, groups,  │
│  plugin: my_inv  │     │  parse() method   │     │   hostvars)       │
└──────────────────┘     └──────────────────┘     └──────────────────┘

User creates a YAML file that declares plugin: my_plugin
Ansible loads the plugin and calls verify_file() then parse()
Plugin queries the data source and populates hosts, groups, and variables
Ansible uses the populated inventory for playbook execution

Inventory Plugin API

Every inventory plugin is a Python class that inherits from BaseInventoryPlugin:

from ansible.plugins.inventory import BaseInventoryPlugin

class InventoryModule(BaseInventoryPlugin):
    NAME = 'my_namespace.my_collection.my_inventory'

    def verify_file(self, path):
        """Return True if this file should be handled by this plugin."""
        pass

    def parse(self, inventory, loader, path, cache=True):
        """Parse the inventory source and populate inventory."""
        pass

Complete Example: CSV Inventory Plugin

The Plugin

# plugins/inventory/csv_inventory.py
from __future__ import annotations

import csv
import os

from ansible.errors import AnsibleParserError
from ansible.plugins.inventory import BaseInventoryPlugin, Constructable, Cacheable

DOCUMENTATION = """
name: my_namespace.my_collection.csv_inventory
plugin_type: inventory
short_description: CSV file inventory
description:
  - Reads inventory from a CSV file
  - Supports groups, hostvars, and compose
options:
  plugin:
    description: Token that ensures this is a source file for this plugin.
    required: true
    choices: ['my_namespace.my_collection.csv_inventory']
  csv_file:
    description: Path to the CSV file
    required: true
    type: string
  group_by:
    description: CSV column to use for grouping
    type: string
    default: group
extends_documentation_fragment:
  - constructed
"""

EXAMPLES = """
# inventory.csv_inventory.yml
plugin: my_namespace.my_collection.csv_inventory
csv_file: hosts.csv
group_by: role

# Optional compose (from Constructable mixin)
compose:
  ansible_host: ip_address
groups:
  webservers: "'web' in role"
keyed_groups:
  - key: os
    prefix: os
"""


class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
    NAME = 'my_namespace.my_collection.csv_inventory'

    def verify_file(self, path):
        """Return True if the file is a valid inventory source."""
        valid = False
        if super().verify_file(path):
            if path.endswith(('.csv_inventory.yml', '.csv_inventory.yaml')):
                valid = True
        return valid

    def parse(self, inventory, loader, path, cache=True):
        """Parse the CSV file and populate inventory."""
        # Call parent parse
        super().parse(inventory, loader, path, cache)

        # Read plugin configuration from the YAML file
        self._read_config_data(path)

        csv_file = self.get_option('csv_file')
        group_column = self.get_option('group_by')

        # Resolve relative path
        if not os.path.isabs(csv_file):
            csv_file = os.path.join(os.path.dirname(path), csv_file)

        if not os.path.exists(csv_file):
            raise AnsibleParserError(f"CSV file not found: {csv_file}")

        # Check cache
        cache_key = self.get_cache_key(path)
        attempt_to_read_cache = cache and self.use_cache
        rows = None

        if attempt_to_read_cache:
            try:
                rows = self._cache[cache_key]
            except KeyError:
                pass

        if rows is None:
            # Read CSV
            with open(csv_file, 'r') as f:
                reader = csv.DictReader(f)
                rows = list(reader)

            if cache:
                self._cache[cache_key] = rows

        # Populate inventory
        for row in rows:
            hostname = row.get('hostname') or row.get('name') or row.get('host')
            if not hostname:
                continue

            # Add host
            self.inventory.add_host(hostname)

            # Add to group
            if group_column and group_column in row:
                group_name = self._sanitize_group_name(row[group_column])
                self.inventory.add_group(group_name)
                self.inventory.add_child(group_name, hostname)

            # Set host variables from CSV columns
            for key, value in row.items():
                if key not in ('hostname', 'name', 'host'):
                    self.inventory.set_variable(hostname, key, value)

            # Handle compose (from Constructable mixin)
            strict = self.get_option('strict')
            self._set_composite_vars(
                self.get_option('compose'),
                self.inventory.get_host(hostname).get_vars(),
                hostname, strict
            )
            self._add_host_to_composed_groups(
                self.get_option('groups'),
                self.inventory.get_host(hostname).get_vars(),
                hostname, strict
            )
            self._add_host_to_keyed_groups(
                self.get_option('keyed_groups'),
                self.inventory.get_host(hostname).get_vars(),
                hostname, strict
            )

    def _sanitize_group_name(self, name):
        """Convert a string to a valid Ansible group name."""
        import re
        name = re.sub(r'[^A-Za-z0-9_]', '_', name)
        if name[0].isdigit():
            name = '_' + name
        return name

The CSV File

hostname,ip_address,role,os,environment,ssh_port
web01,10.0.1.10,web,ubuntu2404,production,22
web02,10.0.1.11,web,ubuntu2404,production,22
db01,10.0.2.10,database,rockylinux9,production,22
db02,10.0.2.11,database,rockylinux9,production,22
cache01,10.0.3.10,cache,ubuntu2404,staging,22

The Inventory Source File

# inventory.csv_inventory.yml
plugin: my_namespace.my_collection.csv_inventory
csv_file: hosts.csv
group_by: role

compose:
  ansible_host: ip_address
  ansible_port: ssh_port | int

groups:
  production: "environment == 'production'"
  staging: "environment == 'staging'"

keyed_groups:
  - key: os
    prefix: os
  - key: environment
    prefix: env

Test It

# List all hosts and groups
ansible-inventory -i inventory.csv_inventory.yml --list

# Show graph
ansible-inventory -i inventory.csv_inventory.yml --graph

# Output:
# @all:
#   |--@web:
#   |  |--web01
#   |  |--web02
#   |--@database:
#   |  |--db01
#   |  |--db02
#   |--@cache:
#   |  |--cache01
#   |--@production:
#   |  |--web01
#   |  |--web02
#   |  |--db01
#   |  |--db02
#   |--@staging:
#   |  |--cache01

Example: REST API Inventory Plugin

# plugins/inventory/api_inventory.py
import json
from urllib.request import urlopen, Request

from ansible.plugins.inventory import BaseInventoryPlugin, Constructable

DOCUMENTATION = """
name: my_namespace.my_collection.api_inventory
plugin_type: inventory
short_description: REST API inventory
options:
  plugin:
    required: true
    choices: ['my_namespace.my_collection.api_inventory']
  api_url:
    description: Base URL of the API
    required: true
  api_token:
    description: Bearer token for authentication
    type: string
    env:
      - name: INVENTORY_API_TOKEN
extends_documentation_fragment:
  - constructed
"""

class InventoryModule(BaseInventoryPlugin, Constructable):
    NAME = 'my_namespace.my_collection.api_inventory'

    def verify_file(self, path):
        valid = False
        if super().verify_file(path):
            if path.endswith(('.api_inventory.yml', '.api_inventory.yaml')):
                valid = True
        return valid

    def parse(self, inventory, loader, path, cache=True):
        super().parse(inventory, loader, path, cache)
        self._read_config_data(path)

        api_url = self.get_option('api_url')
        token = self.get_option('api_token')

        # Fetch hosts from API
        req = Request(f"{api_url}/hosts")
        if token:
            req.add_header('Authorization', f'Bearer {token}')
        req.add_header('Accept', 'application/json')

        with urlopen(req) as resp:
            hosts = json.loads(resp.read())

        for host in hosts:
            hostname = host['name']
            self.inventory.add_host(hostname)

            # Set variables
            if 'ip' in host:
                self.inventory.set_variable(hostname, 'ansible_host', host['ip'])
            for key, value in host.get('vars', {}).items():
                self.inventory.set_variable(hostname, key, value)

            # Add to groups
            for group in host.get('groups', []):
                self.inventory.add_group(group)
                self.inventory.add_child(group, hostname)

Packaging in a Collection

my_namespace/my_collection/
├── galaxy.yml
├── plugins/
│   └── inventory/
│       ├── csv_inventory.py
│       └── api_inventory.py
├── tests/
│   └── integration/
│       └── targets/
│           └── csv_inventory/
│               ├── hosts.csv
│               └── inventory.csv_inventory.yml
└── README.md

# galaxy.yml
namespace: my_namespace
name: my_collection
version: 1.0.0
type: collection

# Build and install
ansible-galaxy collection build
ansible-galaxy collection install my_namespace-my_collection-1.0.0.tar.gz

Enable Plugin in ansible.cfg

[inventory]
enable_plugins = my_namespace.my_collection.csv_inventory, host_list, auto

Caching

Use the Cacheable mixin for expensive API calls:

# inventory source
plugin: my_namespace.my_collection.api_inventory
api_url: https://cmdb.example.com/api
cache: true
cache_plugin: jsonfile
cache_connection: /tmp/inventory-cache
cache_timeout: 3600

Troubleshooting

"No inventory plugin matched"

Check file extension matches verify_file()
Ensure plugin is in enable_plugins in ansible.cfg
Check plugin: value in YAML matches NAME in class

"Plugin not found"

# List available plugins
ansible-doc -t inventory -l

# Check plugin is installed
ansible-doc -t inventory my_namespace.my_collection.csv_inventory

Conclusion

Ansible inventory plugins are Python classes with two methods: verify_file() to claim a source file, and parse() to populate hosts, groups, and variables. Use the Constructable mixin for compose/groups/keyed_groups support, and Cacheable for API response caching. Package plugins in collections for reuse, and test with ansible-inventory --list before using in playbooks.

Ansible Dynamic Inventory Plugins — Write Your Own

Introduction

How Inventory Plugins Work

Inventory Plugin API

Complete Example: CSV Inventory Plugin

The Plugin

The CSV File

The Inventory Source File

Test It

Example: REST API Inventory Plugin

Packaging in a Collection

Enable Plugin in ansible.cfg

Caching

Troubleshooting

"No inventory plugin matched"

"Plugin not found"

Related Articles

Conclusion