Writing Custom Modules

What are Custom Modules? Custom modules are user-created Ansible modules that extend Ansible's capabilities beyond the built-in modules. They allow you to automate tasks specific to your environment, integrate with proprietary systems, or implement specialized logic that doesn't exist in standard modules.

Understanding Ansible Modules

What is a Module?

An Ansible module is a standalone script that Ansible executes on target hosts (or locally) to perform a specific task. Key characteristics:

Self-contained: Runs independently with defined inputs and outputs
Idempotent: Can run multiple times with same result
JSON Output: Returns structured JSON data to Ansible
Any Language: Can be written in Python, PowerShell, Ruby, Bash, etc.
Reusable: Can be shared, published, and versioned

When to Write Custom Modules

Create custom modules when:

No existing module performs your required task
You need to interact with proprietary or internal systems
Complex logic is better encapsulated in a module than a playbook
You want to provide a clean interface to other automation engineers
Performance requires compiled or optimized code
You need to integrate with specific APIs or protocols

Before Writing a Module: Check if existing modules can accomplish your task using command, shell, or script modules. If you're using these repeatedly, it's time to create a custom module.

Python Module Structure

Basic Module Template

Python is the recommended language for Ansible modules. Here's a minimal example:

#!/usr/bin/python
# -*- coding: utf-8 -*-

# Copyright: (c) 2025, Your Name 
# GNU General Public License v3.0+

from __future__ import absolute_import, division, print_function
__metaclass__ = type

DOCUMENTATION = r'''
---
module: my_module
short_description: Short description of what module does
description:
    - Detailed description of module functionality.
    - Can span multiple lines.
options:
    name:
        description:
            - Name of the resource to manage.
        required: true
        type: str
    state:
        description:
            - Desired state of the resource.
        choices: [ present, absent ]
        default: present
        type: str
author:
    - Your Name (@yourgithub)
'''

EXAMPLES = r'''
# Create a resource
- name: Create my resource
  my_module:
    name: example
    state: present

# Remove a resource
- name: Remove my resource
  my_module:
    name: example
    state: absent
'''

RETURN = r'''
message:
    description: Output message
    returned: always
    type: str
    sample: "Resource created successfully"
changed:
    description: Whether resource was modified
    returned: always
    type: bool
    sample: true
'''

from ansible.module_utils.basic import AnsibleModule

def run_module():
    # Define module arguments
    module_args = dict(
        name=dict(type='str', required=True),
        state=dict(type='str', default='present', choices=['present', 'absent'])
    )

    # Seed result dict
    result = dict(
        changed=False,
        message=''
    )

    # Create AnsibleModule object
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )

    # Extract parameters
    name = module.params['name']
    state = module.params['state']

    # Check mode - don't make changes
    if module.check_mode:
        result['message'] = f"Would manage resource: {name}"
        module.exit_json(**result)

    # Implement your logic here
    if state == 'present':
        # Create or update resource
        result['changed'] = True
        result['message'] = f"Resource {name} created"
    else:
        # Remove resource
        result['changed'] = True
        result['message'] = f"Resource {name} removed"

    # Return results
    module.exit_json(**result)

def main():
    run_module()

if __name__ == '__main__':
    main()

Required Documentation Sections

All modules must include three documentation strings:

DOCUMENTATION: Module metadata, options, and requirements
EXAMPLES: Real-world usage examples
RETURN: Description of return values and types

Module Development

Argument Specification

Define module parameters with the argument_spec dictionary:

module_args = dict(
    # String parameter (required)
    name=dict(type='str', required=True),

    # String with default value
    description=dict(type='str', default=''),

    # Integer with validation
    port=dict(type='int', default=8080),

    # Boolean parameter
    enabled=dict(type='bool', default=True),

    # List parameter
    tags=dict(type='list', elements='str', default=[]),

    # Dictionary parameter
    config=dict(type='dict', default={}),

    # Choice parameter (enum)
    state=dict(
        type='str',
        default='present',
        choices=['present', 'absent', 'started', 'stopped']
    ),

    # Path parameter (expands ~ and validates)
    config_file=dict(type='path'),

    # Parameter with aliases
    username=dict(type='str', aliases=['user', 'login']),

    # No-log for sensitive data
    password=dict(type='str', no_log=True)
)

