Configure Zabbix 7 API automation with Python scripts for monitoring and alerting

Install Python dependencies

Install the required Python packages for Zabbix API interaction and HTTP requests.

sudo apt update
sudo apt install -y python3 python3-pip python3-venv
python3 -m venv zabbix-automation
source zabbix-automation/bin/activate
pip install pyzabbix requests

sudo dnf update -y
sudo dnf install -y python3 python3-pip
python3 -m venv zabbix-automation
source zabbix-automation/bin/activate
pip install pyzabbix requests

Create API connection script

Set up the base connection module that handles authentication and API requests to your Zabbix server.

#!/usr/bin/env python3
import json
from pyzabbix import ZabbixAPI
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class ZabbixConnection:
    def __init__(self, url, username, password):
        self.url = url
        self.username = username
        self.password = password
        self.api = None
    
    def connect(self):
        """Establish connection to Zabbix API"""
        try:
            self.api = ZabbixAPI(self.url)
            self.api.login(self.username, self.password)
            logger.info(f"Connected to Zabbix API: {self.url}")
            return True
        except Exception as e:
            logger.error(f"Failed to connect to Zabbix API: {e}")
            return False
    
    def disconnect(self):
        """Close API connection"""
        if self.api:
            self.api.user.logout()
            logger.info("Disconnected from Zabbix API")
    
    def get_api(self):
        """Return the API object"""
        return self.api

Create configuration file

Store your Zabbix server connection details in a separate configuration file for security and flexibility.

{
    "zabbix": {
        "url": "http://your-zabbix-server/zabbix",
        "username": "api-user",
        "password": "your-secure-password"
    },
    "monitoring": {
        "default_group": "Linux servers",
        "default_templates": [
            "Linux by Zabbix agent",
            "Linux filesystems by Zabbix agent"
        ]
    },
    "alerts": {
        "email_media_type": "Email",
        "severity_threshold": 3
    }
}

Set secure file permissions

Protect the configuration file containing credentials from unauthorized access.

chmod 600 /home/user/zabbix-scripts/config.json
chmod +x /home/user/zabbix-scripts/zabbix_api.py

Create host discovery automation script

Implement automated host discovery and registration with template assignment and group membership.

#!/usr/bin/env python3
import json
import sys
import argparse
from zabbix_api import ZabbixConnection
import logging

logger = logging.getLogger(__name__)

class HostManager:
    def __init__(self, config_file):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        
        self.zabbix = ZabbixConnection(
            self.config['zabbix']['url'],
            self.config['zabbix']['username'],
            self.config['zabbix']['password']
        )
    
    def add_host(self, hostname, ip_address, groups=None, templates=None, port=10050):
        """Add a new host to Zabbix monitoring"""
        if not self.zabbix.connect():
            return False
        
        api = self.zabbix.get_api()
        
        try:
            # Get group IDs
            if not groups:
                groups = [self.config['monitoring']['default_group']]
            
            group_ids = []
            for group_name in groups:
                group = api.hostgroup.get(filter={'name': group_name})
                if not group:
                    # Create group if it doesn't exist
                    group_result = api.hostgroup.create(name=group_name)
                    group_ids.append({'groupid': group_result['groupids'][0]})
                    logger.info(f"Created new host group: {group_name}")
                else:
                    group_ids.append({'groupid': group[0]['groupid']})
            
            # Get template IDs
            if not templates:
                templates = self.config['monitoring']['default_templates']
            
            template_ids = []
            for template_name in templates:
                template = api.template.get(filter={'host': template_name})
                if template:
                    template_ids.append({'templateid': template[0]['templateid']})
                else:
                    logger.warning(f"Template not found: {template_name}")
            
            # Check if host already exists
            existing_host = api.host.get(filter={'host': hostname})
            if existing_host:
                logger.warning(f"Host {hostname} already exists")
                return False
            
            # Create host
            host_data = {
                'host': hostname,
                'name': hostname,
                'groups': group_ids,
                'templates': template_ids,
                'interfaces': [{
                    'type': 1,  # Agent interface
                    'main': 1,
                    'useip': 1,
                    'ip': ip_address,
                    'dns': '',
                    'port': str(port)
                }]
            }
            
            result = api.host.create(**host_data)
            logger.info(f"Successfully added host: {hostname} ({ip_address})")
            return result['hostids'][0]
            
        except Exception as e:
            logger.error(f"Failed to add host {hostname}: {e}")
            return False
        finally:
            self.zabbix.disconnect()
    
    def bulk_add_hosts(self, hosts_file):
        """Add multiple hosts from a JSON file"""
        with open(hosts_file, 'r') as f:
            hosts = json.load(f)
        
        results = []
        for host in hosts:
            result = self.add_host(
                host['hostname'],
                host['ip_address'],
                host.get('groups'),
                host.get('templates'),
                host.get('port', 10050)
            )
            results.append({'hostname': host['hostname'], 'success': bool(result)})
        
        return results

