Skip to content

Instantly share code, notes, and snippets.

@bergelfs

bergelfs/README.md

Last active December 8, 2025 17:10
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save bergelfs/546de362b4e81847673dd70fb8d298e3 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save bergelfs/546de362b4e81847673dd70fb8d298e3 to your computer and use it in GitHub Desktop.
Download ZIP
Ansible NTFY Callback Plugin
Raw
README.md

Ansible NTFY Callback Plugin

This setup lets you run your Ansible playbook automatically (for example, via a daily cron job) and receive notifications from your ntfy.sh server about the status.

To do this, the we can use the callback plugin Ansible feature.

You can run the playbook and set the required environment variables however you like.
I chose to do it in a bash script ansible-playbook.sh, and run that script in my cron job.

Example Notifications

⚙️ ✅ Ansible Complete - 2 Changes
Changed Tasks:
[inventory_hostname] role1 : Task1
[inventory_hostname] role2 : Task2
Duration: 0:05:00

⚙️ ❌ ⚠️ Ansible Failed!
Unreachable Hosts:
inventory_hostname
Duration: 0:04:20

Setup

  1. Place ntfy_callback.py inside the callback_plugins directory in your Ansible project.
  2. Make sure the python requests package is installed:
    • With a package manager:
      sudo apt install python3-requests
    • Or with pip:
      pip3 install requests
  3. By default, the plugin only sends notifications on changes or errors.
    • To notify every run, remove lines 80–81 in ntfy_callback.py.
  4. Enable the plugin either by:
    • setting environment variable:
      ANSIBLE_CALLBACKS_ENABLED="ntfy_callback"
    • adding to your playbook ansible.cfg file:
      callbacks_enabled = ntfy_callback
  5. Define ntfy environment variables according to your setup:
 NTFY_URL="https://<ntfy.sh/topic>"
 NTFY_USER="<myuser>"
 NTFY_PASSWORD="<mypassword>"

A password is required. If you prefer not to use one, you can likely remove it from the script. I have not tested this.

  1. Optionally, define a log file:
LOG_FILE="/var/log/ansible/<playbook>.log"
  • Make sure this directory exists and is writable by the user running ansible-playbook.
  • It’s recommended to use logrotate to manage file sizes over time.
Raw
ansible-playbook.sh
#!/bin/bash
PLAYBOOK_PATH="<playbook path>"
LOG_FILE="/var/log/ansible/<playbook>.log"
export ANSIBLE_CALLBACKS_ENABLED="ntfy_callback"
export NTFY_URL="https://<ntfy.sh/topic>"
export NTFY_USER="<myuser>"
export NTFY_PASSWORD="<mypassword>"
START_TIME=$(date '+%Y-%m-%d %H:%M:%S')
echo "[$START_TIME] Starting Ansible playbook execution" >> "$LOG_FILE"
cd $PLAYBOOK_PATH
ansible-playbook main.yml --diff 2>&1 | tee -a "$LOG_FILE"
Raw
ntfy_callback.py
import requests
import base64
import os
from datetime import datetime
from ansible.plugins.callback import CallbackBase
class CallbackModule(CallbackBase):
CALLBACK_VERSION = 2.0
CALLBACK_TYPE = 'notification'
CALLBACK_NAME = 'ntfy_callback'
CALLBACK_NEEDS_ENABLED = True
def __init__(self, *args, **kwargs):
super(CallbackModule, self).__init__()
self.ntfy_url = os.environ.get('NTFY_URL')
self.ntfy_user = os.environ.get('NTFY_USER')
self.ntfy_password = os.environ.get('NTFY_PASSWORD')
assert all([self.ntfy_url, self.ntfy_user, self.ntfy_password]), (
"Environment variables NTFY_URL, NTFY_USER, and NTFY_PASSWORD must all be set"
)
self.auth_header = base64.b64encode(f"{self.ntfy_user}:{self.ntfy_password}".encode()).decode()
self.changed_tasks = []
self.failed_tasks = []
self.unreachable_hosts = []
def send_notification(self, title, message, priority="default", tags="gear"):
try:
headers = {
"Authorization": f"Basic {self.auth_header}",
"Title": title,
"Priority": priority,
"Tags": tags,
"Content-Type": "text/plain; charset=utf-8"
}
response = requests.post(
self.ntfy_url,
data=message.encode('utf-8'),
headers=headers,
timeout=10
)
response.raise_for_status()
except Exception as e:
self._display.warning(f"Failed to send ntfy notification: {e}")
def v2_playbook_on_start(self, playbook):
self.playbook = playbook
self.start_time = datetime.now()
def v2_runner_on_ok(self, result):
if result._result.get('changed', False):
host = result._host.get_name()
task_name = result._task.get_name()
self.changed_tasks.append(f"[{host}] {task_name}")
def v2_runner_on_failed(self, result, ignore_errors=False):
if ignore_errors:
return
host = result._host.get_name()
task_name = result._task.get_name()
self.failed_tasks.append(f"[{host}] {task_name}\nError: {result._result['msg']}")
def v2_runner_on_unreachable(self, result):
host = result._host.get_name()
self.unreachable_hosts.append(host)
def v2_playbook_on_stats(self, stats):
end_time = datetime.now()
duration = end_time - self.start_time
message = ""
total_changes = len(self.changed_tasks)
total_failures = len(self.failed_tasks) + len(self.unreachable_hosts)
if total_changes == 0 and total_failures == 0:
return
if self.changed_tasks:
message += "Changed Tasks:\n"
for task in self.changed_tasks:
message += f"{task}\n"
if self.failed_tasks:
message += "Failed Tasks:\n"
for task in self.failed_tasks:
message += f"{task}\n"
if self.unreachable_hosts:
message += "Unreachable Hosts:\n"
for host in self.unreachable_hosts:
message += f"{host}\n"
message += f"Duration: {str(duration).split('.')[0]}"
if total_failures > 0:
priority = "high"
tags = "gear,x,warning"
title = "Ansible Failed!"
else:
priority = "default"
tags = "gear,white_check_mark"
title = f"Ansible Complete - {total_changes} Changes"
self.send_notification(title, message.strip(), priority, tags)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment