Write Custom Ansible Modules in Python

When existing Ansible modules don't meet your needs, you can write custom modules in Python. Custom modules let you interface with proprietary APIs, implement complex business logic, or create abstractions specific to your infrastructure. This guide covers building, testing, and distributing custom Ansible modules.

Module Anatomy

An Ansible module is a Python script that receives JSON input (arguments), performs an action, and returns JSON output. Modules must be idempotent — running them multiple times should produce the same result.

Basic Module Template

#!/usr/bin/python
# library/my_custom_module.py

from ansible.module_utils.basic import AnsibleModule

DOCUMENTATION = r'''
---
module: my_custom_module
short_description: Manage custom application configuration
description:
    - Creates, updates, or removes custom application configuration files.
options:
    name:
        description: Configuration name
        required: true
        type: str
    state:
        description: Desired state
        choices: ['present', 'absent']
        default: present
        type: str
    value:
        description: Configuration value
        required: false
        type: str
    path:
        description: Path to configuration directory
        default: /etc/myapp/conf.d
        type: str
author:
    - Your Name (@github_handle)
'''

EXAMPLES = r'''
- name: Set application config
  my_custom_module:
    name: database_url
    value: "postgresql://localhost:5432/mydb"
    state: present

- name: Remove a config entry
  my_custom_module:
    name: old_setting
    state: absent
'''

RETURN = r'''
path:
    description: Full path to the config file
    type: str
    returned: always
changed:
    description: Whether the config was modified
    type: bool
    returned: always
'''

import os
import json


def run_module():
    # Define module arguments
    module_args = dict(
        name=dict(type='str', required=True),
        state=dict(type='str', default='present', choices=['present', 'absent']),
        value=dict(type='str', required=False, default=None),
        path=dict(type='str', default='/etc/myapp/conf.d'),
    )

    # Initialize the module
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True,
        required_if=[
            ('state', 'present', ['value']),
        ],
    )

    # Get parameters
    name = module.params['name']
    state = module.params['state']
    value = module.params['value']
    config_dir = module.params['path']
    config_file = os.path.join(config_dir, f"{name}.conf")

    result = dict(
        changed=False,
        path=config_file,
    )

    # Check current state
    file_exists = os.path.isfile(config_file)
    current_value = None
    if file_exists:
        with open(config_file, 'r') as f:
            current_value = f.read().strip()

    # Determine if changes are needed
    if state == 'present':
        if not file_exists or current_value != value:
            result['changed'] = True
            if not module.check_mode:
                os.makedirs(config_dir, exist_ok=True)
                with open(config_file, 'w') as f:
                    f.write(value)
            result['diff'] = {
                'before': current_value or '',
                'after': value,
            }
    elif state == 'absent':
        if file_exists:
            result['changed'] = True
            if not module.check_mode:
                os.remove(config_file)

    module.exit_json(**result)


def main():
    run_module()


if __name__ == '__main__':
    main()

Module for API Integration

#!/usr/bin/python
# library/api_service.py

from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.urls import fetch_url
import json


def run_module():
    module_args = dict(
        api_url=dict(type='str', required=True),
        api_token=dict(type='str', required=True, no_log=True),
        resource_name=dict(type='str', required=True),
        resource_data=dict(type='dict', required=False, default={}),
        state=dict(type='str', default='present', choices=['present', 'absent']),
    )

    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True,
    )

    api_url = module.params['api_url'].rstrip('/')
    api_token = module.params['api_token']
    resource_name = module.params['resource_name']
    resource_data = module.params['resource_data']
    state = module.params['state']

    headers = {
        'Authorization': f'Bearer {api_token}',
        'Content-Type': 'application/json',
    }

    result = dict(changed=False)

    # Check if resource exists
    response, info = fetch_url(
        module,
        f"{api_url}/resources/{resource_name}",
        headers=headers,
        method='GET',
    )

    exists = info['status'] == 200
    current_data = json.loads(response.read()) if exists else None

    if state == 'present':
        if not exists:
            result['changed'] = True
            if not module.check_mode:
                response, info = fetch_url(
                    module,
                    f"{api_url}/resources",
                    headers=headers,
                    method='POST',
                    data=json.dumps({
                        'name': resource_name,
                        **resource_data
                    }),
                )
                if info['status'] not in (200, 201):
                    module.fail_json(msg=f"API error: {info['status']}")
                result['resource'] = json.loads(response.read())
        elif current_data and resource_data:
            # Check if update needed
            needs_update = any(
                current_data.get(k) != v
                for k, v in resource_data.items()
            )
            if needs_update:
                result['changed'] = True
                if not module.check_mode:
                    response, info = fetch_url(
                        module,
                        f"{api_url}/resources/{resource_name}",
                        headers=headers,
                        method='PUT',
                        data=json.dumps(resource_data),
                    )
                    result['resource'] = json.loads(response.read())

    elif state == 'absent' and exists:
        result['changed'] = True
        if not module.check_mode:
            fetch_url(
                module,
                f"{api_url}/resources/{resource_name}",
                headers=headers,
                method='DELETE',
            )

    module.exit_json(**result)


def main():
    run_module()

if __name__ == '__main__':
    main()

Using Custom Modules

# Place modules in your project's library/ directory
# or set ANSIBLE_LIBRARY in ansible.cfg

# ansible.cfg
[defaults]
library = ./library
module_utils = ./module_utils

# Use in playbooks
- name: Configure application
  hosts: webservers
  tasks:
    - name: Set database URL
      my_custom_module:
        name: database_url
        value: "postgresql://db.example.com:5432/production"
        state: present

    - name: Create API resource
      api_service:
        api_url: "https://api.example.com/v1"
        api_token: "{{ vault_api_token }}"
        resource_name: "my-service"
        resource_data:
          tier: production
          replicas: 3

Testing Modules

# tests/test_my_module.py
import pytest
import json
from unittest.mock import patch, MagicMock
from library.my_custom_module import run_module

@pytest.fixture
def module_args():
    return {
        'name': 'test_config',
        'state': 'present',
        'value': 'test_value',
        'path': '/tmp/test_config',
    }

def test_create_config(module_args, tmp_path):
    module_args['path'] = str(tmp_path)
    with patch('library.my_custom_module.AnsibleModule') as mock_module:
        mock_module.return_value.params = module_args
        mock_module.return_value.check_mode = False
        run_module()
        mock_module.return_value.exit_json.assert_called_once()
        call_args = mock_module.return_value.exit_json.call_args
        assert call_args[1]['changed'] == True

# Run tests
# pytest tests/ -v

Best Practices

• Always support check mode: Use supports_check_mode=True and skip changes when module.check_mode
• Be idempotent: Check current state before making changes — don't change what doesn't need changing
• Use no_log=True for sensitive parameters like passwords and API tokens
• Return meaningful diffs: Include before and after in results for --diff mode
• Use module.fail_json() for errors with descriptive messages
• Write documentation strings (DOCUMENTATION, EXAMPLES, RETURN) for ansible-doc integration
• Add to a collection for distribution and version management