import os
import json
from typing import List, Dict, Any
from minzhi.cmdb_client import QueryBuilder
from minzhi import CmdbClient
from loguru import logger
class CMDBTool:
def __init__(self):
# Configuration loaded from Environment Variables
self.cmdb_server = os.getenv("CMDB_SERVER")
self.app_id = os.getenv("CMDB_APP_ID")
self.app_secret = os.getenv("CMDB_APP_SECRET")
self.view_id = os.getenv("CMDB_VIEW_ID")
def search_cmdb_asset(self, keyword: str) -> List[Dict[str, Any]]:
"""
Search for assets in the CMDB based on a keyword (name, IP, or system).
Args:
keyword (str): The search term (e.g., "Finance", "10.1.1.5").
Returns:
List[Dict[str, Any]]: A list of matching asset details.
"""
logger.info(f"[CMDB] Searching for asset with keyword: {keyword}")
# Strict Validation
missing_vars = []
if not self.cmdb_server: missing_vars.append("CMDB_SERVER")
if not self.app_id: missing_vars.append("CMDB_APP_ID")
if not self.app_secret: missing_vars.append("CMDB_APP_SECRET")
if missing_vars:
raise ValueError(f"Configuration Error: Missing required environment variables: {', '.join(missing_vars)}")
try:
client = CmdbClient(
view_id=self.view_id,
CMDB_SERVER=self.cmdb_server,
APPID=self.app_id,
APPSECRET=self.app_secret
)
query = (
QueryBuilder()
.like("businessSys-businessSysName", keyword)
.not_equal("VM-businessIp", "")
.build()
)
# Request specific fields as per user requirement
requested_fields = [
"VM-businessIp",
"VM-vHostName",
"businessSys-businessSysName",
"businessSys-businessSysAlias",
"_id",
"businessSys-abbreviation",
"businessSys-sysDatacenter_show_value",
"businessSys-desc",
"businessSys-level",
"businessSys-Type",
"businessSys-Deploymentway_show_value"
]
data = client.search(
queryKey=query,
fields=requested_fields,
count=10
)
# Map raw CMDB data to standardized tool output
results = []
for item in data:
# Construct logical tags from metadata
tags = []
if item.get("businessSys-level"):
tags.append(f"Level:{item.get('businessSys-level')}")
if item.get("businessSys-Type"):
tags.append(f"Type:{item.get('businessSys-Type')}")
if item.get("businessSys-Deploymentway_show_value"):
tags.append(f"Deploy:{item.get('businessSys-Deploymentway_show_value')}")
asset = {
"ip": item.get("VM-businessIp"),
"name": item.get("VM-vHostName"),
"system": item.get("businessSys-businessSysName"),
"app_id": item.get("businessSys-businessSysAlias"),
"id": item.get("_id"),
"abbreviation": item.get("businessSys-abbreviation", "UNKNOWN"),
"data_center": item.get("businessSys-sysDatacenter_show_value", "Default-DC"),
"description": item.get("businessSys-desc", ""),
"owner": item.get("businessSys-owner", "Ops"), # Keep default as owner key is missing in sample
"tags": tags
}
# Filter out empty IPs if any
if asset["ip"]:
results.append(asset)
return results
except Exception as e:
logger.error(f"[CMDB] API Call failed: {e}")
raise e
# Example usage for Dify Tool script
def main(keyword: str):
tool = CMDBTool()
return tool.search_cmdb_asset(keyword)
if __name__ == "__main__":
import os
# For local test only
os.environ["CMDB_SERVER"] = "http://10.1.4.100"
os.environ["CMDB_APP_ID"] = "c9a373fd4e31475faff965f75e760339"
os.environ["CMDB_APP_SECRET"] = "3d22ecb9bd844587ae4d493a9237f5b3"
os.environ["CMDB_VIEW_ID"] = "f0410fb453294ff998f49f583ffe38eb"
try:
logger.info(json.dumps(main("Automation Engine"), indent=2, ensure_ascii=False))
except Exception as e:
logger.error(f"Error: {e}")
