After using Ansible for several years, I hit a wall: I needed to automate a proprietary API that had no existing Ansible module. I could have cobbled together shell scripts, but I knew Ansible was capable of more. That's when I learned to develop custom modules. Later, I discovered the power of plugins to extend Ansible's core functionality. This journey transformed me from an Ansible user to an Ansible developer.
Why Develop Custom Modules and Plugins?
While Ansible ships with thousands of modules and plugins, you may encounter scenarios where custom development makes sense:
When to Create a Custom Module
Interact with proprietary systems or APIs
Encapsulate complex logic into reusable components
Improve performance over shell/command modules
Ensure idempotence for specific operations
Provide better error handling and reporting
When to Create a Custom Plugin
Extend Ansible's core functionality
Create custom filters for data transformation
Implement specialized connection methods
Build custom inventory sources
Add application-specific callbacks
Understanding the Difference: Modules vs Plugins
Modules
Execute on managed nodes (remote systems)
Perform specific tasks (install package, create user, etc.)
Every Ansible module follows a standard structure:
Real-World Example: Custom API Module
Here's a practical example that interacts with a REST API:
Testing Your Module Locally
Create an arguments file for local testing:
Using Your Module in a Playbook
Developing Custom Plugins
Plugin Types
Ansible supports several plugin types:
Filter Plugins - Transform data in templates
Test Plugins - Validate data in conditionals
Lookup Plugins - Retrieve data from external sources
Callback Plugins - Respond to events and control output
Connection Plugins - Connect to remote hosts
Inventory Plugins - Load inventory from external sources
Vars Plugins - Inject additional variables
Action Plugins - Execute logic on control node before modules run
Cache Plugins - Store and retrieve facts
Creating a Custom Filter Plugin
Filter plugins are the most commonly created custom plugins:
Using Custom Filters
Creating a Custom Callback Plugin
Callback plugins control output and respond to events:
Creating a Custom Lookup Plugin
Lookup plugins retrieve data from external sources:
Testing and Validation
Testing Modules with ansible-test
Creating Unit Tests
Integration Testing
Packaging and Distribution
Creating a Collection
Modern Ansible development uses collections:
Building and Publishing
Best Practices
Documentation - Always include DOCUMENTATION, EXAMPLES, and RETURN
Error Handling - Use proper exception handling and meaningful messages
Idempotence - Ensure modules can run multiple times safely
Check Mode - Support --check mode for dry runs
Testing - Write unit and integration tests
Python 3 - Use Python 3 compatibility headers
No Logging - Mark sensitive parameters with no_log: true
Type Hints - Use proper type specifications in argument_spec
Dependencies - Document and check for required libraries
Performance - Be mindful of API rate limits and performance
Conclusion
Developing custom Ansible modules and plugins opens up infinite possibilities for automation. Whether you're integrating with proprietary systems, creating specialized filters, or building custom callbacks, these skills transform you from an Ansible user to an Ansible developer.
Start small, test thoroughly, and contribute back to the community when possible!
# Install development requirements
pip install ansible-core
# Create your project structure
mkdir -p my-ansible-project/{library,module_utils,filter_plugins,callback_plugins}
cd my-ansible-project
# Optional: Clone Ansible for reference and testing tools
git clone https://github.com/ansible/ansible.git
cd ansible
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
source hacking/env-setup
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright: (c) 2026, Your Name <[email protected]>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r'''
---
module: my_custom_module
short_description: Brief description of what this module does
version_added: "1.0.0"
description:
- Longer description explaining what the module does
- Can be multiple lines
- Should explain the primary use cases
options:
name:
description:
- Name of the resource to manage
required: true
type: str
state:
description:
- Desired state of the resource
required: false
type: str
choices: ['present', 'absent']
default: 'present'
api_key:
description:
- API key for authentication
required: true
type: str
no_log: true
author:
- Your Name (@your_github_handle)
requirements:
- python >= 3.6
- requests >= 2.25.0
'''
EXAMPLES = r'''
# Create a resource
- name: Create new resource
my_custom_module:
name: my-resource
state: present
api_key: "{{ vault_api_key }}"
# Remove a resource
- name: Remove resource
my_custom_module:
name: my-resource
state: absent
api_key: "{{ vault_api_key }}"
'''
RETURN = r'''
resource_id:
description: The ID of the created/modified resource
type: str
returned: success
sample: 'resource-12345'
changed:
description: Whether the module made changes
type: bool
returned: always
sample: true
message:
description: Human-readable message about what happened
type: str
returned: always
sample: 'Resource created successfully'
'''
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_native
# Your imports here
try:
import requests
HAS_REQUESTS = True
except ImportError:
HAS_REQUESTS = False
def create_resource(module, name, api_key):
"""Create a new resource"""
# Your logic here
result = {
'changed': True,
'resource_id': 'resource-12345',
'message': f'Resource {name} created successfully'
}
return result
def delete_resource(module, name, api_key):
"""Delete an existing resource"""
# Your logic here
result = {
'changed': True,
'message': f'Resource {name} deleted successfully'
}
return result
def resource_exists(name, api_key):
"""Check if resource already exists"""
# Your logic here
return False
def run_module():
# Define module arguments
module_args = dict(
name=dict(type='str', required=True),
state=dict(type='str', default='present', choices=['present', 'absent']),
api_key=dict(type='str', required=True, no_log=True)
)
# Seed the result dict
result = dict(
changed=False,
resource_id='',
message=''
)
# Create AnsibleModule object
module = AnsibleModule(
argument_spec=module_args,
supports_check_mode=True
)
# Check for required libraries
if not HAS_REQUESTS:
module.fail_json(msg='The requests library is required for this module')
# Extract parameters
name = module.params['name']
state = module.params['state']
api_key = module.params['api_key']
# Check mode - don't make actual changes
if module.check_mode:
exists = resource_exists(name, api_key)
if state == 'present' and not exists:
result['changed'] = True
elif state == 'absent' and exists:
result['changed'] = True
module.exit_json(**result)
# Execute based on desired state
try:
if state == 'present':
if not resource_exists(name, api_key):
result = create_resource(module, name, api_key)
else:
result['message'] = f'Resource {name} already exists'
elif state == 'absent':
if resource_exists(name, api_key):
result = delete_resource(module, name, api_key)
else:
result['message'] = f'Resource {name} does not exist'
except Exception as e:
module.fail_json(msg=to_native(e), **result)
# Success!
module.exit_json(**result)
def main():
run_module()
if __name__ == '__main__':
main()
#!/usr/bin/python
# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = r'''
---
module: api_resource
short_description: Manage resources via REST API
description:
- Create, update, or delete resources via a REST API
- Supports idempotent operations
options:
api_url:
description: Base URL of the API
required: true
type: str
api_token:
description: Authentication token
required: true
type: str
no_log: true
resource_name:
description: Name of the resource
required: true
type: str
resource_type:
description: Type of resource to manage
required: true
type: str
choices: ['database', 'server', 'network']
properties:
description: Resource properties as dictionary
required: false
type: dict
default: {}
state:
description: Desired state
required: false
type: str
choices: ['present', 'absent']
default: 'present'
author:
- Your Name (@yourhandle)
'''
EXAMPLES = r'''
- name: Create database resource
api_resource:
api_url: https://api.example.com/v1
api_token: "{{ vault_token }}"
resource_name: production-db
resource_type: database
properties:
size: large
backup: true
state: present
- name: Delete server resource
api_resource:
api_url: https://api.example.com/v1
api_token: "{{ vault_token }}"
resource_name: old-server
resource_type: server
state: absent
'''
RETURN = r'''
resource:
description: Full resource object returned from API
type: dict
returned: success
sample: {
"id": "res-123",
"name": "production-db",
"type": "database",
"status": "active"
}
'''
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_native
from ansible.module_utils.urls import open_url
import json
class APIResourceManager:
def __init__(self, module):
self.module = module
self.api_url = module.params['api_url']
self.api_token = module.params['api_token']
self.headers = {
'Authorization': f'Bearer {self.api_token}',
'Content-Type': 'application/json'
}
def _make_request(self, method, endpoint, data=None):
"""Make HTTP request to API"""
url = f"{self.api_url.rstrip('/')}/{endpoint.lstrip('/')}"
try:
if method == 'GET':
response = open_url(url, headers=self.headers, method=method)
else:
response = open_url(
url,
headers=self.headers,
method=method,
data=json.dumps(data) if data else None
)
return json.loads(response.read())
except Exception as e:
self.module.fail_json(msg=f'API request failed: {to_native(e)}')
def get_resource(self, name, resource_type):
"""Get resource by name and type"""
endpoint = f'resources/{resource_type}/{name}'
try:
return self._make_request('GET', endpoint)
except:
return None
def create_resource(self, name, resource_type, properties):
"""Create new resource"""
endpoint = f'resources/{resource_type}'
data = {
'name': name,
'properties': properties
}
return self._make_request('POST', endpoint, data)
def update_resource(self, name, resource_type, properties):
"""Update existing resource"""
endpoint = f'resources/{resource_type}/{name}'
data = {'properties': properties}
return self._make_request('PUT', endpoint, data)
def delete_resource(self, name, resource_type):
"""Delete resource"""
endpoint = f'resources/{resource_type}/{name}'
return self._make_request('DELETE', endpoint)
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_type=dict(type='str', required=True, choices=['database', 'server', 'network']),
properties=dict(type='dict', default={}),
state=dict(type='str', default='present', choices=['present', 'absent'])
)
result = dict(changed=False, resource={})
module = AnsibleModule(
argument_spec=module_args,
supports_check_mode=True
)
manager = APIResourceManager(module)
name = module.params['resource_name']
resource_type = module.params['resource_type']
properties = module.params['properties']
state = module.params['state']
# Get current resource state
existing_resource = manager.get_resource(name, resource_type)
# Check mode
if module.check_mode:
if state == 'present' and not existing_resource:
result['changed'] = True
elif state == 'absent' and existing_resource:
result['changed'] = True
module.exit_json(**result)
# Execute desired state
try:
if state == 'present':
if existing_resource:
# Check if update needed
if existing_resource.get('properties') != properties:
result['resource'] = manager.update_resource(name, resource_type, properties)
result['changed'] = True
else:
result['resource'] = existing_resource
else:
result['resource'] = manager.create_resource(name, resource_type, properties)
result['changed'] = True
elif state == 'absent':
if existing_resource:
manager.delete_resource(name, resource_type)
result['changed'] = True
except Exception as e:
module.fail_json(msg=to_native(e), **result)
module.exit_json(**result)
def main():
run_module()
if __name__ == '__main__':
main()