""" 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