Back to Details

zabbix

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Zabbix Automation Skill

Zabbix自动化技能

Overview

概述

This skill provides guidance for automating Zabbix monitoring operations via the API and official Python library 
zabbix_utils
.
本技能提供了通过API和官方Python库
zabbix_utils
实现Zabbix监控操作自动化的指导。

Quick Start

快速开始

Installation

安装

bash
pip install zabbix-utils --break-system-packages
bash
pip install zabbix-utils --break-system-packages

Authentication

身份验证

python
from zabbix_utils import ZabbixAPI
python
from zabbix_utils import ZabbixAPI

Option 1: Username/password

Option 1: Username/password

api = ZabbixAPI(url="https://zabbix.example.com") api.login(user="Admin", password="zabbix")
api = ZabbixAPI(url="https://zabbix.example.com") api.login(user="Admin", password="zabbix")

Option 2: API token (Zabbix 5.4+, preferred)

Option 2: API token (Zabbix 5.4+, preferred)

api = ZabbixAPI(url="https://zabbix.example.com") api.login(token="your_api_token")
api = ZabbixAPI(url="https://zabbix.example.com") api.login(token="your_api_token")

Verify connection

Verify connection

print(api.api_version())
undefined
print(api.api_version())
undefined

Environment Variables Pattern

环境变量模式

python
import os
from zabbix_utils import ZabbixAPI

api = ZabbixAPI(url=os.environ.get("ZABBIX_URL", "http://localhost/zabbix"))
api.login(token=os.environ["ZABBIX_TOKEN"])
python
import os
from zabbix_utils import ZabbixAPI

api = ZabbixAPI(url=os.environ.get("ZABBIX_URL", "http://localhost/zabbix"))
api.login(token=os.environ["ZABBIX_TOKEN"])

Core API Methods

核心API方法

All APIs follow pattern: 
api.<object>.<method>()
with methods: 
get
, 
create
, 
update
, 
delete
.
所有API遵循以下模式：
api.<对象>.<方法>()
，支持的方法包括：
get
、
create
、
update
、
delete
。

Host Operations

主机操作

python
undefined
python
undefined

Get hosts

Get hosts

hosts = api.host.get(output=["hostid", "host", "name"], selectInterfaces=["ip"])
hosts = api.host.get(output=["hostid", "host", "name"], selectInterfaces=["ip"])

Create host

Create host

