MySQL Hub MCP Server

from pydantic import BaseModel, Field from typing import Dict, Any, List, Optional import logging import os import sys #from database import db_manager #from ai_provider import ai_manager #from config import config logger = logging.getLogger(__name__) # Pydantic 모델들 class SQLQueryRequest(BaseModel): query: str class NaturalLanguageRequest(BaseModel): question: str class TableSchemaRequest(BaseModel): table_name: str class AIProviderRequest(BaseModel): provider: str = Field(..., description="AI Provider 이름", pattern="^(groq|ollama|google)$") class Response(BaseModel): success: bool data: Any = None error: Optional[str] = None def model_dump(self, **kwargs): """UTF-8 인코딩 문제를 해결하기 위한 커스텀 model_dump""" try: return super().model_dump(**kwargs) except UnicodeDecodeError as e: # UTF-8 인코딩 오류 발생 시 데이터 정리 logger.warning(f"UTF-8 인코딩 오류 발생, 데이터 정리 중: {e}") # 데이터 정리 cleaned_data = self._clean_data(self.data) cleaned_error = self._clean_string(self.error) if self.error else None return { "success": self.success, "data": cleaned_data, "error": cleaned_error } def _clean_data(self, data): """데이터에서 UTF-8 문제가 있는 부분을 정리""" if isinstance(data, str): return self._clean_string(data) elif isinstance(data, dict): return {k: self._clean_data(v) for k, v in data.items()} elif isinstance(data, list): return [self._clean_data(item) for item in data] else: return data def _clean_string(self, text): """문자열에서 UTF-8 문제가 있는 부분을 정리""" if not isinstance(text, str): return str(text) try: # UTF-8로 인코딩/디코딩하여 문제 있는 문자 제거 return text.encode('utf-8', errors='ignore').decode('utf-8') except Exception: # 최후의 수단: ASCII로 변환 return text.encode('ascii', errors='ignore').decode('ascii') def clear_screen(): """화면을 지우는 함수""" if os.name == 'nt': # Windows 환경이지만 Git Bash 등에서는 'clear'를 사용해야 함 if 'MSYSTEM' in os.environ or 'MINGW64' in os.environ or 'MINGW32' in os.environ: os.system('clear') else: os.system('cls') else: os.system('clear') def init_environment(db_manager, ai_manager): """어플리케이션 실행될 수 있도록 DB connection 생성 및 AI Provider 생성""" try: # 데이터베이스 연결 초기화 db_manager.constructor() # AI Provider 초기화 ai_manager.constructor() logger.info("환경 초기화가 완료되었습니다.") except Exception as e: logger.error(f"환경 초기화 중 오류가 발생했습니다: {e}") raise def convert_for_json_serialization(obj): """JSON 직렬화를 위해 모든 데이터 타입을 변환""" if obj is None: return None # 날짜/시간 타입을 문자열로 변환 import datetime if isinstance(obj, (datetime.date, datetime.datetime)): return obj.isoformat() # Decimal 타입을 float로 변환 from decimal import Decimal if isinstance(obj, Decimal): return float(obj) # 바이너리 데이터를 16진수 문자열로 변환 if isinstance(obj, bytes): return obj.hex() # 딕셔너리와 리스트는 재귀적으로 처리 elif isinstance(obj, dict): return {k: convert_for_json_serialization(v) for k, v in obj.items()} elif isinstance(obj, list): return [convert_for_json_serialization(item) for item in obj] # 다른 타입은 그대로 반환 else: return obj def convert_decimal_in_result(obj): """결과 데이터에서 Decimal 타입을 float로, date 타입을 문자열로 변환 (기존 호환성 유지)""" return convert_for_json_serialization(obj) import datetime if isinstance(obj, Decimal): return float(obj) elif isinstance(obj, (datetime.date, datetime.datetime)): return obj.isoformat() elif isinstance(obj, dict): return {k: convert_decimal_in_result(v) for k, v in obj.items()} elif isinstance(obj, list): return [convert_decimal_in_result(item) for item in obj] else: return obj def check_init_environment(db_manager, args, ai_manager, config): # 데이터베이스 연결 확인 if not db_manager.is_connected(): logger.error("데이터베이스에 연결할 수 없습니다.") logger.error("환경 변수를 확인하거나 .env 파일을 설정해주세요.") sys.exit(1) logger.info("MySQL Hub Server를 시작합니다.") logger.info(f"실행 모드: {args}") logger.info(f"AI Provider: {ai_manager.get_current_provider()}") logger.info(f"HTTP 서버: http://{config.HTTP_SERVER_HOST}:{config.HTTP_SERVER_PORT}") logger.info(f"log level : {config.LOG_LEVEL}") # json을 입력받아 json format에 맞게 string으로 변환해서 리턴하는 함수 def json_to_pretty_string(data): """ json 객체를 예쁘게 포맷된 문자열로 변환하여 반환합니다. Args: data (dict or list): JSON 객체 Returns: str: 예쁘게 포맷된 JSON 문자열 """ try: import json # Response 객체인 경우 model_dump() 사용 if hasattr(data, 'model_dump'): data = data.model_dump() # 공통 변환 함수를 사용하여 모든 데이터 타입 변환 converted_data = convert_for_json_serialization(data) return json.dumps(converted_data, ensure_ascii=False, indent=2, sort_keys=True) except Exception as e: return f"JSON 변환 오류: {e}"