Module Initialization

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

    # Mutually exclusive parameters
    mutually_exclusive=[
        ['password', 'ssh_key']
    ],

    # Required together
    required_together=[
        ['username', 'password']
    ],

    # Require one of these
    required_one_of=[
        ['name', 'id']
    ],

    # Required if another parameter is set
    required_if=[
        ['state', 'present', ['name']],
        ['auth_type', 'password', ['password']]
    ]
)

Returning Results

Use exit_json() for success and fail_json() for failures:

# Successful execution
result = {
    'changed': True,
    'message': 'Operation completed',
    'resource_id': '12345',
    'metadata': {
        'created_at': '2025-01-01T00:00:00Z',
        'status': 'active'
    }
}
module.exit_json(**result)

# Failed execution
module.fail_json(
    msg='Failed to create resource',
    error='Connection timeout',
    attempted_host='api.example.com'
)

# Changed but with warnings
result['changed'] = True
result['warnings'] = ['Resource already exists, updated instead']
module.exit_json(**result)

Complete Working Example

User Management Module

#!/usr/bin/python
# -*- coding: utf-8 -*-

from __future__ import absolute_import, division, print_function
__metaclass__ = type

DOCUMENTATION = r'''
---
module: company_user
short_description: Manage users in company system
description:
    - Create, update, or remove users in company user management system.
    - Integrates with internal API endpoint.
options:
    username:
        description: Username to manage
        required: true
        type: str
    email:
        description: User email address
        type: str
    full_name:
        description: User's full name
        type: str
    department:
        description: User's department
        type: str
        choices: [engineering, sales, marketing, operations]
    state:
        description: Desired state
        type: str
        default: present
        choices: [present, absent]
    api_url:
        description: API endpoint URL
        type: str
        default: https://api.company.local
    api_token:
        description: Authentication token
        type: str
        required: true
        no_log: true
author:
    - Your Name (@yourgithub)
'''

EXAMPLES = r'''
- name: Create user
  company_user:
    username: jdoe
    email: jdoe@company.com
    full_name: John Doe
    department: engineering
    state: present
    api_token: "{{ api_token }}"

- name: Remove user
  company_user:
    username: jdoe
    state: absent
    api_token: "{{ api_token }}"
'''

RETURN = r'''
user:
    description: User information
    returned: success
    type: dict
    sample:
        username: jdoe
        email: jdoe@company.com
        department: engineering
        created_at: "2025-01-01T00:00:00Z"
'''

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

def get_user(module, username):
    """Check if user exists."""
    url = f"{module.params['api_url']}/users/{username}"
    headers = {
        'Authorization': f"Bearer {module.params['api_token']}",
        'Content-Type': 'application/json'
    }

    response, info = fetch_url(module, url, headers=headers, method='GET')

    if info['status'] == 200:
        return json.loads(response.read())
    elif info['status'] == 404:
        return None
    else:
        module.fail_json(msg=f"Failed to query user: {info['msg']}")

def create_user(module, username, data):
    """Create new user."""
    url = f"{module.params['api_url']}/users"
    headers = {
        'Authorization': f"Bearer {module.params['api_token']}",
        'Content-Type': 'application/json'
    }

    payload = {
        'username': username,
        'email': data.get('email'),
        'full_name': data.get('full_name'),
        'department': data.get('department')
    }

    response, info = fetch_url(
        module,
        url,
        headers=headers,
        data=json.dumps(payload),
        method='POST'
    )

    if info['status'] in [200, 201]:
        return json.loads(response.read())
    else:
        module.fail_json(msg=f"Failed to create user: {info['msg']}")

