MCP Localization Project

#!/usr/bin/env python3 """ Context7 MCP 로컬화 - 메인 통합 클래스 Context7의 모든 기능을 통합하여 MCP 인터페이스 제공 주요 기능: - Context7MCP: 통합 MCP 클래스 - resolve_library_id와 get_library_docs 통합 제공 - 로컬 데이터베이스 관리 - 웹 데이터 소스 통합 - 캐싱 및 최적화 """ import asyncio import json import time import logging from typing import Dict, List, Optional, Any, Union from dataclasses import dataclass, asdict from pathlib import Path from context7_base import Context7Database, LibraryInfo, LibraryDocs, Context7Utils from context7_resolver import LibraryResolver, ResolveResult from context7_docs import LibraryDocsProvider, DocResult logger = logging.getLogger(__name__) @dataclass class MCPResponse: """MCP 응답 형식""" success: bool data: Any message: str metadata: Dict[str, Any] class Context7MCP: """Context7 MCP 통합 클래스""" def __init__(self, db_path: str = "context7_mcp.db"): """ Context7 MCP 초기화 Args: db_path: 데이터베이스 파일 경로 """ self.db_path = db_path self.db = Context7Database(db_path) self.resolver = LibraryResolver(self.db) self.docs_provider = LibraryDocsProvider(self.db) # 통계 정보 self.stats = { "resolve_calls": 0, "docs_calls": 0, "cache_hits": 0, "web_fetches": 0, "errors": 0 } logger.info(f"🚀 Context7 MCP 초기화 완료 (DB: {db_path})") async def resolve_library_id(self, library_name: str) -> MCPResponse: """ 라이브러리 이름을 Context7 호환 ID로 해결 Args: library_name: 검색할 라이브러리 이름 Returns: MCPResponse: 해결 결과 """ try: self.stats["resolve_calls"] += 1 start_time = time.time() logger.info(f"🔍 resolve_library_id 호출: '{library_name}'") # 입력 검증 if not library_name or not library_name.strip(): return MCPResponse( success=False, data=None, message="Library name cannot be empty", metadata={"error_type": "validation_error"} ) # 라이브러리 ID 해결 result = await self.resolver.resolve_library_id(library_name.strip()) # 응답 데이터 구성 response_data = { "selected_library_id": result.selected_library_id, "explanation": result.explanation, "confidence": result.confidence, "matches": [ { "library_id": match.library_id, "name": match.name, "description": match.description, "version": match.version, "trust_score": match.trust_score, "code_snippets": match.code_snippets } for match in result.matches ] } # 성공 여부 판단 success = bool(result.selected_library_id) message = "Library ID resolved successfully" if success else "No matching library found" if not success: self.stats["errors"] += 1 execution_time = time.time() - start_time return MCPResponse( success=success, data=response_data, message=message, metadata={ "execution_time": round(execution_time, 3), "matches_count": len(result.matches), "confidence": result.confidence } ) except Exception as e: self.stats["errors"] += 1 logger.error(f"❌ resolve_library_id 오류: {e}") return MCPResponse( success=False, data=None, message=f"Error resolving library ID: {str(e)}", metadata={"error_type": "internal_error", "error": str(e)} ) async def get_library_docs( self, library_id: str, topic: str = "", tokens: int = 10000 ) -> MCPResponse: """ 라이브러리 문서 가져오기 Args: library_id: Context7 호환 라이브러리 ID topic: 특정 주제 (선택사항) tokens: 최대 토큰 수 Returns: MCPResponse: 문서 검색 결과 """ try: self.stats["docs_calls"] += 1 start_time = time.time() logger.info(f"📚 get_library_docs 호출: {library_id} (topic: {topic})") # 입력 검증 if not library_id or not library_id.strip(): return MCPResponse( success=False, data=None, message="Library ID cannot be empty", metadata={"error_type": "validation_error"} ) # 토큰 수 검증 if tokens <= 0 or tokens > 50000: tokens = 10000 # 기본값으로 설정 # 캐시 확인 cached_docs = await self.db.get_cached_docs(library_id, topic) if cached_docs: self.stats["cache_hits"] += 1 logger.info(f"💾 캐시에서 문서 반환: {library_id}") else: self.stats["web_fetches"] += 1 # 라이브러리 문서 가져오기 result = await self.docs_provider.get_library_docs(library_id, topic, tokens) # 응답 데이터 구성 response_data = { "library_id": result.library_id, "content": result.content, "topic": result.topic, "tokens": result.tokens, "source": result.source, "metadata": result.metadata } execution_time = time.time() - start_time if not result.success: self.stats["errors"] += 1 return MCPResponse( success=result.success, data=response_data, message="Documentation retrieved successfully" if result.success else "Failed to retrieve documentation", metadata={ "execution_time": round(execution_time, 3), "content_length": len(result.content), "tokens_used": result.tokens, "source_type": result.source, "from_cache": bool(cached_docs) } ) except Exception as e: self.stats["errors"] += 1 logger.error(f"❌ get_library_docs 오류: {e}") return MCPResponse( success=False, data=None, message=f"Error retrieving documentation: {str(e)}", metadata={"error_type": "internal_error", "error": str(e)} ) async def get_library_suggestions(self, partial_name: str, limit: int = 5) -> MCPResponse: """ 부분 이름으로 라이브러리 제안 Args: partial_name: 부분 라이브러리 이름 limit: 반환할 제안 수 Returns: MCPResponse: 제안 결과 """ try: start_time = time.time() logger.info(f"💡 get_library_suggestions 호출: '{partial_name}'") suggestions = await self.resolver.get_library_suggestions(partial_name, limit) response_data = { "suggestions": [ { "library_id": lib.library_id, "name": lib.name, "description": lib.description, "trust_score": lib.trust_score } for lib in suggestions ] } execution_time = time.time() - start_time return MCPResponse( success=True, data=response_data, message=f"Found {len(suggestions)} suggestions", metadata={ "execution_time": round(execution_time, 3), "suggestions_count": len(suggestions) } ) except Exception as e: logger.error(f"❌ get_library_suggestions 오류: {e}") return MCPResponse( success=False, data=None, message=f"Error getting suggestions: {str(e)}", metadata={"error_type": "internal_error", "error": str(e)} ) async def add_library(self, library_info: Dict[str, Any]) -> MCPResponse: """ 새로운 라이브러리 정보 추가 Args: library_info: 라이브러리 정보 딕셔너리 Returns: MCPResponse: 추가 결과 """ try: logger.info(f"➕ add_library 호출: {library_info.get('library_id')}") # 필수 필드 검증 required_fields = ["library_id", "name"] for field in required_fields: if field not in library_info: return MCPResponse( success=False, data=None, message=f"Missing required field: {field}", metadata={"error_type": "validation_error"} ) # LibraryInfo 객체 생성 lib_info = LibraryInfo( library_id=library_info["library_id"], name=library_info["name"], description=library_info.get("description", ""), version=library_info.get("version", ""), trust_score=library_info.get("trust_score", 7.0), code_snippets=library_info.get("code_snippets", 0), last_updated=library_info.get("last_updated", "") ) # 데이터베이스에 추가 success = await self.db.add_library(lib_info) return MCPResponse( success=success, data={"library_id": lib_info.library_id}, message="Library added successfully" if success else "Failed to add library", metadata={"operation": "add_library"} ) except Exception as e: logger.error(f"❌ add_library 오류: {e}") return MCPResponse( success=False, data=None, message=f"Error adding library: {str(e)}", metadata={"error_type": "internal_error", "error": str(e)} ) async def get_stats(self) -> MCPResponse: """ Context7 통계 정보 가져오기 Returns: MCPResponse: 통계 정보 """ try: # 데이터베이스 통계 추가 with self.db.db_path and True: # 간단한 연결 확인 # 라이브러리 수 조회 import sqlite3 with sqlite3.connect(self.db.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM libraries") library_count = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM docs_cache") cache_count = cursor.fetchone()[0] stats_data = { "runtime_stats": self.stats, "database_stats": { "total_libraries": library_count, "cached_docs": cache_count }, "performance": { "cache_hit_rate": round( self.stats["cache_hits"] / max(self.stats["docs_calls"], 1) * 100, 2 ), "error_rate": round( self.stats["errors"] / max( self.stats["resolve_calls"] + self.stats["docs_calls"], 1 ) * 100, 2 ) } } return MCPResponse( success=True, data=stats_data, message="Statistics retrieved successfully", metadata={"timestamp": time.time()} ) except Exception as e: logger.error(f"❌ get_stats 오류: {e}") return MCPResponse( success=False, data=None, message=f"Error retrieving statistics: {str(e)}", metadata={"error_type": "internal_error", "error": str(e)} ) async def clear_cache(self, library_id: str = "", older_than_hours: int = 24) -> MCPResponse: """ 캐시 정리 Args: library_id: 특정 라이브러리 캐시만 정리 (빈 문자열이면 전체) older_than_hours: 지정된 시간보다 오래된 캐시 정리 Returns: MCPResponse: 정리 결과 """ try: import sqlite3 from datetime import datetime, timedelta cutoff_time = datetime.now() - timedelta(hours=older_than_hours) with sqlite3.connect(self.db.db_path) as conn: cursor = conn.cursor() if library_id: # 특정 라이브러리 캐시 정리 cursor.execute( "DELETE FROM docs_cache WHERE library_id = ? AND expires_at < ?", (library_id, cutoff_time) ) else: # 오래된 캐시 전체 정리 cursor.execute( "DELETE FROM docs_cache WHERE expires_at < ?", (cutoff_time,) ) deleted_count = cursor.rowcount conn.commit() return MCPResponse( success=True, data={"deleted_count": deleted_count}, message=f"Cleared {deleted_count} cache entries", metadata={"operation": "clear_cache", "library_id": library_id} ) except Exception as e: logger.error(f"❌ clear_cache 오류: {e}") return MCPResponse( success=False, data=None, message=f"Error clearing cache: {str(e)}", metadata={"error_type": "internal_error", "error": str(e)} ) async def close(self): """리소스 정리""" try: await self.docs_provider.close_session() logger.info("🔚 Context7 MCP 리소스 정리 완료") except Exception as e: logger.error(f"❌ 리소스 정리 오류: {e}") # Context7MCPServer 클래스 (MCP 서버 호환) class Context7MCPServer: """Context7 MCP 서버 래퍼""" def __init__(self, db_path: str = "context7_mcp.db"): self.context7 = Context7MCP(db_path) async def resolve_library_id(self, library_name: str) -> Dict[str, Any]: """MCP 도구 호환 인터페이스""" response = await self.context7.resolve_library_id(library_name) return asdict(response) async def get_library_docs( self, context7_compatible_library_id: str, topic: str = "", tokens: int = 10000 ) -> Dict[str, Any]: """MCP 도구 호환 인터페이스""" response = await self.context7.get_library_docs( context7_compatible_library_id, topic, tokens ) return asdict(response) async def close(self): """리소스 정리""" await self.context7.close() # 테스트 함수 async def test_context7_mcp(): """Context7 MCP 통합 테스트""" print("🧪 Context7 MCP 통합 테스트 시작...") # Context7 MCP 초기화 context7 = Context7MCP("test_context7_mcp.db") # 1. resolve_library_id 테스트 print("\n🔍 resolve_library_id 테스트...") test_libraries = ["axios", "react", "express", "nonexistent"] for lib_name in test_libraries: print(f"\n테스트: '{lib_name}'") result = await context7.resolve_library_id(lib_name) print(f" ✅ 성공: {result.success}") print(f" 📝 메시지: {result.message}") if result.success and result.data: print(f" 🆔 선택된 ID: {result.data['selected_library_id']}") print(f" 📊 신뢰도: {result.data['confidence']:.2f}") print(f" 🔢 매치 수: {len(result.data['matches'])}") # 2. get_library_docs 테스트 print("\n📚 get_library_docs 테스트...") test_docs = [ ("/axios/axios", ""), ("/axios/axios", "usage"), ("/facebook/react", ""), ("/nonexistent/lib", "") ] for lib_id, topic in test_docs: print(f"\n테스트: {lib_id} (topic: {topic or 'none'})") result = await context7.get_library_docs(lib_id, topic, tokens=5000) print(f" ✅ 성공: {result.success}") print(f" 📝 메시지: {result.message}") if result.success and result.data: print(f" 📄 소스: {result.data['source']}") print(f" 🔢 토큰: {result.data['tokens']}") print(f" 📏 길이: {len(result.data['content'])} 문자") if result.metadata.get("from_cache"): print(f" 💾 캐시에서 로드됨") # 3. 제안 기능 테스트 print("\n💡 get_library_suggestions 테스트...") suggestion_result = await context7.get_library_suggestions("ax", limit=3) print(f"✅ 성공: {suggestion_result.success}") if suggestion_result.success: suggestions = suggestion_result.data["suggestions"] print(f"📝 제안 수: {len(suggestions)}") for suggestion in suggestions: print(f" - {suggestion['name']}: {suggestion['library_id']}") # 4. 통계 정보 테스트 print("\n📊 get_stats 테스트...") stats_result = await context7.get_stats() print(f"✅ 성공: {stats_result.success}") if stats_result.success: stats = stats_result.data print(f"📞 resolve 호출: {stats['runtime_stats']['resolve_calls']}") print(f"📚 docs 호출: {stats['runtime_stats']['docs_calls']}") print(f"💾 캐시 히트: {stats['runtime_stats']['cache_hits']}") print(f"📈 캐시 히트율: {stats['performance']['cache_hit_rate']}%") # 5. MCP 서버 인터페이스 테스트 print("\n🖥️ MCP 서버 인터페이스 테스트...") server = Context7MCPServer("test_server.db") # resolve_library_id 테스트 resolve_result = await server.resolve_library_id("axios") print(f"🔍 MCP resolve 테스트: {resolve_result['success']}") # get_library_docs 테스트 docs_result = await server.get_library_docs("/axios/axios", "usage", 5000) print(f"📚 MCP docs 테스트: {docs_result['success']}") # 리소스 정리 await context7.close() await server.close() print("\n🎯 Context7 MCP 통합 테스트 완료!") # MCP 도구 래퍼 클래스 (tool_wrappers.py와 호환) class Context7Wrapper: """Context7 MCP 도구 래퍼""" def __init__(self, db_path: str = "context7.db"): self.server = Context7MCPServer(db_path) async def resolve_library_id(self, library_name: str) -> Dict[str, Any]: """라이브러리 ID 해결""" return await self.server.resolve_library_id(library_name) async def get_library_docs( self, library_id: str, topic: str = "", tokens: int = 10000 ) -> Dict[str, Any]: """라이브러리 문서 가져오기""" return await self.server.get_library_docs(library_id, topic, tokens) async def close(self): """리소스 정리""" await self.server.close() if __name__ == "__main__": asyncio.run(test_context7_mcp())