Zabbix API 是一个用 Python 实现的 Zabbix API 接口库。Zabbix 是一个企业级的开源分布式监控解决方案，支持从网络设备到应用程序的广泛监控。Zabbix API 允许用户通过编程方式与 Zabbix 服务器进行交互，实现自动化配置、数据提取和报告生成等功能。
Zabbix API 是基于 HTTP 的，通常以 JSON 格式请求和响应。所有的 API 请求都通过 HTTP POST 方法进行，URL 通常是 http://<zabbix_server>/zabbix/api_jsonrpc.php。通过调用不同的 API 方法，可以实现对 Zabbix 系统的各种操作。
1.认证与会话
在使用 Zabbix API 之前，首先需要通过 user.login 方法进行身份验证，获取一个会话（session）ID。认证后，API 会返回一个认证 token，后续所有的请求都需要携带这个 token。
请求示例：
{
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"user": "Admin",
"password": "zabbix"
}, #Zabbix的用户密码
"id": 1
}
响应示例：
{
"jsonrpc": "2.0",
"result": "f8c7fd915ca219070c8a70d51a7e9b83",
"id": 1
}
2.常用API方法
Zabbix API 提供了多种方法，可以用于获取、创建、更新和删除 Zabbix 系统中的各种对象，如主机、触发器、监控项等。
常见的方法包括：
host.get：获取主机信息。
host.create：创建主机。
host.update：更新主机信息。
host.delete：删除主机。
item.get：获取监控项信息。
item.create：创建监控项。
item.update：更新监控项。
trigger.get：获取触发器信息。
trigger.create：创建触发器。
trigger.update：更新触发器。
event.get：获取事件。
user.get：获取用户信息。
3.请求格式
API 请求通常包括以下几个部分：
jsonrpc: 表示 JSON-RPC 协议的版本，通常为 "2.0"。
method: 要调用的 API 方法。
params: 请求的参数，格式根据方法不同而变化。
auth: 认证 token，通常是从 user.login 方法中获得。
id: 请求的唯一标识符，可以是任何整数。
注：请求除了必须是POST方法之外，HTTP Header Content-Type必须为：application/jsonrequest，application/json-rpc，application/json其中之一。
示例：
{
"jsonrpc": "2.0",
"result": [
{
"hostid": "10105",
"name": "Zabbix server",
"status": "0"
},
{
"hostid": "10106",
"name": "Zabbix agent",
"status": "0"
}
],
"id": 1
}
1.获取主机信息
# -*- coding: utf-8 -*-
import requests
import json
zabbix_url = 'http://192.168.88.128/z/api_jsonrpc.php'
username = 'Admin'
password = 'zabbix'
def get_auth_token():
headers = {'Content-Type': 'application/json'}
data = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"username": username,
"password": password
},
"id": 1
}
response = requests.post(zabbix_url, headers=headers, data=json.dumps(data))
return response.json().get('result')
def get_hosts(auth_token):
headers = {'Content-Type': 'application/json'}
data = {
"jsonrpc": "2.0",
"method": "host.get",
"params": {
"output": ["host"],
"selectInterfaces": ["ip"]
},
"auth": auth_token,
"id": 1
}
response = requests.post(zabbix_url, headers=headers, data=json.dumps(data))
return response.json().get('result')
def main():
auth_token = get_auth_token()
hosts = get_hosts(auth_token)
for host in hosts:
host_name = host['host']
ip_address = host['interfaces'][0]['ip']
print("{}, {}".format(host_name, ip_address))
if __name__ == "__main__":
main()
2.获取组信息
# -*- coding: utf-8 -*-
import requests
import json
import sys
if sys.stdout.encoding != 'utf-8':
import io
import sys
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
zabbix_url = 'http://192.168.88.128/z/api_jsonrpc.php'
username = 'Admin'
password = 'zabbix'
def get_auth_token():
headers = {'Content-Type': 'application/json'}
data = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"username": username,
"password": password
},
"id": 1
}
response = requests.post(zabbix_url, headers=headers, data=json.dumps(data))
return response.json().get('result')
def get_user_groups(auth_token):
headers = {'Content-Type': 'application/json'}
data = {
"jsonrpc": "2.0",
"method": "usergroup.get",
"params": {
"output": ["usrgrpid","name"]
},
"auth": auth_token,
"id": 1
}
response = requests.post(zabbix_url, headers=headers, data=json.dumps(data))
return response.json().get('result')
def main():
auth_token = get_auth_token()
user_groups = get_user_groups(auth_token)
for group in user_groups:
group_id = group['usrgrpid']
group_name = group['name']
print("{}, {}".format(group_id, group_name))
if __name__ == "__main__":
main()
1.根据xlsx文件创建用户
import pandas as pd
import requests
ZABBIX_URL = 'http://192.168.88.128/z/api_jsonrpc.php'
USERNAME = 'Admin'
PASSWORD = 'zabbix'
def zabbix_login():
auth_payload = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"user": USERNAME,
"password": PASSWORD
},
"id": 1
}
response = requests.post(ZABBIX_URL, json=auth_payload)
return response.json().get('result')
def create_user(auth_token, name, login, password, user_groups, role_id):
user_payload = {
"jsonrpc": "2.0",
"method": "user.create",
"params": {
"alias": login,
"name": name,
"passwd": password,
"usrgrps": user_groups,
"roleid": role_id
},
"auth": auth_token,
"id": 1
}
response = requests.post(ZABBIX_URL, json=user_payload)
return response.json()
def create_user_group(auth_token, group_name):
group_payload = {
"jsonrpc": "2.0",
"method": "usergroup.create",
"params": {
"name": group_name
},
"auth": auth_token,
"id": 1
}
response = requests.post(ZABBIX_URL, json=group_payload)
return response.json().get('result')
def get_user_group_id(group_name, auth_token):
group_payload = {
"jsonrpc": "2.0",
"method": "usergroup.get",
"params": {
"output": ["usrgrpid", "name"],
"filter": {
"name": group_name
}
},
"auth": auth_token,
"id": 1
}
response = requests.post(ZABBIX_URL, json=group_payload)
user_groups = response.json().get('result', [])
return user_groups[0]['usrgrpid'] if user_groups else None
def read_users_from_excel(file_path):
df = pd.read_excel(file_path)
users = []
for index, row in df.iterrows():
role_id = None
if row['普通用户权限'] == 1 and pd.isnull(row['管理员权限']):
role_id = 4
elif pd.isnull(row['普通用户权限']) or row['管理员权限'] == 1:
role_id = 3
users.append({
"department": row['部门'],
"user_id": row['工号'],
"name": row['姓名'],
"role_id": role_id
})
return users
def zabbix_logout(auth_token):
logout_payload = {
"jsonrpc": "2.0",
"method": "user.logout",
"params": [],
"auth": auth_token,
"id": 1
}
response = requests.post(ZABBIX_URL, json=logout_payload)
return response.json()
def main():
auth_token = zabbix_login()
if not auth_token:
print("认证失败")
return
users = read_users_from_excel('/tmp/test.xlsx')
for user in users:
group_id = get_user_group_id(user['department'], auth_token)
if not group_id: # 如果用户组不存在，创建它
result = create_user_group(auth_token, user['department'])
group_id = result.get('usrgrpid')
if group_id:
print("用户组 '{}' 创建成功，ID: {}".format(user['department'], group_id))
else:
print("创建用户组 '{}' 失败".format(user['department']))
continue
user_groups = [{"usrgrpid": group_id}]
role_id = user['role_id']
if role_id is not None:
result = create_user(auth_token, str(user['name']), str(user['user_id']), 'Zxcvbnm111', user_groups, role_id)
print("创建用户 {} 的结果: {}".format(user['name'], result))
else:
print("用户 {} 的权限信息不完整，跳过创建".format(user['name']))
zabbix_logout(auth_token)
if __name__ == "__main__":
main()
文件格式
结果：
