Ansible Custom Modules — Write Your Own in Python

Introduction

When Ansible's 6,000+ built-in modules don't cover your specific use case, writing a custom module is the solution. Custom modules let you extend Ansible with your own logic while maintaining idempotency, check mode support, and seamless integration with playbooks.

This guide walks you through creating production-quality custom modules in Python, from basic structure to testing and distribution.

Why Write Custom Modules?

Use cases for custom modules:

• Interact with proprietary APIs not covered by existing modules
• Wrap complex shell commands in an idempotent interface
• Enforce organization-specific business logic
• Create simplified interfaces for common multi-step operations
• Integrate with internal tools and services

Module Structure

Every Ansible module follows a standard structure:

```python #!/usr/bin/python

-- coding: utf-8 --

DOCUMENTATION = r'''

module: my_custom_module short_description: Description of what the module does version_added: "1.0.0" description:

• Detailed description of the module functionality.
• Supports check mode and diff mode. 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 force: description:
  • Force the operation even if the resource exists. type: bool default: false author:
• Your Name (@github_handle) '''

EXAMPLES = r'''

• name: Create a resource my_custom_module: name: my_resource state: present

• name: Remove a resource my_custom_module: name: my_resource state: absent force: true '''

RETURN = r''' message: description: Status message from the module. returned: always type: str sample: "Resource created successfully" changed: description: Whether any changes were made. returned: always type: bool original_state: description: State of the resource before module execution. returned: always type: dict '''

from ansible.module_utils.basic import AnsibleModule

def run_module(): # Define the argument spec module_args = dict( name=dict(type='str', required=True), state=dict(type='str', default='present', choices=['present', 'absent']), force=dict(type='bool', default=False), )

# Initialize the module
module = AnsibleModule(
    argument_spec=module_args,
    supports_check_mode=True,
)

# Collect parameters
name = module.params['name']
state = module.params['state']
force = module.params['force']

# Initialize result
result = dict(
    changed=False,
    message='',
    original_state={},
)

# Check current state
current_state = get_resource_state(name)
result['original_state'] = current_state

# Determine if changes are needed
if state == 'present' and not current_state.get('exists'):
    result['changed'] = True
    result['message'] = f'Resource {name} will be created'
elif state == 'absent' and current_state.get('exists'):
    result['changed'] = True
    result['message'] = f'Resource {name} will be removed'

# Check mode — report what would change without making changes
if module.check_mode:
    module.exit_json(**result)

# Apply changes
if result['changed']:
    try:
        if state == 'present':
            create_resource(name)
            result['message'] = f'Resource {name} created successfully'
        elif state == 'absent':
            delete_resource(name, force=force)
            result['message'] = f'Resource {name} removed successfully'
    except Exception as e:
        module.fail_json(msg=f'Operation failed: {str(e)}', **result)

module.exit_json(**result)

def get_resource_state(name): """Check if the resource exists.""" # Replace with your actual logic return {'exists': False, 'name': name}

def create_resource(name): """Create the resource.""" # Replace with your actual creation logic pass

def delete_resource(name, force=False): """Delete the resource.""" # Replace with your actual deletion logic pass

def main(): run_module()

if name == 'main': main() ```

Argument Specification

The `argument_spec` dictionary defines module parameters:

Common Parameter Types

Advanced Parameter Features

```python module_args = dict( # Required parameter name=dict(type='str', required=True),

# Parameter with choices
state=dict(type='str', choices=['present', 'absent', 'restarted']),

# Parameter with aliases
dest=dict(type='path', aliases=['destination', 'target']),

# Mutually exclusive groups
# (defined separately in AnsibleModule)

# No-log parameter (passwords, secrets)
password=dict(type='str', no_log=True),

# Nested spec (sub-options)
config=dict(
    type='dict',
    options=dict(
        host=dict(type='str', default='localhost'),
        port=dict(type='int', default=5432),
    ),
),

)

module = AnsibleModule( argument_spec=module_args, mutually_exclusive=[ ('file', 'content'), ], required_one_of=[ ('file', 'content'), ], required_together=[ ('username', 'password'), ], supports_check_mode=True, ) ```

