MCP Localization Project

#!/usr/bin/env python3 """ Filesystem MCP Main - 통합 실행 파일 Complete Filesystem MCP의 모든 기능을 통합하여 제공하는 메인 클래스입니다. - 3개 모듈 (base, core, advanced) 통합 - 12개 핵심 기능 제공 - 테스트 및 검증 기능 포함 """ import os import sys import json import tempfile from pathlib import Path from typing import Dict, List, Optional, Any from datetime import datetime # 분할된 모듈들 import from filesystem_base import FilesystemMCPBase, PermissionLevel, FileOperation, RootConfig from filesystem_core import FilesystemMCPCore from filesystem_advanced import FilesystemMCPAdvanced class CompleteFilesystemMCP(FilesystemMCPCore, FilesystemMCPAdvanced): """ Complete Filesystem MCP 통합 클래스 FilesystemMCPBase를 기반으로 Core와 Advanced 기능을 모두 상속받아 완전한 파일시스템 MCP 서버 기능을 제공합니다. """ def __init__(self, db_path: Optional[str] = None): """ CompleteFilesystemMCP 초기화 Args: db_path: SQLite 데이터베이스 파일 경로 (None 시 임시 파일 생성) """ # 기본 클래스 초기화 (FilesystemMCPBase) FilesystemMCPBase.__init__(self, db_path) self.logger.info("CompleteFilesystemMCP 통합 클래스 초기화 완료") def get_capabilities(self) -> Dict[str, Any]: """ MCP 서버 능력 정보 반환 Returns: Dict: 서버가 제공하는 기능들의 목록과 설명 """ return { "name": "Complete Filesystem MCP", "version": "1.0.0", "description": "완전한 파일시스템 MCP 서버 - 12개 핵심 기능 제공", "capabilities": { "file_operations": { "read": "파일 읽기 (head/tail 지원)", "write": "파일 쓰기 (원자적 쓰기)", "edit": "파일 편집 (텍스트 교체)", "create_directory": "디렉토리 생성", "list_directory": "디렉토리 목록 조회", "list_directory_with_sizes": "크기 포함 디렉토리 목록", "directory_tree": "디렉토리 트리 구조", "move_file": "파일 이동/이름 변경", "search_files": "파일 검색", "get_file_info": "파일 정보 조회", "list_allowed_directories": "허용된 디렉토리 목록", "read_multiple_files": "여러 파일 읽기" }, "security": { "roots_protocol": "경로 기반 권한 관리", "audit_logging": "모든 작업 감사 로그", "atomic_operations": "원자적 파일 작업", "permission_checking": "세밀한 권한 확인" }, "metadata": { "file_hashing": "파일 해시 계산", "mime_detection": "MIME 타입 감지", "symlink_support": "심볼릭 링크 지원", "timestamp_tracking": "생성/수정 시간 추적" } }, "functions": [ "filesystem__read_file", "read_multiple_files", "filesystem__write_file", "edit_file", "create_directory", "filesystem__list_directory", "list_directory_with_sizes", "directory_tree", "move_file", "search_files", "get_file_info", "list_allowed_directories" ] } def get_status(self) -> Dict[str, Any]: """ MCP 서버 상태 정보 반환 Returns: Dict: 서버 상태 정보 """ try: # Root 통계 total_roots = len(self.roots) accessible_roots = sum(1 for config in self.roots.values() if os.path.exists(config.path) and os.access(config.path, os.R_OK)) # 권한 레벨별 통계 permission_stats = {} for level in PermissionLevel: count = sum(1 for config in self.roots.values() if config.permission_level == level) permission_stats[level.value] = count # 감사 로그 통계 audit_count = len(self.audit_log) return { "status": "running", "timestamp": datetime.now().isoformat(), "database": { "path": self.db_path, "exists": os.path.exists(self.db_path) }, "roots": { "total": total_roots, "accessible": accessible_roots, "permission_distribution": permission_stats }, "audit": { "log_entries": audit_count, "enabled": True }, "memory": { "thread_safe": True, "lock_available": self.file_lock is not None } } except Exception as e: self.logger.error(f"상태 정보 수집 실패: {e}") return { "status": "error", "error": str(e), "timestamp": datetime.now().isoformat() } def test_all_functions(self) -> Dict[str, Any]: """ 모든 MCP 기능들을 테스트 Returns: Dict: 테스트 결과 보고서 """ test_results = { "timestamp": datetime.now().isoformat(), "total_tests": 0, "passed_tests": 0, "failed_tests": 0, "results": {} } # 테스트용 임시 디렉토리 생성 test_dir = tempfile.mkdtemp(prefix="filesystem_mcp_test_") test_file = os.path.join(test_dir, "test_file.txt") test_content = "Hello, World!\nThis is a test file." # 테스트용 임시 Root 추가 temp_root = RootConfig( name="temp_test_directory", path="/tmp", permission_level=PermissionLevel.FULL_ACCESS, allowed_operations=[ FileOperation.READ, FileOperation.WRITE, FileOperation.CREATE, FileOperation.DELETE, FileOperation.MOVE, FileOperation.SEARCH ], max_file_size=10 * 1024 * 1024, # 10MB audit_enabled=True ) self.add_root(temp_root) try: self.logger.info(f"테스트 시작: {test_dir}") # 1. 파일 쓰기 테스트 test_results["total_tests"] += 1 try: result = self.filesystem__write_file(test_file, test_content) if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["write_file"] = {"status": "PASS", "details": "파일 쓰기 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["write_file"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["write_file"] = {"status": "ERROR", "error": str(e)} # 2. 파일 읽기 테스트 test_results["total_tests"] += 1 try: result = self.filesystem__read_file(test_file) if result["success"] and result["content"] == test_content: test_results["passed_tests"] += 1 test_results["results"]["read_file"] = {"status": "PASS", "details": "파일 읽기 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["read_file"] = {"status": "FAIL", "error": "내용 불일치"} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["read_file"] = {"status": "ERROR", "error": str(e)} # 3. 파일 편집 테스트 test_results["total_tests"] += 1 try: result = self.edit_file(test_file, "Hello", "Hi") if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["edit_file"] = {"status": "PASS", "details": "파일 편집 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["edit_file"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["edit_file"] = {"status": "ERROR", "error": str(e)} # 4. 디렉토리 목록 테스트 test_results["total_tests"] += 1 try: result = self.filesystem__list_directory(test_dir) if result["success"] and any(item["name"] == "test_file.txt" for item in result["items"]): test_results["passed_tests"] += 1 test_results["results"]["list_directory"] = {"status": "PASS", "details": "디렉토리 목록 조회 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["list_directory"] = {"status": "FAIL", "error": "파일 찾을 수 없음"} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["list_directory"] = {"status": "ERROR", "error": str(e)} # 5. 파일 정보 테스트 test_results["total_tests"] += 1 try: result = self.get_file_info(test_file) if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["get_file_info"] = {"status": "PASS", "details": "파일 정보 조회 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["get_file_info"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["get_file_info"] = {"status": "ERROR", "error": str(e)} # 6. 파일 검색 테스트 test_results["total_tests"] += 1 try: result = self.search_files(test_dir, "*.txt") if result["success"] and result["total_files"] > 0: test_results["passed_tests"] += 1 test_results["results"]["search_files"] = {"status": "PASS", "details": f"{result['total_files']}개 파일 발견"} else: test_results["failed_tests"] += 1 test_results["results"]["search_files"] = {"status": "FAIL", "error": "파일 찾을 수 없음"} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["search_files"] = {"status": "ERROR", "error": str(e)} # 7. 허용된 디렉토리 목록 테스트 test_results["total_tests"] += 1 try: result = self.list_allowed_directories() if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["list_allowed_directories"] = { "status": "PASS", "details": f"{result['total_directories']}개 디렉토리 허용됨" } else: test_results["failed_tests"] += 1 test_results["results"]["list_allowed_directories"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["list_allowed_directories"] = {"status": "ERROR", "error": str(e)} # 8. 디렉토리 트리 테스트 test_results["total_tests"] += 1 try: result = self.directory_tree(test_dir, max_depth=2) if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["directory_tree"] = {"status": "PASS", "details": "디렉토리 트리 생성 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["directory_tree"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["directory_tree"] = {"status": "ERROR", "error": str(e)} # 9. 파일 이동 테스트 test_results["total_tests"] += 1 try: new_test_file = os.path.join(test_dir, "moved_test_file.txt") result = self.move_file(test_file, new_test_file) if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["move_file"] = {"status": "PASS", "details": "파일 이동 성공"} test_file = new_test_file # 다음 테스트를 위해 경로 업데이트 else: test_results["failed_tests"] += 1 test_results["results"]["move_file"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["move_file"] = {"status": "ERROR", "error": str(e)} # 10. 여러 파일 읽기 테스트 (추가 파일 생성 후) test_results["total_tests"] += 1 try: test_file2 = os.path.join(test_dir, "test_file2.txt") self.filesystem__write_file(test_file2, "Second test file") result = self.read_multiple_files([test_file, test_file2]) if result["success"] and result["successful_reads"] == 2: test_results["passed_tests"] += 1 test_results["results"]["read_multiple_files"] = {"status": "PASS", "details": "여러 파일 읽기 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["read_multiple_files"] = {"status": "FAIL", "error": "일부 파일 읽기 실패"} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["read_multiple_files"] = {"status": "ERROR", "error": str(e)} # 11. 디렉토리 생성 테스트 test_results["total_tests"] += 1 try: new_dir = os.path.join(test_dir, "subdir", "nested") result = self.create_directory(new_dir, parents=True) if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["create_directory"] = {"status": "PASS", "details": "디렉토리 생성 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["create_directory"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["create_directory"] = {"status": "ERROR", "error": str(e)} # 12. 크기 포함 디렉토리 목록 테스트 test_results["total_tests"] += 1 try: result = self.list_directory_with_sizes(test_dir) if result["success"]: test_results["passed_tests"] += 1 test_results["results"]["list_directory_with_sizes"] = {"status": "PASS", "details": "크기 포함 목록 조회 성공"} else: test_results["failed_tests"] += 1 test_results["results"]["list_directory_with_sizes"] = {"status": "FAIL", "error": result.get("error")} except Exception as e: test_results["failed_tests"] += 1 test_results["results"]["list_directory_with_sizes"] = {"status": "ERROR", "error": str(e)} finally: # 테스트 디렉토리 정리 try: import shutil shutil.rmtree(test_dir) self.logger.info(f"테스트 디렉토리 정리 완료: {test_dir}") except Exception as e: self.logger.warning(f"테스트 디렉토리 정리 실패: {e}") # 결과 요약 test_results["success_rate"] = (test_results["passed_tests"] / test_results["total_tests"]) * 100 test_results["summary"] = f"{test_results['passed_tests']}/{test_results['total_tests']} 테스트 통과 ({test_results['success_rate']:.1f}%)" self.logger.info(f"테스트 완료: {test_results['summary']}") return test_results def main(): """메인 실행 함수 - 테스트 및 예제 실행""" print("Complete Filesystem MCP 통합 테스트") print("=" * 50) # MCP 인스턴스 생성 mcp = CompleteFilesystemMCP() try: # 1. 능력 정보 출력 print("\n1. MCP 서버 능력 정보:") capabilities = mcp.get_capabilities() print(f" - 이름: {capabilities['name']}") print(f" - 버전: {capabilities['version']}") print(f" - 기능 수: {len(capabilities['functions'])}개") # 2. 상태 정보 출력 print("\n2. MCP 서버 상태 정보:") status = mcp.get_status() print(f" - 상태: {status['status']}") print(f" - Root 개수: {status['roots']['total']}개") print(f" - 접근 가능: {status['roots']['accessible']}개") # 3. 전체 기능 테스트 실행 print("\n3. 전체 기능 테스트 실행:") test_results = mcp.test_all_functions() print(f" - {test_results['summary']}") # 4. 실패한 테스트 상세 정보 failed_tests = [name for name, result in test_results['results'].items() if result['status'] != 'PASS'] if failed_tests: print(f"\n⚠️ 실패한 테스트 ({len(failed_tests)}개):") for test_name in failed_tests: result = test_results['results'][test_name] print(f" - {test_name}: {result['status']} - {result.get('error', 'Unknown error')}") else: print("\n✅ 모든 테스트가 성공적으로 완료되었습니다!") print(f"\n📊 테스트 결과 요약:") print(f" 총 테스트: {test_results['total_tests']}개") print(f" 성공: {test_results['passed_tests']}개") print(f" 실패: {test_results['failed_tests']}개") print(f" 성공률: {test_results['success_rate']:.1f}%") except Exception as e: print(f"\n❌ 테스트 실행 중 오류 발생: {e}") return 1 finally: # 리소스 정리 mcp.cleanup() print(f"\n🧹 리소스 정리 완료") return 0 if __name__ == "__main__": sys.exit(main())