Ansible Custom Action Plugin Development Tutorial
| by jpic | python dev ansible devopsWhen your orchestration logic becomes complicated, you can end up with lots of hardly readable YAML code, and it’s not very nice to maintain, ie. when mutating variables with combinations of template filters, set_fact and the like. The solution is to switch the logic code from YAML to Python, which is an actual programing language.
As you might know: Ansible Modules which are uploaded to the target host, executed there, and have their result returned to Ansible.
Here, we’re presenting Ansible Action Plugins, they execute on the control host, and are able to output data in live, access the whole inventory variables and execute any number of modules for example.
Getting started
Create a new role with an action_plugins directory:
ansible-galaxy role init example
mkdir example/action_plugins
Create an example/action_plugins/example.py script:
from ansible.plugins.action import ActionBase
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
result = dict(changed=False)
return result
Call the plugin from your example/tasks/main.yml script:
---
- name: Example action
example:
And a playbook:
- hosts: localhost
connection: local
roles:
- example
Run it with:
ansible-playbook test.yml
Printing
You can’t use Python’s print() function in Ansible plugins, instead, you must
use the Display object they provide, available in the action object as
self._display.
The print() equivalent is self._display.display(), and it will always
print. But other methods are available, such as self._display.verbose(), self._display.warning(), self._display.error().
The verbose() method allows to pass a verbosity level. If ansible is running
with at most that level of verbosity then the message is printed, otherwise
it isn’t.
For example:
from ansible.plugins.action import ActionBase
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
result = dict(changed=False)
self._display.display('Always printed')
self._display.verbose('Verbose message', caplevel=0)
self._display.verbose('Very verbose message', caplevel=1)
return result
Will produce different outputs depending on the Ansible runtime verbosity.
Without any verbose flag, only display() message is printed:
$ ansible-playbook test.yml
PLAY [localhost] *******************************************************************************************
TASK [example : Example action] ****************************************************************************
Always printed
ok: [localhost]
PLAY RECAP *************************************************************************************************
localhost : ok=2 changed=0 unreachable=0 failed=0 skipped=0 rescued=0 ignored=0
With one -v, only display() and verbose(..., caplevel=0) are printed:
$ ansible-playbook test.yml -v
PLAY [localhost] *******************************************************************************************
TASK [example : Example action] ****************************************************************************
Always printed
Verbose message
ok: [localhost]
PLAY RECAP *************************************************************************************************
localhost : ok=2 changed=0 unreachable=0 failed=0 skipped=0 rescued=0 ignored=0
And with -vv, all our messages are printed:
$ ansible-playbook test.yml -vv
PLAY [localhost] *******************************************************************************************
TASK [example : Example action] ****************************************************************************
task path: /home/jpic/src/action_plugin_example/example/tasks/main.yml:2
Always printed
Verbose message
Very verbose message
ok: [localhost] => {"changed": false}
PLAY RECAP *************************************************************************************************
localhost : ok=2 changed=0 unreachable=0 failed=0 skipped=0 rescued=0 ignored=0
Note that caplevel must be in-between 0 and 5, and that shortcut method names
are available, the above code could have been written with them as such:
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
result = dict(changed=False)
self._display.display('Always printed')
self._display.v('Verbose message')
self._display.vv('Very verbose message')
return result
Note that the verbose printing methods accept a host argument, so that:
self._display.vv('Very verbose message', 'some-host')
Will display as:
<some-host> Very verbose message
Other printing methods can be useful:
self._display.warning(message): to print a warningself._display.error(message): to print an errorself._display.deprecated(message): to print a message in a deprecation warning, to advise the user to upgrade something
Example:
self._display.warning('Example warning')
self._display.error('Example error')
self._display.deprecated('Example deprecation')
Output:
[ERROR]: Example error
[WARNING]: Example warning
[WARNING]: Deprecation warnings can be disabled by setting `deprecation_warnings=False` in ansible.cfg.
[DEPRECATION WARNING]: Example deprecation. This feature will be removed in the future.
Prompting
Since we can’t use print() directly for displaying stuff, we can’t use
input() for reading neither. Instead, we must use the prompt_until method
from the Display object. Example:
from ansible.module_utils._text import to_text
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
your_input = to_text(self._display.prompt_until(
'Type your input here:',
))
self._display.display('You typed: ' + your_input)
return dict(changed=False)
Output:
TASK [example : Example action] ****************************************************************************
Type your input here: My message
You typed: My message
Variables
In task_vars, you get all the variables that would be available in YAML,
such as the facts, example:
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
self._display.display(f'Running for {task_vars["inventory_hostname"]}')
self._display.display(f'BTW I use {task_vars["ansible_os_family"]}')
Will output:
TASK [example : Example action] ********************************************************
Running for localhost
BTW I use Archlinux
Use the debugger (described later) to find out all the available keys in
task_vars. But note that you can access everything, including hostvars,
which contains the variables for every host. As such, you’ve already got your
hands on the wide range of variables that Ansible runtime has to offer which is
already a lot more than what you’re probably used to in modules.
Parameters
Like modules, ansible action plugins may accept arguments, change your
example/tasks/main.yml as such:
- name: Example action
example:
foo: bar
some:
nested: variable
alist:
- item
The variables will be available in self._task.args, dump them as such:
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
self._display.display(repr(self._task.args))
return dict(changed=False)
Output:
TASK [example : Example action] ********************************************************
{'alist': ['item'], 'foo': 'bar', 'some': {'nested': 'variable'}}
ok: [localhost]
Debugging
Prior to Ansible 2.19, you could just use madbg out of the box. Install it with:
pip install madbg
And in your module, add the import madbg; madbg.set_trace() incantation:
from ansible.plugins.action import ActionBase
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
import madbg; madbg.set_trace()
return dict(changed=False)
Run your playbook again:
$ ansible-playbook test.yml
PLAY [localhost] ***********************************************************************
TASK [Gathering Facts] *****************************************************************
ok: [localhost]
TASK [example : Example action] ********************************************************
Waiting for debugger client on 127.0.0.1:3513
In another terminal, run madbg connect:
$ madbg connect
> /home/jpic/src/action_plugin_example/example/action_plugins/example.py(9)run()
7 class ActionModule(ActionBase):
8 def run(self, tmp=None, task_vars=None):
----> 9 import madbg; madbg.set_trace()
10 result = dict(changed=False)
11 self._display.display(cli2.render([*task_vars.keys()]))
ipdb>
And you are now in an IPython debugger. If you don’t know how to use it, see this Python-fu: pdb debugging tutorial.
Debugging with Ansible >= 2.19
However, when trying to use madbg or pdb with ansible-core >= 2.19, the above will result in something like:
[ERROR]: Task failed: I/O operation on closed file
Origin: /home/jpic/src/action_plugin_example/example/tasks/main.yml:2:3
1 ---
2 - name: Example action
^ column 3
fatal: [localhost]: FAILED! => {"changed": false, "msg": "Task failed: I/O operation on closed file"}
And a traceback for the same error in madbg connect. This was introduced by
the patch to Not inherit from
stdio, which makes the worker
detach from stdin.
The solution is to shunt the detach method. First, find where is ansible installed:
$ python -c 'import ansible; print(ansible.__path__)'
['/home/jpic/.local/lib/python3.13/site-packages/ansible']
Find the _detach function there with grep for example:
$ grep -r def._detach /home/jpic/.local/lib/python3.13/site-packages/ansible
/home/jpic/.local/lib/python3.13/site-packages/ansible/executor/process/worker.py: def _detach(self) -> None:
And comment out the line where it does sys.stdin.close(). Then, madbg connect will work again.
Modules
Now that we’ve covered the basics, let’s move onto actually fun stuff: replacing YAML with Python to support larger complexities, which means we’re going to call modules like we do in YAML, but in Python.
In this example, we’ll run the “file” and “copy” ansible modules, equivalent to this YAML:
- copy:
src: /tmp/src
dest: /tmp/dest
In Python:
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
# make sure we have something in /tmp/src
with open('/tmp/src', 'w') as f:
f.write('foo')
# actually execute the copy module
copy_result = self._execute_module(
'copy',
module_args=dict(
src='/tmp/src',
dest='/tmp/dest',
),
task_vars=task_vars,
)
# we could create our own result dict here too
return copy_result
Result:
TASK [example : example] ********************************************************************************
ok: [localhost] => changed=false
checksum: 0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33
dest: /tmp/dest
gid: 1000
group: jpic
md5sum: acbd18db4cc2f85cedef654fccc4a4d8
mode: '0644'
owner: jpic
size: 3
src: /tmp/src
state: file
uid: 1000
Actions
The shell module is actually an action plugin, as we can see in the ansible source code, because it’s in the plugins/action directory.
In this example, we’ll run this simple YAML:
- shell: date
In Python:
class ActionModule(ActionBase):
def run(self, tmp=None, task_vars=None):
# Create a new Ansible Task in Python
new_task = self._task.copy()
# So that we can set our args in Python
new_task.args = dict(_raw_params='date')
# And load the shell action
shell_action = self._shared_loader_obj.action_loader.get(
'ansible.builtin.shell',
task=new_task,
connection=self._connection,
play_context=self._play_context,
loader=self._loader,
templar=self._templar,
shared_loader_obj=self._shared_loader_obj,
)
result = shell_action.run(task_vars=task_vars.copy())
return result
TASK [example : Example action] ********************************************************
changed: [localhost] => changed=true
cmd: date
delta: '0:00:00.007122'
end: '2025-11-05 15:15:14.996482'
msg: ''
rc: 0
start: '2025-11-05 15:15:14.989360'
stderr: ''
stderr_lines: <omitted>
stdout: Wed Nov 5 15:15:14 CET 2025
stdout_lines: <omitted>
cansible action plugin framework
Get more out of action plugins with our ansible action plugin framework