Check Mode Support

Check mode (`--check`) lets users preview changes without applying them:

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

Your logic determines what WOULD change

result['changed'] = needs_update(current, desired)

Exit early in check mode — don't make changes

if module.check_mode: module.exit_json(**result)

Actually apply changes

apply_changes(current, desired) module.exit_json(**result) ```

Diff Mode Support

Diff mode (`--diff`) shows what changed:

```python if module._diff: result['diff'] = dict( before=current_config, after=desired_config, ) ```

Return Values

Modules communicate results through `exit_json` and `fail_json`:

```python

Success

module.exit_json( changed=True, msg='Resource updated', resource_id='abc123', diff={'before': old_state, 'after': new_state}, )

Failure

module.fail_json( msg='Cannot connect to API', status_code=503, response_body=response.text, ) ```

Module Placement

Place custom modules in one of these locations:

```

Project-level (recommended)

my_project/ ├── library/ │ └── my_custom_module.py ├── playbooks/ │ └── site.yml └── ansible.cfg

In ansible.cfg

[defaults] library = ./library

In a collection

collections/ └── ansible_collections/ └── myorg/ └── mycollection/ └── plugins/ └── modules/ └── my_custom_module.py ```

Running Shell Commands

Use `module.run_command()` instead of `subprocess`:

```python def get_package_version(module, package_name): """Get installed package version using rpm.""" rc, stdout, stderr = module.run_command( ['rpm', '-q', '--queryformat', '%{VERSION}', package_name] )

if rc != 0:
    return None
return stdout.strip()

```

Making HTTP Requests

Use `ansible.module_utils.urls` for HTTP requests:

```python from ansible.module_utils.urls import open_url import json

def api_request(module, url, method='GET', data=None): """Make an API request.""" headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {module.params["api_token"]}', }

try:
    response = open_url(
        url,
        method=method,
        data=json.dumps(data) if data else None,
        headers=headers,
        validate_certs=module.params.get('validate_certs', True),
    )
    return json.loads(response.read())
except Exception as e:
    module.fail_json(msg=f'API request failed: {str(e)}')

```

Testing Your Module

Unit Testing

```python

tests/unit/plugins/modules/test_my_module.py

import pytest from unittest.mock import patch, MagicMock from ansible_collections.myorg.mycollection.plugins.modules import my_custom_module

@pytest.fixture def module_args(): return { 'name': 'test_resource', 'state': 'present', 'force': False, }

def test_create_resource(module_args): with patch.object(my_custom_module, 'get_resource_state') as mock_state: mock_state.return_value = {'exists': False} # Test module logic ```

Integration Testing with ansible-test

```yaml

tests/integration/targets/my_custom_module/tasks/main.yml

• name: Create resource (check mode) my_custom_module: name: test_resource state: present check_mode: true register: check_result

• name: Verify check mode ansible.builtin.assert: that: - check_result is changed

• name: Create resource my_custom_module: name: test_resource state: present register: create_result

• name: Verify creation ansible.builtin.assert: that: - create_result is changed - create_result.message == "Resource test_resource created successfully"

• name: Create resource again (idempotency) my_custom_module: name: test_resource state: present register: idem_result

• name: Verify idempotency ansible.builtin.assert: that: - idem_result is not changed ```

Run tests

```bash

Sanity tests

ansible-test sanity --docker

Unit tests

ansible-test units --docker

Integration tests

ansible-test integration my_custom_module --docker ```

Real-World Example: Managing API Resources