def main():
    parser = argparse.ArgumentParser(description='Zabbix Host Discovery Automation')
    parser.add_argument('--config', default='config.json', help='Configuration file path')
    parser.add_argument('--hostname', help='Single hostname to add')
    parser.add_argument('--ip', help='IP address for single host')
    parser.add_argument('--bulk-file', help='JSON file with multiple hosts')
    
    args = parser.parse_args()
    
    host_manager = HostManager(args.config)
    
    if args.bulk_file:
        results = host_manager.bulk_add_hosts(args.bulk_file)
        for result in results:
            print(f"Host {result['hostname']}: {'Success' if result['success'] else 'Failed'}")
    elif args.hostname and args.ip:
        result = host_manager.add_host(args.hostname, args.ip)
        print(f"Host addition: {'Success' if result else 'Failed'}")
    else:
        parser.print_help()

if __name__ == '__main__':
    main()

Create custom monitoring script

Implement custom monitoring checks that extend beyond standard Zabbix templates for application-specific metrics.

#!/usr/bin/env python3
import json
import sys
from zabbix_api import ZabbixConnection
import logging

logger = logging.getLogger(__name__)

class CustomMonitoring:
    def __init__(self, config_file):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        
        self.zabbix = ZabbixConnection(
            self.config['zabbix']['url'],
            self.config['zabbix']['username'],
            self.config['zabbix']['password']
        )
    
    def create_custom_item(self, hostname, item_config):
        """Create a custom monitoring item for a host"""
        if not self.zabbix.connect():
            return False
        
        api = self.zabbix.get_api()
        
        try:
            # Get host ID
            host = api.host.get(filter={'host': hostname})
            if not host:
                logger.error(f"Host not found: {hostname}")
                return False
            
            host_id = host[0]['hostid']
            
            # Create item
            item_data = {
                'hostid': host_id,
                'name': item_config['name'],
                'key_': item_config['key'],
                'type': item_config.get('type', 0),  # Zabbix agent
                'value_type': item_config.get('value_type', 3),  # Numeric unsigned
                'delay': item_config.get('delay', '60s'),
                'history': item_config.get('history', '90d'),
                'trends': item_config.get('trends', '365d'),
                'description': item_config.get('description', '')
            }
            
            result = api.item.create(**item_data)
            logger.info(f"Created custom item: {item_config['name']} for host {hostname}")
            return result['itemids'][0]
            
        except Exception as e:
            logger.error(f"Failed to create item: {e}")
            return False
        finally:
            self.zabbix.disconnect()
    
    def create_custom_trigger(self, hostname, trigger_config):
        """Create a custom trigger for monitoring alerts"""
        if not self.zabbix.connect():
            return False
        
        api = self.zabbix.get_api()
        
        try:
            # Create trigger
            trigger_data = {
                'description': trigger_config['description'],
                'expression': trigger_config['expression'],
                'priority': trigger_config.get('priority', 3),  # Average severity
                'status': trigger_config.get('status', 0),  # Enabled
                'type': trigger_config.get('type', 0),  # Single problem
                'recovery_mode': trigger_config.get('recovery_mode', 0),  # Expression
                'recovery_expression': trigger_config.get('recovery_expression', ''),
                'comments': trigger_config.get('comments', '')
            }
            
            result = api.trigger.create(**trigger_data)
            logger.info(f"Created custom trigger: {trigger_config['description']}")
            return result['triggerids'][0]
            
        except Exception as e:
            logger.error(f"Failed to create trigger: {e}")
            return False
        finally:
            self.zabbix.disconnect()
    
    def setup_application_monitoring(self, hostname, app_name, metrics):
        """Set up comprehensive application monitoring"""
        results = []
        
        for metric in metrics:
            # Create monitoring item
            item_config = {
                'name': f"{app_name} - {metric['name']}",
                'key': f"custom.{app_name}.{metric['key']}",
                'type': metric.get('type', 0),
                'value_type': metric.get('value_type', 3),
                'delay': metric.get('delay', '60s'),
                'description': metric.get('description', '')
            }
            
            item_result = self.create_custom_item(hostname, item_config)
            
            # Create associated trigger if specified
            if 'trigger' in metric:
                trigger_config = {
                    'description': f"{app_name} - {metric['trigger']['description']}",
                    'expression': metric['trigger']['expression'].format(
                        hostname=hostname,
                        key=item_config['key']
                    ),
                    'priority': metric['trigger'].get('priority', 3)
                }
                
                trigger_result = self.create_custom_trigger(hostname, trigger_config)
                results.append({
                    'metric': metric['name'],
                    'item': bool(item_result),
                    'trigger': bool(trigger_result)
                })
            else:
                results.append({
                    'metric': metric['name'],
                    'item': bool(item_result),
                    'trigger': None
                })
        
        return results

