Ansible Action Plugins — Custom Task Execution Logic

Introduction

Action plugins run on the controller node and execute before the module runs on the target host. They control how tasks are dispatched — handling file transfers, template rendering, connection management, and module invocation. The built-in copy, template, fetch, and script modules are all implemented as action plugins.

When you need behavior that happens on the controller (not the target), or need to modify how a module is called, action plugins are the right tool.

How Action Plugins Work

Task Execution Flow:
  1. Ansible reads task from playbook
  2. Looks for action plugin matching module name
  3. If found → action plugin runs on CONTROLLER
     └─ Action plugin may call module on TARGET
  4. If not found → default action plugin runs module on TARGET

Plugin Directory Structure

my_collection/
├── plugins/
│   └── action/
│       ├── __init__.py
│       └── my_action.py
└── plugins/
    └── modules/
        └── my_action.py    # Companion module (optional)

Or in a role:

my_role/
├── action_plugins/
│   └── my_action.py
└── library/
    └── my_action.py

Basic Action Plugin

# plugins/action/validate_and_deploy.py
from ansible.plugins.action import ActionBase
from ansible.errors import AnsibleError


class ActionModule(ActionBase):
    """Action plugin that validates config before deploying."""

    def run(self, tmp=None, task_vars=None):
        # Always call super first
        result = super().run(tmp, task_vars)
        task_vars = task_vars or {}

        # Get task arguments
        src = self._task.args.get('src')
        dest = self._task.args.get('dest')
        validate_cmd = self._task.args.get('validate')

        if not src or not dest:
            return {'failed': True, 'msg': 'src and dest are required'}

        # Step 1: Template the source file on controller
        try:
            source_full = self._find_needle('files', src)
            with open(source_full, 'r') as f:
                content = f.read()
        except AnsibleError as e:
            return {'failed': True, 'msg': str(e)}

        # Step 2: Transfer file to target
        tmp_dest = self._connection._shell.join_path(
            self._make_tmp_path(), 'validate_file'
        )
        self._transfer_data(tmp_dest, content)

        # Step 3: Validate on target (if validate command given)
        if validate_cmd:
            validate_full = validate_cmd.replace('%s', tmp_dest)
            validate_result = self._low_level_execute_command(validate_full)
            if validate_result['rc'] != 0:
                return {
                    'failed': True,
                    'msg': f'Validation failed: {validate_result["stderr"]}',
                }

        # Step 4: Move to final destination using copy module
        module_args = {
            'src': tmp_dest,
            'dest': dest,
            'remote_src': True,
        }
        result = self._execute_module(
            module_name='ansible.builtin.copy',
            module_args=module_args,
            task_vars=task_vars,
        )

        return result

Using Built-In Methods

Key ActionBase Methods

class ActionModule(ActionBase):
    def run(self, tmp=None, task_vars=None):
        result = super().run(tmp, task_vars)

        # Execute a module on the remote host
        result = self._execute_module(
            module_name='ansible.builtin.command',
            module_args={'_raw_params': 'hostname'},
            task_vars=task_vars,
        )

        # Run a raw command on the remote
        cmd_result = self._low_level_execute_command('uname -a')

        # Transfer a file to the remote
        self._transfer_file('/local/path', '/remote/path')

        # Transfer string data to remote
        self._transfer_data('/remote/path', 'file content here')

        # Create temporary directory on remote
        tmp_path = self._make_tmp_path()

        # Template a string with Jinja2
        templated = self._templar.template('{{ ansible_hostname }}')

        # Find a file in role's files/ directory
        path = self._find_needle('files', 'myconfig.conf')

        # Find a template
        tmpl = self._find_needle('templates', 'myconfig.conf.j2')

        return result

Action Plugin with Check Mode

class ActionModule(ActionBase):

    def run(self, tmp=None, task_vars=None):
        result = super().run(tmp, task_vars)
        check_mode = self._play_context.check_mode

        # Get current state
        current = self._execute_module(
            module_name='ansible.builtin.stat',
            module_args={'path': self._task.args['dest']},
            task_vars=task_vars,
        )

        needs_change = not current.get('stat', {}).get('exists', False)

        if check_mode:
            result['changed'] = needs_change
            result['msg'] = 'Would create file' if needs_change else 'File exists'
            return result

        if needs_change:
            # Actually make the change
            result = self._execute_module(
                module_name='ansible.builtin.copy',
                module_args={
                    'content': self._task.args['content'],
                    'dest': self._task.args['dest'],
                },
                task_vars=task_vars,
            )

        return result

Multi-Host Coordination

class ActionModule(ActionBase):
    """Action plugin that coordinates across hosts."""

    TRANSFERS_FILES = False

    def run(self, tmp=None, task_vars=None):
        result = super().run(tmp, task_vars)

        # Access host variables
        hostname = task_vars.get('inventory_hostname')
        hostvars = task_vars.get('hostvars', {})
        groups = task_vars.get('groups', {})

        # Get variable from another host
        db_host = groups.get('databases', ['db1'])[0]
        db_ip = hostvars.get(db_host, {}).get('ansible_host', db_host)

        # Use it in module execution
        result = self._execute_module(
            module_name='ansible.builtin.template',
            module_args={
                'src': 'app.conf.j2',
                'dest': '/etc/app/config.yml',
            },
            task_vars={**task_vars, 'db_server_ip': db_ip},
        )

        return result

Testing Action Plugins

# tests/unit/plugins/action/test_my_action.py
import pytest
from unittest.mock import MagicMock, patch
from plugins.action.my_action import ActionModule


@pytest.fixture
def action_module():
    task = MagicMock()
    task.args = {'src': 'test.conf', 'dest': '/etc/test.conf'}
    connection = MagicMock()
    play_context = MagicMock()
    play_context.check_mode = False

    action = ActionModule(
        task=task,
        connection=connection,
        play_context=play_context,
        loader=MagicMock(),
        templar=MagicMock(),
        shared_loader_obj=MagicMock(),
    )
    return action


def test_requires_src_and_dest(action_module):
    action_module._task.args = {}
    result = action_module.run(task_vars={})
    assert result['failed'] is True
    assert 'required' in result['msg']

Troubleshooting

Plugin not found:

# Check plugin search path
ansible-config dump | grep ACTION_PLUGINS

# Verify file permissions
ls -la plugins/action/my_action.py

# Class must be named ActionModule
grep "class ActionModule" plugins/action/my_action.py

Module vs action plugin confusion:

Action plugin filename must match the module name
If my_module module exists, my_module.py action plugin wraps it automatically

Conclusion

Action plugins give you full control over task execution on the controller side. Use them when you need to validate before deploying, coordinate across hosts, or implement complex transfer logic. Combined with custom modules, they let you extend Ansible's task execution model to handle any workflow.

Ansible Action Plugins — Custom Task Execution Logic

Ansible Action Plugins — Custom Task Execution Logic

Introduction

How Action Plugins Work

Plugin Directory Structure

Basic Action Plugin

Using Built-In Methods

Key ActionBase Methods

Action Plugin with Check Mode

Multi-Host Coordination

Testing Action Plugins

Troubleshooting

Related Articles

Conclusion