server.py•3.75 kB
"""MCP 시장 통계 서버"""
import logging
from typing import Dict, Any, Optional
from src.utils.database import DatabaseManager
from src.utils.cache import CacheManager
from src.exceptions import DatabaseConnectionError, CacheError
class MarketStatsServer:
"""시장 통계 MCP 서버"""
def __init__(self, config: Dict[str, Any]):
self.name = config["name"]
self.version = config["version"]
self.config = config
self.tools: Dict[str, Any] = {}
self.is_running = False
# 데이터베이스 매니저 초기화
self.db_manager = DatabaseManager(config["database"])
# 캐시 매니저 초기화
self.cache_manager = CacheManager(config["redis"])
# 로깅 설정
self.logger = logging.getLogger(__name__)
def register_tool(self, name: str, tool: Any) -> None:
"""도구 등록"""
if name in self.tools:
raise ValueError(f"Tool '{name}' already registered")
self.tools[name] = tool
self.logger.info(f"Tool '{name}' registered")
def get_available_tools(self) -> Dict[str, Any]:
"""등록된 도구 목록 반환"""
return self.tools.copy()
async def startup(self) -> None:
"""서버 시작"""
try:
# 데이터베이스 연결
await self.db_manager.connect()
self.logger.info("Database connected")
# 캐시 연결
await self.cache_manager.connect()
self.logger.info("Cache connected")
self.is_running = True
self.logger.info("Market Stats Server started")
except (DatabaseConnectionError, CacheError) as e:
self.logger.error(f"Server startup failed: {e}")
raise
async def shutdown(self) -> None:
"""서버 종료"""
try:
# 데이터베이스 연결 해제
await self.db_manager.disconnect()
self.logger.info("Database disconnected")
# 캐시 연결 해제
await self.cache_manager.disconnect()
self.logger.info("Cache disconnected")
self.is_running = False
self.logger.info("Market Stats Server stopped")
except Exception as e:
self.logger.error(f"Server shutdown error: {e}")
raise
async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""도구 실행"""
if tool_name not in self.tools:
raise KeyError(f"Tool '{tool_name}' not found")
tool = self.tools[tool_name]
try:
result = await tool.execute(arguments)
self.logger.info(f"Tool '{tool_name}' executed successfully")
return result
except Exception as e:
self.logger.error(f"Tool '{tool_name}' execution failed: {e}")
raise
async def health_check(self) -> Dict[str, str]:
"""서버 상태 확인"""
health_status = {
"status": "healthy",
"database": "disconnected",
"cache": "disconnected"
}
# 데이터베이스 상태 확인
if await self.db_manager.health_check():
health_status["database"] = "connected"
else:
health_status["status"] = "unhealthy"
# 캐시 상태 확인
if await self.cache_manager.health_check():
health_status["cache"] = "connected"
else:
health_status["status"] = "unhealthy"
return health_status