Notion MCP Server

import os import asyncio import uvicorn import sys import logging import importlib import pkgutil from pathlib import Path from fastmcp import FastMCP def get_logger(): log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'logs') os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, 'info.log') logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s', handlers=[ logging.FileHandler(log_file, encoding="utf-8"), logging.StreamHandler() ] ) return logging.getLogger("MCP") class MCPServer(FastMCP): def __init__(self, specific_tools=None): super().__init__(name="MCP Server") self.logger = get_logger() self.logger.info("MCP Server 시작") self.specific_tools = specific_tools if specific_tools: self.logger.info(f"지정된 도구들만 로드: {', '.join(specific_tools)}") self.register_api() def register_api(self): """tools 디렉토리의 Python 파일을 로드하고 등록""" tools_dir = Path(__file__).parent / "tools" package_name = "tools" if not tools_dir.exists(): self.logger.error("tools 디렉토리가 존재하지 않습니다.") return # 로드할 도구 파일들 결정 if self.specific_tools: # 지정된 도구들만 로드 python_files = [] for tool_name in self.specific_tools: tool_file = tools_dir / f"{tool_name}.py" if tool_file.exists(): python_files.append(tool_file) else: self.logger.error(f"지정된 도구 파일을 찾을 수 없습니다: {tool_name}.py") self.logger.info(f"지정된 {len(python_files)}개의 도구 파일을 로드합니다.") else: # 모든 .py 파일 로드 python_files = [f for f in tools_dir.glob("*.py") if f.name != "__init__.py"] self.logger.info(f"{len(python_files)}개의 도구 파일을 발견했습니다.") if not python_files: self.logger.warning("로드할 도구 파일이 없습니다.") return loaded_tools = [] failed_tools = [] for tool_file in python_files: tool_name = tool_file.stem try: # 모듈 동적 로드 module = importlib.import_module(f"{package_name}.{tool_name}") # register 함수가 있는지 확인 if hasattr(module, "register"): module.register(self) loaded_tools.append(tool_name) self.logger.info(f"[OK] {tool_name} 도구가 성공적으로 로드되었습니다.") else: self.logger.warning(f"{tool_name} 모듈에 register 함수가 없습니다.") failed_tools.append(f"{tool_name} (register 함수 없음)") except Exception as e: error_msg = f"{tool_name} 로드 실패: {str(e)}" self.logger.error(f"[ERROR] {error_msg}") failed_tools.append(f"{tool_name} (로드 실패: {str(e)})") # 로딩 결과 요약 self.logger.info(f"도구 로딩 완료: 성공 {len(loaded_tools)}개, 실패 {len(failed_tools)}개") if loaded_tools: self.logger.info(f"로드된 도구들: {', '.join(loaded_tools)}") if failed_tools: self.logger.warning(f"실패한 도구들: {', '.join(failed_tools)}") def get_available_tools(self): """현재 등록된 도구들의 목록을 반환""" tools_dir = Path(__file__).parent / "tools" available_tools = [] if not tools_dir.exists(): return available_tools python_files = [f for f in tools_dir.glob("*.py") if f.name != "__init__.py"] for tool_file in python_files: tool_name = tool_file.stem try: module = importlib.import_module(f"tools.{tool_name}") if hasattr(module, "register"): available_tools.append(tool_name) except Exception: pass # 로드 실패한 도구는 무시 return available_tools def check_tools_availability(): """사용 가능한 도구들을 확인하고 표시""" tools_dir = Path(__file__).parent / "tools" available_tools = [] if not tools_dir.exists(): print("[ERROR] tools 디렉토리가 존재하지 않습니다.") return [] # .py 파일들만 필터링 python_files = [f for f in tools_dir.glob("*.py") if f.name != "__init__.py"] if not python_files: print("[ERROR] tools 디렉토리에 Python 파일이 없습니다.") return [] print("사용 가능한 도구들:") for tool_file in python_files: tool_name = tool_file.stem available_tools.append(tool_name) print(f" [OK] {tool_name}.py") return available_tools def check_specific_tool(tool_name): """특정 도구가 사용 가능한지 확인""" tools_dir = Path(__file__).parent / "tools" tool_file = tools_dir / f"{tool_name}.py" if tool_file.exists(): print(f"[OK] {tool_name} 도구가 사용 가능합니다.") return True else: print(f"[ERROR] {tool_name} 도구를 찾을 수 없습니다.") return False async def run_mcp_server(specific_tools=None): """MCP 서버 실행""" print("MCP 서버를 시작합니다...") if specific_tools: print(f"지정된 도구들만 로드합니다: {', '.join(specific_tools)}") # 지정된 도구들이 존재하는지 확인 available_tools = check_tools_availability() missing_tools = [tool for tool in specific_tools if tool not in available_tools] if missing_tools: print(f"[ERROR] 다음 도구들을 찾을 수 없습니다: {', '.join(missing_tools)}") print("사용 가능한 도구들:") for tool in available_tools: print(f" [OK] {tool}") return print(f"[OK] 모든 지정된 도구들이 존재합니다: {', '.join(specific_tools)}") else: print("모든 사용 가능한 도구들을 로드합니다.") available_tools = check_tools_availability() app = MCPServer(specific_tools=specific_tools) await app.run_async(transport=os.getenv("TRANSPORT", "stdio")) async def check_server_tools(): """서버에서 실제로 로드된 도구들을 확인""" try: app = MCPServer() available_tools = app.get_available_tools() if available_tools: print("서버에서 로드된 도구들:") for tool in available_tools: print(f" [OK] {tool}") else: print("[ERROR] 서버에서 로드된 도구가 없습니다.") except Exception as e: print(f"[ERROR] 서버 도구 확인 중 오류 발생: {e}") async def main(): # 명령행 인수 처리 if len(sys.argv) > 1: command = sys.argv[1] if command == "check-tools": check_tools_availability() return elif command == "check-tool" and len(sys.argv) > 2: check_specific_tool(sys.argv[2]) return elif command == "check-server-tools": await check_server_tools() return elif command == "help": print("사용법:") print(" python main.py # MCP 서버 실행 (모든 도구)") print(" python main.py check-tools # 사용 가능한 도구 목록 확인") print(" python main.py check-tool notion # 특정 도구 확인") print(" python main.py check-server-tools # 서버에서 로드된 도구 확인") print(" python main.py help # 도움말 표시") return else: # 첫 번째 인수가 명령어가 아니라면 도구 이름으로 간주 specific_tools = sys.argv[1:] await run_mcp_server(specific_tools) return await run_mcp_server() if __name__ == "__main__": asyncio.run(main())