def delete_user(module, username):
    """Delete user."""
    url = f"{module.params['api_url']}/users/{username}"
    headers = {
        'Authorization': f"Bearer {module.params['api_token']}",
        'Content-Type': 'application/json'
    }

    response, info = fetch_url(module, url, headers=headers, method='DELETE')

    if info['status'] in [200, 204]:
        return True
    else:
        module.fail_json(msg=f"Failed to delete user: {info['msg']}")

def run_module():
    module_args = dict(
        username=dict(type='str', required=True),
        email=dict(type='str'),
        full_name=dict(type='str'),
        department=dict(
            type='str',
            choices=['engineering', 'sales', 'marketing', 'operations']
        ),
        state=dict(type='str', default='present', choices=['present', 'absent']),
        api_url=dict(type='str', default='https://api.company.local'),
        api_token=dict(type='str', required=True, no_log=True)
    )

    result = dict(
        changed=False,
        user={}
    )

    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True,
        required_if=[
            ['state', 'present', ['email', 'full_name']]
        ]
    )

    username = module.params['username']
    state = module.params['state']

    # Check if user exists
    existing_user = get_user(module, username)

    if state == 'present':
        if existing_user:
            # User exists, no changes needed (could add update logic)
            result['user'] = existing_user
            result['changed'] = False
        else:
            # User doesn't exist, create it
            if not module.check_mode:
                user_data = {
                    'email': module.params['email'],
                    'full_name': module.params['full_name'],
                    'department': module.params['department']
                }
                result['user'] = create_user(module, username, user_data)
            result['changed'] = True

    elif state == 'absent':
        if existing_user:
            # User exists, delete it
            if not module.check_mode:
                delete_user(module, username)
            result['changed'] = True
        else:
            # User doesn't exist, nothing to do
            result['changed'] = False

    module.exit_json(**result)

def main():
    run_module()

if __name__ == '__main__':
    main()

Check Mode Support

Implementing Check Mode

Check mode (dry-run) shows what would change without making changes:

def run_module():
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True  # Enable check mode
    )

    # Check if running in check mode
    if module.check_mode:
        # Perform read-only checks
        result['changed'] = would_change()
        result['message'] = "Would make these changes..."
        module.exit_json(**result)

    # Normal execution - make actual changes
    make_changes()
    module.exit_json(**result)

Testing Modules

Local Testing with ansible Command

# Create library directory for custom modules
mkdir -p library
cp my_module.py library/

# Test with ansible command
ansible localhost -m my_module -a "name=test state=present"

# Test with specific module path
ANSIBLE_LIBRARY=./library ansible localhost -m my_module -a "name=test"

# Test in check mode
ansible localhost -m my_module -a "name=test" --check

Test Playbook

---
# test_module.yml
- name: Test custom module
  hosts: localhost
  gather_facts: false

  tasks:
    - name: Test module with valid parameters
      my_module:
        name: test_resource
        state: present
      register: result

    - name: Display result
      debug:
        var: result

    - name: Verify changed status
      assert:
        that:
          - result.changed
          - result.message is defined

    - name: Test idempotency
      my_module:
        name: test_resource
        state: present
      register: result

    - name: Verify no changes on second run
      assert:
        that:
          - not result.changed

Unit Testing with pytest

# tests/unit/test_my_module.py
import pytest
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
import json

def set_module_args(args):
    """Prepare module arguments."""
    args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
    basic._ANSIBLE_ARGS = to_bytes(args)

def test_module_success():
    """Test successful module execution."""
    set_module_args({
        'name': 'test',
        'state': 'present'
    })

    with pytest.raises(SystemExit) as exc:
        from library import my_module
        my_module.main()

    result = json.loads(exc.value.args[0])
    assert result['changed'] is True

def test_module_check_mode():
    """Test check mode."""
    set_module_args({
        'name': 'test',
        'state': 'present',
        '_ansible_check_mode': True
    })

    # Module should not make changes
    with pytest.raises(SystemExit):
        from library import my_module
        my_module.main()

