#!/usr/bin/env python3
"""
Zabbix MCP 서버
Zabbix API와 연결하여 현재 발생 중인 문제와 트리거 정보를 조회하는 MCP 서버
"""
import asyncio
import logging
import sys
from typing import Any, Dict, List
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from config import validate_environment
from tools import get_tools, handle_tool_call
# 로깅 설정: STDIO 프로토콜을 오염시키지 않도록 stderr로만 출력
logger = logging.getLogger(__name__)
if not logger.handlers:
handler = logging.StreamHandler(stream=sys.stderr)
formatter = logging.Formatter("%(levelname)s:%(name)s:%(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# MCP 서버 인스턴스 생성
server = Server("zabbix-mcp-server")
@server.list_tools()
async def list_tools() -> List[Tool]:
"""사용 가능한 도구 목록을 반환합니다."""
return get_tools()
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""도구 호출을 처리합니다."""
return await handle_tool_call(name, arguments)
async def main():
"""MCP 서버를 시작합니다."""
logger.info("Zabbix MCP 서버를 시작합니다...")
logger.info(f"python={sys.executable} cwd={os.getcwd()}")
# 환경 변수 검증
is_valid, missing_vars = validate_environment()
if not is_valid:
logger.error(f"환경 변수 설정 오류: {', '.join(missing_vars)}")
# 명확한 실패 신호를 위해 비정상 종료 코드 반환
raise SystemExit(2)
try:
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
except Exception as e:
logger.error(f"서버 실행 오류: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())