api.host.create( host="server01", groups=[{"groupid": "2"}], # Linux servers interfaces=[{ "type": 1, # 1=agent, 2=SNMP, 3=IPMI, 4=JMX "main": 1, "useip": 1, "ip": "192.168.1.100", "dns": "", "port": "10050" }], templates=[{"templateid": "10001"}] )
api.host.create( host="server01", groups=[{"groupid": "2"}], # Linux servers interfaces=[{ "type": 1, # 1=agent, 2=SNMP, 3=IPMI, 4=JMX "main": 1, "useip": 1, "ip": "192.168.1.100", "dns": "", "port": "10050" }], templates=[{"templateid": "10001"}] )

Update host

Update host

api.host.update(hostid="10084", status=0) # 0=enabled, 1=disabled
api.host.update(hostid="10084", status=0) # 0=enabled, 1=disabled

Delete host

Delete host

api.host.delete("10084")
undefined
api.host.delete("10084")
undefined

Template Operations

模板操作

python
undefined
python
undefined

Get templates

Get templates

templates = api.template.get(output=["templateid", "host", "name"], selectHosts=["hostid", "name"])
templates = api.template.get(output=["templateid", "host", "name"], selectHosts=["hostid", "name"])

Link template to host

Link template to host

api.host.update(hostid="10084", templates=[{"templateid": "10001"}])
api.host.update(hostid="10084", templates=[{"templateid": "10001"}])

Import template from XML

Import template from XML

with open("template.xml") as f: api.configuration.import_( source=f.read(), format="xml", rules={ "templates": {"createMissing": True, "updateExisting": True}, "items": {"createMissing": True, "updateExisting": True}, "triggers": {"createMissing": True, "updateExisting": True} } )
undefined
with open("template.xml") as f: api.configuration.import_( source=f.read(), format="xml", rules={ "templates": {"createMissing": True, "updateExisting": True}, "items": {"createMissing": True, "updateExisting": True}, "triggers": {"createMissing": True, "updateExisting": True} } )
undefined

Item Operations

监控项操作

python
undefined
python
undefined

Get items

Get items

items = api.item.get(hostids="10084", output=["itemid", "name", "key_"], search={"key_": "system.cpu"})
items = api.item.get(hostids="10084", output=["itemid", "name", "key_"], search={"key_": "system.cpu"})

Create item

Create item

api.item.create( name="CPU Load", key_="system.cpu.load[percpu,avg1]", hostid="10084", type=0, # 0=Zabbix agent value_type=0, # 0=float, 3=integer, 4=text delay="30s", interfaceid="1" )
undefined
api.item.create( name="CPU Load", key_="system.cpu.load[percpu,avg1]", hostid="10084", type=0, # 0=Zabbix agent value_type=0, # 0=float, 3=integer, 4=text delay="30s", interfaceid="1" )
undefined

Trigger Operations

触发器操作

python
undefined
python
undefined

Get triggers

Get triggers

triggers = api.trigger.get(hostids="10084", output=["triggerid", "description", "priority"], selectFunctions="extend")
triggers = api.trigger.get(hostids="10084", output=["triggerid", "description", "priority"], selectFunctions="extend")

Create trigger

Create trigger

api.trigger.create( description="High CPU on {HOST.NAME}", expression="last(/server01/system.cpu.load[percpu,avg1])>5", priority=3 # 0=not classified, 1=info, 2=warning, 3=average, 4=high, 5=disaster )
undefined
api.trigger.create( description="High CPU on {HOST.NAME}", expression="last(/server01/system.cpu.load[percpu,avg1])>5", priority=3 # 0=not classified, 1=info, 2=warning, 3=average, 4=high, 5=disaster )
undefined

Host Group Operations

主机组操作

python
undefined
python
undefined

Get groups

Get groups

groups = api.hostgroup.get(output=["groupid", "name"])
groups = api.hostgroup.get(output=["groupid", "name"])

Create group

Create group

api.hostgroup.create(name="Production/Web Servers")
api.hostgroup.create(name="Production/Web Servers")

Add hosts to group

Add hosts to group

api.hostgroup.massadd(groups=[{"groupid": "5"}], hosts=[{"hostid": "10084"}])
undefined
api.hostgroup.massadd(groups=[{"groupid": "5"}], hosts=[{"hostid": "10084"}])
undefined

Maintenance Windows

维护窗口

python
import time
python
import time

Create maintenance

Create maintenance

api.maintenance.create( name="Server Maintenance", active_since=int(time.time()), active_till=int(time.time()) + 3600, # 1 hour hostids=["10084"], timeperiods=[{ "timeperiod_type": 0, # One-time "period": 3600 }] )
undefined
api.maintenance.create( name="Server Maintenance", active_since=int(time.time()), active_till=int(time.time()) + 3600, # 1 hour hostids=["10084"], timeperiods=[{ "timeperiod_type": 0, # One-time "period": 3600 }] )
undefined

Events and Problems

事件与问题

python
undefined
python
undefined

Get current problems

Get current problems

problems = api.problem.get(output=["eventid", "name", "severity"], recent=True)
problems = api.problem.get(output=["eventid", "name", "severity"], recent=True)

Get events

Get events

events = api.event.get(hostids="10084", time_from=int(time.time()) - 86400, output="extend")
undefined
events = api.event.get(hostids="10084", time_from=int(time.time()) - 86400, output="extend")
undefined

History Data

历史数据

python
undefined
python
undefined

Get history (value_type must match item's value_type)

Get history (value_type must match item's value_type)

0=float, 1=character, 2=log, 3=integer, 4=text

0=float, 1=character, 2=log, 3=integer, 4=text

history = api.history.get( itemids="28269", history=0, # float time_from=int(time.time()) - 3600, output="extend", sortfield="clock", sortorder="DESC" )
undefined
history = api.history.get( itemids="28269", history=0, # float time_from=int(time.time()) - 3600, output="extend", sortfield="clock", sortorder="DESC" )
undefined

Zabbix Sender (Trapper Items)

Zabbix Sender（Trapper监控项）

python
from zabbix_utils import Sender

sender = Sender(server="zabbix.example.com", port=10051)
python
from zabbix_utils import Sender

sender = Sender(server="zabbix.example.com", port=10051)

Send single value

Send single value

response = sender.send_value("hostname", "trap.key", "value123") print(response) # {"processed": 1, "failed": 0, "total": 1}
response = sender.send_value("hostname", "trap.key", "value123") print(response) # {"processed": 1, "failed": 0, "total": 1}

Send multiple values

Send multiple values

from zabbix_utils import ItemValue values = [ ItemValue("host1", "key1", "value1"), ItemValue("host2", "key2", 42), ] response = sender.send(values)
undefined
from zabbix_utils import ItemValue values = [ ItemValue("host1", "key1", "value1"), ItemValue("host2", "key2", 42), ] response = sender.send(values)
undefined

Zabbix Getter (Agent Query)

Zabbix Getter（Agent查询）

python
from zabbix_utils import Getter

agent = Getter(host="192.168.1.100", port=10050)
response = agent.get("system.uname")
print(response.value)
python
from zabbix_utils import Getter

agent = Getter(host="192.168.1.100", port=10050)
response = agent.get("system.uname")
print(response.value)

Common Patterns

常见模式

Bulk Host Creation from CSV

从CSV批量创建主机

python
import csv
from zabbix_utils import ZabbixAPI

api = ZabbixAPI(url="https://zabbix.example.com")
api.login(token="your_token")

with open("hosts.csv") as f:
    for row in csv.DictReader(f):
        try:
            api.host.create(
                host=row["hostname"],
                groups=[{"groupid": row["groupid"]}],
                interfaces=[{
                    "type": 1, "main": 1, "useip": 1,
                    "ip": row["ip"], "dns": "", "port": "10050"
                }]
            )
            print(f"Created: {row['hostname']}")
        except Exception as e:
            print(f"Failed {row['hostname']}: {e}")
python
import csv
from zabbix_utils import ZabbixAPI

api = ZabbixAPI(url="https://zabbix.example.com")
api.login(token="your_token")

with open("hosts.csv") as f:
    for row in csv.DictReader(f):
        try:
            api.host.create(
                host=row["hostname"],
                groups=[{"groupid": row["groupid"]}],
                interfaces=[{
                    "type": 1, "main": 1, "useip": 1,
                    "ip": row["ip"], "dns": "", "port": "10050"
                }]
            )
            print(f"Created: {row['hostname']}")
        except Exception as e:
            print(f"Failed {row['hostname']}: {e}")

Find Hosts Without Template

查找未关联模板的主机

python
undefined
python
undefined

Get all hosts

Get all hosts

all_hosts = api.host.get(output=["hostid", "host"], selectParentTemplates=["templateid"])
all_hosts = api.host.get(output=["hostid", "host"], selectParentTemplates=["templateid"])

Filter hosts without specific template

Filter hosts without specific template

template_id = "10001" hosts_without = [h for h in all_hosts if not any(t["templateid"] == template_id for t in h.get("parentTemplates", []))]
undefined
template_id = "10001" hosts_without = [h for h in all_hosts if not any(t["templateid"] == template_id for t in h.get("parentTemplates", []))]
undefined

Disable Triggers by Pattern

按模式禁用触发器

python
triggers = api.trigger.get(
    search={"description": "test"},
    output=["triggerid"]
)
for t in triggers:
    api.trigger.update(triggerid=t["triggerid"], status=1)  # 1=disabled
python
triggers = api.trigger.get(
    search={"description": "test"},
    output=["triggerid"]
)
for t in triggers:
    api.trigger.update(triggerid=t["triggerid"], status=1)  # 1=disabled

Item Types Reference

监控项类型参考

TypeValueDescription
Zabbix agent0Active checks
Zabbix trapper2Passive, data pushed via sender
Simple check3ICMP, TCP, etc.
Zabbix internal5Server internal metrics
Zabbix agent (active)7Agent-initiated
HTTP agent19HTTP/REST API monitoring
Dependent item18Derived from master item
Script21Custom scripts
类型值描述
Zabbix agent0主动检查
Zabbix trapper2被动模式，通过sender推送数据
Simple check3ICMP、TCP等
Zabbix internal5服务器内部指标
Zabbix agent (active)7Agent主动发起
HTTP agent19HTTP/REST API监控
Dependent item18从主监控项派生
Script21自定义脚本

Value Types Reference

值类型参考

TypeValueDescription
Float0Numeric (float)
Character1Character string
Log2Log file
Unsigned3Numeric (integer)
Text4Text
类型值描述
Float0数值（浮点型）
Character1字符串
Log2日志文件
Unsigned3数值（整数）
Text4文本

Trigger Severity Reference

触发器严重级别参考

SeverityValueColor
Not classified0Gray
Information1Light blue
Warning2Yellow
Average3Orange
High4Light red
Disaster5Red
严重级别值颜色
Not classified0灰色
Information1浅蓝色
Warning2黄色
Average3橙色
High4浅红色
Disaster5红色

Error Handling

错误处理

python
from zabbix_utils import ZabbixAPI
from zabbix_utils.exceptions import APIRequestError

try:
    api.host.create(host="duplicate_host", groups=[{"groupid": "2"}])
except APIRequestError as e:
    print(f"API Error: {e.message}")
    print(f"Code: {e.code}")
python
from zabbix_utils import ZabbixAPI
from zabbix_utils.exceptions import APIRequestError

try:
    api.host.create(host="duplicate_host", groups=[{"groupid": "2"}])
except APIRequestError as e:
    print(f"API Error: {e.message}")
    print(f"Code: {e.code}")

Debugging

调试

python
import logging
logging.basicConfig(level=logging.DEBUG)
python
import logging
logging.basicConfig(level=logging.DEBUG)

Now all API calls will be logged

Now all API calls will be logged

undefined
undefined

Scripts Reference

脚本参考

See 
scripts/
directory for ready-to-use automation:
  • zabbix-bulk-hosts.py
    - Bulk host management from CSV
  • zabbix-maintenance.py
    - Create/manage maintenance windows
  • zabbix-export.py
    - Export hosts/templates to JSON/XML
查看
scripts/
目录获取可直接使用的自动化脚本：
  • zabbix-bulk-hosts.py
    - 从CSV批量管理主机
  • zabbix-maintenance.py
    - 创建/管理维护窗口
  • zabbix-export.py
    - 将主机/模板导出为JSON/XML

Best Practices

最佳实践

  1. Use API tokens over username/password when possible
  2. Limit output fields - Always specify 
    output=["field1", "field2"]
    instead of 
    output="extend"
  3. Use search/filter - Never fetch all objects and filter in Python
  4. Handle pagination - Large result sets may need 
    limit
    and 
    offset
  5. Batch operations - Use 
    massadd
    , 
    massupdate
    for bulk changes
  6. Error handling - Always wrap API calls in try/except
  7. Idempotency - Check if object exists before creating
  1. 优先使用API令牌：尽可能使用API令牌而非用户名/密码
  2. 限制输出字段：始终指定
    output=["field1", "field2"]
    而非
    output="extend"
  3. 使用搜索/过滤：切勿获取所有对象后在Python中过滤
  4. 处理分页：大型结果集可能需要使用
    limit
    和
    offset
  5. 批量操作：使用
    massadd
    、
    massupdate
    执行批量变更
  6. 错误处理：始终将API调用包裹在try/except块中
  7. 幂等性：创建对象前先检查其是否已存在