Module Utilities

Using Common Functions

Ansible provides utility modules for common tasks:

from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.urls import fetch_url
from ansible.module_utils.six.moves.urllib.parse import urlencode
from ansible.module_utils._text import to_native, to_text

# HTTP requests
response, info = fetch_url(
    module,
    url='https://api.example.com/endpoint',
    headers={'Authorization': 'Bearer token'},
    method='GET'
)

# File operations
from ansible.module_utils.common.file import is_executable

# Running commands
rc, stdout, stderr = module.run_command(['ls', '-la'])
if rc != 0:
    module.fail_json(msg=f"Command failed: {stderr}")

Error Handling

try:
    # Risky operation
    result = perform_operation()
except Exception as e:
    module.fail_json(
        msg=f"Operation failed: {to_native(e)}",
        exception=traceback.format_exc()
    )

Module Organization

Project Structure

my_project/
├── library/                    # Custom modules
│   ├── my_module.py
│   ├── another_module.py
│   └── __init__.py
├── module_utils/               # Shared utilities
│   ├── my_helpers.py
│   └── __init__.py
├── plugins/
│   └── modules/               # For collections
│       └── my_module.py
├── tests/
│   ├── unit/
│   │   └── test_my_module.py
│   └── integration/
│       └── test_my_module.yml
└── ansible.cfg

ansible.cfg Configuration

[defaults]
library = ./library
module_utils = ./module_utils

Sharing Modules

Include in Collection

Package modules in an Ansible collection for distribution:

my_namespace/
└── my_collection/
    ├── galaxy.yml
    ├── plugins/
    │   └── modules/
    │       ├── my_module.py
    │       └── another_module.py
    └── docs/
        └── my_module.md

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

Standalone Distribution

# Share via Git repository
git clone https://github.com/yourname/ansible-modules.git
export ANSIBLE_LIBRARY=./ansible-modules/library

# Install from requirements
---
collections:
  - name: my_namespace.my_collection
    source: git+https://github.com/yourname/ansible-collection.git

Best Practices

Module Development Best Practices:

Idempotency: Ensure modules can run multiple times safely
Check Mode: Always implement check mode support
Documentation: Provide comprehensive docs with examples
Error Handling: Catch exceptions and return meaningful errors
Return Values: Document all possible return values
Security: Use no_log for sensitive parameters
Testing: Write unit and integration tests
Python 3: Support Python 3.6+ (Python 2 is deprecated)
Validation: Validate all input parameters
Facts vs Info: Use _facts suffix only for ansible_facts

Advanced Topics

Info and Facts Modules

Modules that gather information should follow naming conventions:

# Info module (returns data in result)
- name: Get user info
  company_user_info:
    username: jdoe
  register: user_info

- debug:
    var: user_info.users

# Facts module (sets ansible_facts)
- name: Gather custom facts
  company_facts:

- debug:
    var: ansible_company_version

Async Module Support

# Mark module as async-compatible
def run_module():
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )

    # Long-running operation
    result = long_operation()

    module.exit_json(**result)

# Use in playbook
- name: Run async operation
  my_long_module:
    name: test
  async: 300
  poll: 10

Common Pitfalls

Issues to Avoid

State Changes: Don't modify state in check mode
Print Statements: Don't use print(); use module.log() or module.debug()
Exit Codes: Always use exit_json() or fail_json(), never sys.exit()
Side Effects: Avoid side effects outside the main logic
Dependencies: Document all required Python packages
Credentials: Never log sensitive data; use no_log=True

Next Steps

Ansible Collections - Package modules in collections
Ansible Galaxy - Publish modules to Galaxy
Testing with Molecule - Test custom modules
Advanced Topics - Module plugins and filters
Try the Playground - Practice using modules

Try in Playground Practice in Labs

Getting Started

Content Distribution

Core Concepts

Platforms

Development

Infrastructure

Operations

Security

Enterprise