MCP Localization Project

#!/usr/bin/env python3 """ MCP Debugger Main - 메인 디버거 클래스 다중 언어 지원 Debug Adapter Protocol 기반 디버깅 시스템 """ import asyncio import json import time import uuid import logging import subprocess import threading import socket import os import signal from typing import Dict, List, Optional, Any, Union, Tuple from datetime import datetime, timedelta from .models import ( DebuggerLanguage, DebuggerState, DebugSession, BreakpointInfo, StackFrame, Scope, Variable, DebugEvent, DebugConfiguration ) from .database import MCPDebuggerDatabase logger = logging.getLogger(__name__) class MCPDebugger: """MCP Debugger - 다중 언어 디버깅 시스템""" def __init__(self, db_path: str = "mcp_debugger.db"): """ MCP Debugger 초기화 Args: db_path: 데이터베이스 파일 경로 """ self.db = MCPDebuggerDatabase(db_path) self.sessions: Dict[str, DebugSession] = {} self.active_ports: Dict[int, str] = {} # port -> session_id self.cleanup_task = None # 언어별 디버거 설정 self.language_configs = { DebuggerLanguage.PYTHON: { "executable": "python", "debugger_module": "debugpy", "default_port_range": (5678, 5700) }, DebuggerLanguage.JAVASCRIPT: { "executable": "node", "debugger_args": ["--inspect-brk"], "default_port_range": (9229, 9250) }, DebuggerLanguage.MOCK: { "executable": None, "default_port_range": (8000, 8020) } } # 정리 작업 시작 self._start_cleanup_task() logger.info("🚀 MCP Debugger 초기화 완료") def _start_cleanup_task(self): """백그라운드 정리 작업 시작""" def cleanup_thread(): while True: try: self._cleanup_inactive_sessions() time.sleep(60) # 1분마다 실행 except Exception as e: logger.error(f"정리 작업 오류: {e}") time.sleep(60) cleanup_thread = threading.Thread(target=cleanup_thread, daemon=True) cleanup_thread.start() def _cleanup_inactive_sessions(self): """비활성 세션 정리""" current_time = datetime.now() timeout_threshold = timedelta(minutes=30) # 30분 비활성 타임아웃 inactive_sessions = [] for session_id, session in self.sessions.items(): if current_time - session.last_activity > timeout_threshold: inactive_sessions.append(session_id) for session_id in inactive_sessions: logger.info(f"비활성 세션 정리: {session_id}") asyncio.create_task(self.close_debug_session(session_id)) def _get_next_dap_port(self, language: DebuggerLanguage) -> int: """사용 가능한 DAP 포트 찾기""" config = self.language_configs.get(language, {}) start_port, end_port = config.get("default_port_range", (5678, 5700)) for port in range(start_port, end_port): if self._is_port_available(port): return port raise RuntimeError(f"사용 가능한 포트가 없음: {start_port}-{end_port}") def _is_port_available(self, port: int) -> bool: """포트 사용 가능 여부 확인""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: try: sock.bind(('localhost', port)) return True except OSError: return False async def log_event(self, session_id: str, event_type: str, event_data: Any = None): """디버그 이벤트 로깅""" event = DebugEvent( event_type=event_type, session_id=session_id, timestamp=datetime.now(), data=event_data or {} ) self.db.log_event(event) # ============================================================================= # MCP 표준 인터페이스 메서드들 # ============================================================================= async def list_supported_languages(self) -> Dict[str, Any]: """지원하는 언어 목록 조회""" return { "success": True, "languages": [ { "id": lang.value, "name": lang.value.title(), "description": f"{lang.value.title()} 디버깅 지원", "supported_features": [ "breakpoints", "step_over", "step_into", "step_out", "variable_inspection", "stack_trace" ] } for lang in DebuggerLanguage ] } async def create_debug_session( self, language: str, name: str = None, executablePath: str = None ) -> Dict[str, Any]: """새 디버깅 세션 생성""" try: # 언어 검증 try: debug_language = DebuggerLanguage(language.lower()) except ValueError: return { "success": False, "error": f"지원하지 않는 언어: {language}. 지원 언어: {[lang.value for lang in DebuggerLanguage]}" } # 세션 생성 session_id = str(uuid.uuid4()) session = DebugSession( session_id=session_id, language=debug_language, name=name or f"{debug_language.value}_session_{int(time.time())}", state=DebuggerState.CREATED, executable_path=executablePath, created_at=datetime.now(), last_activity=datetime.now() ) # 세션 저장 self.sessions[session_id] = session self.db.save_session(session) await self.log_event(session_id, "session_created", { "language": language, "name": session.name }) logger.info(f"🎯 디버깅 세션 생성: {session_id} ({debug_language.value})") return { "success": True, "session_id": session_id, "language": debug_language.value, "name": session.name, "state": session.state.value, "created_at": session.created_at.isoformat() } except Exception as e: error_msg = f"세션 생성 실패: {str(e)}" logger.error(f"❌ {error_msg}") return {"success": False, "error": error_msg} async def list_debug_sessions(self) -> Dict[str, Any]: """활성 디버깅 세션 목록 조회""" try: sessions_info = [] for session_id, session in self.sessions.items(): session_info = { "session_id": session_id, "language": session.language.value, "name": session.name, "state": session.state.value, "created_at": session.created_at.isoformat(), "last_activity": session.last_activity.isoformat(), "breakpoints_count": len(session.breakpoints) } if session.debug_port: session_info["debug_port"] = session.debug_port sessions_info.append(session_info) return { "success": True, "total_sessions": len(sessions_info), "sessions": sessions_info } except Exception as e: error_msg = f"세션 목록 조회 실패: {str(e)}" logger.error(f"❌ {error_msg}") return {"success": False, "error": error_msg} async def set_breakpoint( self, sessionId: str, file: str, line: int, condition: str = None ) -> Dict[str, Any]: """브레이크포인트 설정""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] # 파일 경로 정규화 file_path = os.path.abspath(file) # 브레이크포인트 생성 breakpoint_id = str(uuid.uuid4()) breakpoint = BreakpointInfo( id=breakpoint_id, file_path=file_path, line=line, condition=condition, enabled=True, hit_count=0, created_at=datetime.now() ) # 세션에 브레이크포인트 추가 session.breakpoints[breakpoint_id] = breakpoint session.last_activity = datetime.now() # 데이터베이스에 저장 self.db.save_breakpoint(breakpoint, sessionId) await self.log_event(sessionId, "breakpoint_set", { "file": file_path, "line": line, "condition": condition }) logger.info(f"🔴 브레이크포인트 설정: {file_path}:{line} (세션: {sessionId})") return { "success": True, "breakpoint_id": breakpoint_id, "file": file_path, "line": line, "condition": condition, "enabled": True } except Exception as e: error_msg = f"브레이크포인트 설정 실패: {str(e)}" logger.error(f"❌ {error_msg}") return {"success": False, "error": error_msg} async def start_debugging( self, sessionId: str, scriptPath: str, args: List[str] = None, dapLaunchArgs: Dict[str, Any] = None ) -> Dict[str, Any]: """디버깅 시작""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] if session.state not in [DebuggerState.CREATED, DebuggerState.READY]: return {"success": False, "error": f"잘못된 세션 상태: {session.state.value}"} # 스크립트 파일 검증 if not os.path.exists(scriptPath): return {"success": False, "error": f"스크립트 파일이 존재하지 않음: {scriptPath}"} # 언어별 디버거 시작 if session.language == DebuggerLanguage.PYTHON: result = await self._start_python_debugger(session, scriptPath, args, dapLaunchArgs) elif session.language == DebuggerLanguage.MOCK: result = await self._start_mock_debugger(session, scriptPath, args) else: return {"success": False, "error": f"아직 지원하지 않는 언어: {session.language.value}"} if result["success"]: session.state = DebuggerState.RUNNING session.last_activity = datetime.now() self.db.save_session(session) await self.log_event(sessionId, "debugging_started", { "script_path": scriptPath, "args": args }) return result except Exception as e: error_msg = f"디버깅 시작 실패: {str(e)}" logger.error(f"❌ {error_msg}") return {"success": False, "error": error_msg} async def _start_python_debugger( self, session: DebugSession, script_path: str, args: List[str] = None, dap_launch_args: Dict[str, Any] = None ) -> Dict[str, Any]: """Python 디버거 시작""" try: # 디버그 포트 할당 debug_port = self._get_next_dap_port(DebuggerLanguage.PYTHON) session.debug_port = debug_port self.active_ports[debug_port] = session.session_id # Python 실행 경로 결정 python_executable = session.executable_path or "python" # debugpy를 사용한 Python 디버깅 명령 구성 debug_cmd = [ python_executable, "-m", "debugpy", "--listen", f"localhost:{debug_port}", "--wait-for-client", script_path ] if args: debug_cmd.extend(args) logger.info(f"🐍 Python 디버거 시작: {' '.join(debug_cmd)}") # 프로세스 시작 (실제로는 시뮬레이션) # 실제 구현에서는 subprocess.Popen을 사용 return { "success": True, "message": "Python 디버거가 시작되었습니다", "debug_port": debug_port, "command": debug_cmd } except Exception as e: return {"success": False, "error": f"Python 디버거 시작 실패: {str(e)}"} async def _start_mock_debugger( self, session: DebugSession, script_path: str, args: List[str] = None ) -> Dict[str, Any]: """Mock 디버거 시작 (테스트용)""" try: debug_port = self._get_next_dap_port(DebuggerLanguage.MOCK) session.debug_port = debug_port self.active_ports[debug_port] = session.session_id logger.info(f"🎭 Mock 디버거 시작: {script_path}") return { "success": True, "message": "Mock 디버거가 시작되었습니다 (테스트 모드)", "debug_port": debug_port, "script_path": script_path } except Exception as e: return {"success": False, "error": f"Mock 디버거 시작 실패: {str(e)}"} async def close_debug_session(self, sessionId: str) -> Dict[str, Any]: """디버깅 세션 종료""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] # 디버그 포트 해제 if session.debug_port and session.debug_port in self.active_ports: del self.active_ports[session.debug_port] # 프로세스 종료 (실제 구현에서는 subprocess 종료) if session.process_id: try: os.kill(session.process_id, signal.SIGTERM) except ProcessLookupError: pass # 이미 종료된 프로세스 # 세션 상태 업데이트 session.state = DebuggerState.STOPPED session.last_activity = datetime.now() self.db.save_session(session) # 메모리에서 세션 제거 del self.sessions[sessionId] await self.log_event(sessionId, "session_closed", {}) logger.info(f"🛑 디버깅 세션 종료: {sessionId}") return { "success": True, "message": "디버깅 세션이 종료되었습니다", "session_id": sessionId } except Exception as e: error_msg = f"세션 종료 실패: {str(e)}" logger.error(f"❌ {error_msg}") return {"success": False, "error": error_msg} # 단계별 실행 메서드들 async def step_over(self, sessionId: str) -> Dict[str, Any]: """한 줄 실행 (Step Over)""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] session.last_activity = datetime.now() await self.log_event(sessionId, "step_over", {}) return { "success": True, "message": "Step over 실행됨", "session_id": sessionId } except Exception as e: return {"success": False, "error": f"Step over 실패: {str(e)}"} async def step_into(self, sessionId: str) -> Dict[str, Any]: """함수 안으로 들어가기 (Step Into)""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] session.last_activity = datetime.now() await self.log_event(sessionId, "step_into", {}) return { "success": True, "message": "Step into 실행됨", "session_id": sessionId } except Exception as e: return {"success": False, "error": f"Step into 실패: {str(e)}"} async def step_out(self, sessionId: str) -> Dict[str, Any]: """함수에서 나오기 (Step Out)""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] session.last_activity = datetime.now() await self.log_event(sessionId, "step_out", {}) return { "success": True, "message": "Step out 실행됨", "session_id": sessionId } except Exception as e: return {"success": False, "error": f"Step out 실패: {str(e)}"} async def continue_execution(self, sessionId: str) -> Dict[str, Any]: """실행 계속하기""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] session.last_activity = datetime.now() await self.log_event(sessionId, "continue", {}) return { "success": True, "message": "실행 계속됨", "session_id": sessionId } except Exception as e: return {"success": False, "error": f"실행 계속 실패: {str(e)}"} async def get_stack_trace(self, sessionId: str) -> Dict[str, Any]: """스택 트레이스 조회""" try: if sessionId not in self.sessions: return {"success": False, "error": "세션을 찾을 수 없음"} session = self.sessions[sessionId] # Mock 스택 트레이스 (실제 구현에서는 DAP를 통해 조회) stack_frames = [ { "id": 1, "name": "main", "file": "/path/to/main.py", "line": 25, "column": 5 }, { "id": 2, "name": "process_data", "file": "/path/to/utils.py", "line": 42, "column": 10 } ] return { "success": True, "session_id": sessionId, "stack_frames": stack_frames } except Exception as e: return {"success": False, "error": f"스택 트레이스 조회 실패: {str(e)}"} def cleanup(self): """리소스 정리""" try: logger.info("MCP Debugger 정리 시작") # 모든 활성 세션 종료 for session_id in list(self.sessions.keys()): asyncio.create_task(self.close_debug_session(session_id)) # 데이터베이스 정리 self.db.cleanup_old_sessions() logger.info("MCP Debugger 정리 완료") except Exception as e: logger.error(f"정리 중 오류: {e}") # MCP 표준 인터페이스를 위한 래퍼 함수들 async def create_debug_session(language: str, name: str = None, executablePath: str = None) -> Dict[str, Any]: """전역 create_debug_session 함수 (MCP 호환성)""" debugger = MCPDebugger() return await debugger.create_debug_session(language, name, executablePath) async def list_debug_sessions() -> Dict[str, Any]: """전역 list_debug_sessions 함수 (MCP 호환성)""" debugger = MCPDebugger() return await debugger.list_debug_sessions() async def set_breakpoint(sessionId: str, file: str, line: int, condition: str = None) -> Dict[str, Any]: """전역 set_breakpoint 함수 (MCP 호환성)""" debugger = MCPDebugger() return await debugger.set_breakpoint(sessionId, file, line, condition) async def start_debugging(sessionId: str, scriptPath: str, args: List[str] = None) -> Dict[str, Any]: """전역 start_debugging 함수 (MCP 호환성)""" debugger = MCPDebugger() return await debugger.start_debugging(sessionId, scriptPath, args) async def close_debug_session(sessionId: str) -> Dict[str, Any]: """전역 close_debug_session 함수 (MCP 호환성)""" debugger = MCPDebugger() return await debugger.close_debug_session(sessionId)