"""
title: Zabbix Tools
author: Jonathan Torres
author_url: https://github.com/jonstevet/LLMTOOL_ZABBIXAPI
funding_url: https://github.com/jonstevet/LLMTOOL_ZABBIXAPI
version: 2.0.0
programming language: python
"""
import requests
import json
from pydantic import BaseModel, Field
from typing import List, Dict, Callable, Awaitable, Any
def request(
url: str,
headers: str,
auth: str,
method: str,
param1: str = None,
param2: str = None,
):
"""
Send a request to api server, whit method and params given.
Args:
url : str
Url to send request.
headers : str
Headers to send in request.
auth : str
Authentication token to append in the request method.
method : str
Method is the body request to send, "get_id", "get_status".
param1 : str
First parameter to fill in method.
param2 : str
Second parameter to fill in method.
Returns:
List[Dict[str, Any]]: A list of dicts strings whit hosts that match the given method, or an empty list if no hosts are found.
"""
templatesMethods = {
"get_id": {
"jsonrpc": "2.0",
"method": "host.get",
"params": {
"output": ["hostid", "host"],
"search": {"host": param1},
"selectInterfaces": ["ip"],
},
"id": 2,
"auth": auth,
},
"get_status": {
"jsonrpc": "2.0",
"method": "item.get",
"params": {
"output": ["key_", "lastvalue"],
"hostids": param1,
"filter": {"key_": "icmpping"},
},
"id": 1,
"auth": auth,
},
}
templateMethodsType = {"get_id": "POST", "get_status": "POST"}
data = requestApi(
url, headers, templatesMethods[method], templateMethodsType[method]
)
if data:
return data
else:
print(f"response is empty")
return []
def requestApi(
url: str, headers: str, method: dict[str, Any], methodType: str = "POST"
):
"""
Retriever from api server.
Args:
url : str
Url to send request.
headers : str
Headers to send in request.
method : str
Methods are the body request to send.
methodType : str
Type of Method example: "POST", "GET"... Default: "POST".
Returns:
List[Dict[str, Any]]: A list of host that match the given host name, or an empty list if no hosts are found.
"""
try:
response = requests.request(
methodType, url, headers=headers, data=json.dumps(method)
)
methodName = method["method"]
print(f"request send {methodName} to zabbix")
except requests.exceptions.RequestException as e:
return(f"Request to API failed: {e}")
if response.text:
try:
data = json.loads(response.text)
dataResult = data["result"]
return dataResult
except json.JSONDecodeError as e:
return(f"Failed to parse JSON response: {e}")
else:
return(f"No response from API: {e}")
async def emit(__event_emitter__: Callable[[dict], Awaitable[None]], description: str, done: bool=False):
await __event_emitter__(
{
"type": "status",
"data": {
"status": "in_progress" if not done else "complete",
"description": description,
"done": done,
},
}
)
async def emit_msg(__event_emitter__: Callable[[dict], Awaitable[None]], content: str):
await __event_emitter__(
{
"type": "message",
"data": {
"content": content,
},
}
)
class Tools:
class Valves(BaseModel):
zabbix_url: str = Field(
default="",
description="URL of the Zabbix API 'https://example.com/zabbix/api_jsonrpc.php'",
)
zabbix_auth_token: str = Field(
default="",
description="Authentication token for the Zabbix API",
)
pass
class UserValves(BaseModel):
pass
def __init__(self):
self.valves = self.Valves()
self.userValves = self.UserValves()
pass
async def get_host_status(
self,
__event_emitter__: Callable[[dict], Awaitable[None]],
hostName: str,
) -> str:
"""
Retrieves the operational status of given host switch, access point (ap) or server from the Zabbix network monitor API.
Args:
hostsName : str
Name of host switch, access point (ap) or server to search in zabbix network monitor api.
Returns:
str: A string containing information about the operative status of the host switch, access point (ap) or server.
"""
if not hostName:
await emit(__event_emitter__, f"Error, no host name", True)
return("The host NAME must not be empty, query to user the hostname acording use this tool")
else:
print(f"Host name, {hostName}")
self.url: str = self.valves.zabbix_url
self.auth_token: str = self.valves.zabbix_auth_token
if not self.url:
await emit(__event_emitter__, f"Error, no zabbix url", True)
return("The url config must not be empty, query to user the set it in tools config acording use this tool")
elif not self.auth_token:
await emit(__event_emitter__, f"Error, no zabbix authentication token", True)
return("The authentication token config must not be empty, query to user the set it in tools config acording use this tool")
else:
print(f"Url and Authentication token set, {self.url}")
self.headers = {
"Content-Type": "application/json-rpc",
"Authorization": f"Bearer {self.auth_token}",
}
await emit(__event_emitter__, f"Buscando el id para {hostName} en el servidor zabbix")
hosts = request(
url=self.url,
headers=self.headers,
auth=self.auth_token,
method="get_id",
param1=hostName,
)
if hosts:
await emit(__event_emitter__, "Equipos encontrados")
await emit_msg(__event_emitter__, "|Nombre|IP|Estado|\n")
await emit_msg(__event_emitter__, "|:---:|:---:|:---:|\n")
status_list = []
for host in hosts:
hostNameOut = host.get("host", "No data found")
await emit(__event_emitter__, f"Obteniendo el estado de: {hostNameOut}")
data = request(
url=self.url,
headers=self.headers,
auth=self.auth_token,
method="get_status",
param1=host["hostid"],
)
try:
if data[0]:
item = data[0]
status = "up" if item["lastvalue"] == "1" else "down"
# Ensure the 'interfaces' list exists and has at least one item before accessing it
if host.get("interfaces") and len(host["interfaces"]) > 0:
ip = host["interfaces"][0].get("ip", "No data found")
else:
ip = "No data found"
status_list.append(f"The host {hostNameOut} with ip {ip} is {status}")
await emit(__event_emitter__, f"Estado: {status}")
await emit_msg(__event_emitter__, f"|{hostNameOut}|{ip}|{status}|\n")
else:
await emit(__event_emitter__, f"Sin datos")
status_list.append(
f"The host {hostNameOut} no data"
)
except Exception as e:
await emit(__event_emitter__, f"Error Obteniendo el status de {hostNameOut}, {str(e)}", True)
status_list.append(f"The host {hostNameOut} with ip {ip} has an error: {str(e)}")
status_list.append(f"Resume the hosts status")
response = "\n".join(status_list)
await emit(__event_emitter__, "Tarea completa", done=True)
await emit_msg(__event_emitter__, "\n")
return response