"""
MCP 도구 정의 및 핸들러
"""
import logging
from typing import Any, Dict, List
from mcp.types import TextContent, Tool
from zabbix_api import ZabbixAPI
from utils import ok, error, text, clamp_int
logger = logging.getLogger(__name__)
# Zabbix API 인스턴스 (전역 변수)
zabbix_api: ZabbixAPI | None = None
def get_tools() -> List[Tool]:
"""사용 가능한 도구 목록을 반환합니다."""
return [
Tool(
name="list_problems",
description="현재 발생 중인 Zabbix 문제(Problems) 목록을 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"recent": {
"type": "integer",
"description": "최근 N분 이내의 문제만 조회 (기본값: 60분)",
"default": 60,
"minimum": 1
},
"limit": {
"type": "integer",
"description": "최대 반환할 문제 수 (기본값: 100)",
"default": 100,
"minimum": 1,
"maximum": 1000
}
},
"additionalProperties": False
}
),
Tool(
name="list_triggers",
description="현재 활성화된 Zabbix 트리거(Triggers) 목록을 조회합니다.",
inputSchema={
"type": "object",
"properties": {
"min_severity": {
"type": "integer",
"description": "최소 심각도 (0=Not classified, 1=Information, 2=Warning, 3=Average, 4=High, 5=Disaster)",
"default": 0,
"minimum": 0,
"maximum": 5
},
"limit": {
"type": "integer",
"description": "최대 반환할 트리거 수 (기본값: 100)",
"default": 100,
"minimum": 1,
"maximum": 1000
}
},
"additionalProperties": False
}
),
Tool(
name="zabbix_health",
description="Zabbix API 연결 상태와 버전을 확인합니다.",
inputSchema={"type": "object", "properties": {}, "additionalProperties": False}
)
]
async def handle_tool_call(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""도구 호출을 처리합니다."""
global zabbix_api
# Zabbix API 초기화 (최초 호출 시)
if zabbix_api is None:
try:
zabbix_api = ZabbixAPI()
if not zabbix_api.authenticate():
return error("Zabbix 인증에 실패했습니다. 환경 변수를 확인해주세요.", code="AUTH_FAILED")
except Exception as e:
return error("Zabbix API 초기화 실패", details={"exception": str(e)}, code="INIT_FAILED")
try:
if name == "list_problems":
return await _handle_list_problems(arguments)
elif name == "list_triggers":
return await _handle_list_triggers(arguments)
elif name == "zabbix_health":
return await _handle_zabbix_health()
else:
return error(f"알 수 없는 도구: {name}", code="UNKNOWN_TOOL")
except Exception as e:
logger.error(f"도구 실행 오류 ({name}): {e}")
return error("도구 실행 중 오류가 발생했습니다", details={"exception": str(e), "tool": name}, code="TOOL_ERROR")
async def _handle_list_problems(arguments: Dict[str, Any]) -> List[TextContent]:
"""list_problems 도구 처리"""
recent = clamp_int(arguments.get("recent", 60), 60, min_value=1)
limit = clamp_int(arguments.get("limit", 100), 100, min_value=1, max_value=1000)
problems = zabbix_api.get_problems(recent_minutes=recent, limit=limit)
return ok(
{
"problems": problems,
"total_count": len(problems),
"query_params": {"recent_minutes": recent, "limit": limit},
},
message=f"최근 {recent}분 이내 발생한 문제 {len(problems)}개를 조회했습니다.",
)
async def _handle_list_triggers(arguments: Dict[str, Any]) -> List[TextContent]:
"""list_triggers 도구 처리"""
min_severity = clamp_int(arguments.get("min_severity", 0), 0, min_value=0, max_value=5)
limit = clamp_int(arguments.get("limit", 100), 100, min_value=1, max_value=1000)
triggers = zabbix_api.get_triggers(min_severity=min_severity, limit=limit)
return ok(
{
"triggers": triggers,
"total_count": len(triggers),
"query_params": {"min_severity": min_severity, "limit": limit},
},
message=f"심각도 {min_severity} 이상의 활성 트리거 {len(triggers)}개를 조회했습니다.",
)
async def _handle_zabbix_health() -> List[TextContent]:
"""zabbix_health 도구 처리"""
try:
version = zabbix_api.api_info_version()
return ok({"version": version}, message="Zabbix API 연결 정상")
except Exception as e:
logger.error(f"Zabbix health check 실패: {e}")
return error("Zabbix API 헬스체크 실패", details={"exception": str(e)}, code="HEALTHCHECK_FAILED")