```python #!/usr/bin/python

-- coding: utf-8 --

DOCUMENTATION = r'''

module: api_endpoint short_description: Manage API endpoints in a service registry description:

• Create, update, or delete API endpoints in a service registry.
• Supports check mode and diff mode. options: url: description: Registry API URL. required: true type: str api_token: description: Authentication token. required: true type: str no_log: true name: description: Endpoint name. required: true type: str endpoint_url: description: URL of the endpoint. type: str state: description: Desired state. choices: ['present', 'absent'] default: present type: str validate_certs: description: Validate SSL certificates. type: bool default: true author:
• Luca Berton (@lucab85) '''

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

class ServiceRegistry: def init(self, module): self.module = module self.base_url = module.params['url'].rstrip('/') self.token = module.params['api_token'] self.validate_certs = module.params['validate_certs']

def _request(self, path, method='GET', data=None):
    url = f'{self.base_url}{path}'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {self.token}',
    }
    try:
        resp = open_url(
            url, method=method,
            data=json.dumps(data) if data else None,
            headers=headers,
            validate_certs=self.validate_certs,
        )
        return json.loads(resp.read())
    except Exception as e:
        if '404' in str(e):
            return None
        self.module.fail_json(msg=f'API error: {e}')

def get_endpoint(self, name):
    return self._request(f'/endpoints/{name}')

def create_endpoint(self, name, url):
    return self._request('/endpoints', 'POST', {'name': name, 'url': url})

def update_endpoint(self, name, url):
    return self._request(f'/endpoints/{name}', 'PUT', {'url': url})

def delete_endpoint(self, name):
    return self._request(f'/endpoints/{name}', 'DELETE')

def main(): module = AnsibleModule( argument_spec=dict( url=dict(type='str', required=True), api_token=dict(type='str', required=True, no_log=True), name=dict(type='str', required=True), endpoint_url=dict(type='str'), state=dict(type='str', default='present', choices=['present', 'absent']), validate_certs=dict(type='bool', default=True), ), required_if=[ ('state', 'present', ['endpoint_url']), ], supports_check_mode=True, )

registry = ServiceRegistry(module)
name = module.params['name']
state = module.params['state']
endpoint_url = module.params.get('endpoint_url')

current = registry.get_endpoint(name)
result = dict(changed=False, endpoint={})

if state == 'present':
    if current is None:
        result['changed'] = True
        if not module.check_mode:
            result['endpoint'] = registry.create_endpoint(name, endpoint_url)
    elif current.get('url') != endpoint_url:
        result['changed'] = True
        if module._diff:
            result['diff'] = {
                'before': {'url': current.get('url')},
                'after': {'url': endpoint_url},
            }
        if not module.check_mode:
            result['endpoint'] = registry.update_endpoint(name, endpoint_url)
elif state == 'absent':
    if current is not None:
        result['changed'] = True
        if not module.check_mode:
            registry.delete_endpoint(name)

module.exit_json(**result)

if name == 'main': main() ```

Common Mistakes

1. Not supporting check mode: Always set `supports_check_mode=True` and handle `module.check_mode`.

2. Not being idempotent: Check the current state before making changes. Only report `changed=True` when something actually changes.

3. Using subprocess instead of run_command: `module.run_command()` handles PATH, encoding, and error reporting correctly.

4. Forgetting no_log for secrets: Always mark password/token parameters with `no_log=True`.

5. Not validating input properly: Use `argument_spec` constraints (choices, required_if, mutually_exclusive) instead of manual validation.

Troubleshooting

Module not found: ```bash

Check module search path

ansible-config dump | grep DEFAULT_MODULE_PATH

Verify module is importable

python3 -c "import my_custom_module" ```

Debugging module output: ```bash

Run module directly (for testing)

python3 my_custom_module.py /tmp/args.json

Enable verbose output

ansible-playbook site.yml -vvv ```

Related Articles

• Ansible Filter Plugins Guide
• Ansible Callback Plugins
• Ansible Lookup Plugins
• Ansible Collections Guide
• Ansible Testing with Molecule

Conclusion

Custom modules are Ansible's ultimate extension point. Follow these practices:

1. Always support check mode and diff mode
2. Be idempotent — check before changing
3. Use `argument_spec` for input validation
4. Handle errors gracefully with `fail_json`
5. Write DOCUMENTATION, EXAMPLES, and RETURN docstrings
6. Test with ansible-test for sanity, unit, and integration coverage

Start with simple modules wrapping existing APIs, then build complexity as needed.