def main():
    # Example usage
    custom_mon = CustomMonitoring('config.json')
    
    # Example application monitoring setup
    nginx_metrics = [
        {
            'name': 'Active Connections',
            'key': 'nginx.connections.active',
            'description': 'Number of active NGINX connections',
            'trigger': {
                'description': 'High number of active connections',
                'expression': 'last(/{hostname}/custom.nginx.connections.active)>1000',
                'priority': 3
            }
        },
        {
            'name': 'Requests per Second',
            'key': 'nginx.requests.rate',
            'description': 'NGINX requests per second',
            'trigger': {
                'description': 'High request rate detected',
                'expression': 'last(/{hostname}/custom.nginx.requests.rate)>500',
                'priority': 2
            }
        }
    ]
    
    results = custom_mon.setup_application_monitoring('web-server-01', 'nginx', nginx_metrics)
    
    for result in results:
        print(f"Metric: {result['metric']} - Item: {result['item']}, Trigger: {result['trigger']}")

if __name__ == '__main__':
    main()

Create alerting automation script

Implement automated alert configuration with user notifications and escalation policies.

#!/usr/bin/env python3
import json
from zabbix_api import ZabbixConnection
import logging

logger = logging.getLogger(__name__)

class AlertManager:
    def __init__(self, config_file):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        
        self.zabbix = ZabbixConnection(
            self.config['zabbix']['url'],
            self.config['zabbix']['username'],
            self.config['zabbix']['password']
        )
    
    def create_user_group(self, group_name, users=None):
        """Create a user group for alert management"""
        if not self.zabbix.connect():
            return False
        
        api = self.zabbix.get_api()
        
        try:
            # Check if group exists
            existing_group = api.usergroup.get(filter={'name': group_name})
            if existing_group:
                logger.info(f"User group {group_name} already exists")
                return existing_group[0]['usrgrpid']
            
            # Create user group
            group_data = {
                'name': group_name,
                'gui_access': 2,  # Internal authentication
                'users_status': 0,  # Enabled
                'debug_mode': 0  # Disabled
            }
            
            if users:
                user_ids = []
                for username in users:
                    user = api.user.get(filter={'alias': username})
                    if user:
                        user_ids.append({'userid': user[0]['userid']})
                group_data['userids'] = user_ids
            
            result = api.usergroup.create(**group_data)
            logger.info(f"Created user group: {group_name}")
            return result['usrgrpids'][0]
            
        except Exception as e:
            logger.error(f"Failed to create user group: {e}")
            return False
        finally:
            self.zabbix.disconnect()
    
    def create_action(self, action_name, conditions, operations):
        """Create an automated action for alerts"""
        if not self.zabbix.connect():
            return False
        
        api = self.zabbix.get_api()
        
        try:
            # Check if action exists
            e