Meeting Room MCP Server

# -*- coding: utf-8 -*- """ 회의실 예약 MCP 서버 (메인) """ import os import sys import asyncio import logging from typing import Any if sys.platform == 'win32': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') logging.basicConfig( filename='mcp_debug.log', level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', encoding='utf-8' ) logger = logging.getLogger(__name__) from mcp.server import Server from mcp.types import Tool, TextContent import config from auth.login import LoginManager from api.rooms import RoomsAPI from api.reservations import ReservationsAPI from utils.date_parser import DateParser from utils.time_checker import TimeChecker # 전역 변수 login_manager = None rooms_api = None reservations_api = None app = Server("meeting-room-mcp") def initialize_services(): """서비스 초기화""" global login_manager, rooms_api, reservations_api logger.info("서비스 초기화 시작...") try: login_manager = LoginManager(config) rooms_api = RoomsAPI(config.BASE_URL, login_manager) reservations_api = ReservationsAPI(config.BASE_URL, login_manager, config) logger.info("서비스 초기화 완료") except Exception as e: logger.error(f"서비스 초기화 실패: {str(e)}", exc_info=True) raise def ensure_login(): """로그인 상태 확인""" global login_manager if not login_manager: logger.error("login_manager가 None입니다. initialize_services()가 실행되지 않았습니다.") raise Exception("서비스가 초기화되지 않았습니다.") if not login_manager.session_cookies: logger.info("세션 없음. 로그인 시도...") login_manager.login() @app.list_tools() async def list_tools() -> list[Tool]: """사용 가능한 도구 목록""" return [ Tool( name="search_available_rooms", description="특정 날짜와 시간대에 예약 가능한 회의실 검색", inputSchema={ "type": "object", "properties": { "date": {"type": "string", "description": "날짜 (예: '10.24', '2025-10-24')"}, "office": {"type": "string", "description": "사무실 (분당, 방배, 가양, 퓨처)", "default": "분당"}, "floor": {"type": "string", "description": "층 (예: '7층', '12F_A')", "default": ""}, "start_hour": {"type": "integer", "description": "시작 시간", "default": 9}, "duration": {"type": "integer", "description": "필요 시간(시간)", "default": 1} }, "required": ["date"] } ), Tool( name="get_room_details", description="회의실 상세 정보 조회", inputSchema={ "type": "object", "properties": { "room_name": {"type": "string", "description": "회의실 이름"}, "date": {"type": "string", "description": "날짜"} }, "required": ["room_name", "date"] } ), Tool( name="create_reservation", description="회의실 예약 생성", inputSchema={ "type": "object", "properties": { "room_name": {"type": "string", "description": "회의실 이름"}, "date": {"type": "string", "description": "날짜"}, "start_time": {"type": "string", "description": "시작 시간"}, "end_time": {"type": "string", "description": "종료 시간"}, "title": {"type": "string", "description": "회의 제목", "default": "회의"} }, "required": ["room_name", "date", "start_time", "end_time"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: """도구 실행""" logger.info(f"Tool called: {name} with args: {arguments}") try: ensure_login() if name == "search_available_rooms": return await search_available_rooms(arguments) elif name == "get_room_details": return await get_room_details(arguments) elif name == "create_reservation": return await create_reservation(arguments) else: return [TextContent(type="text", text=f"Unknown tool: {name}")] except Exception as e: logger.error(f"오류 발생: {str(e)}", exc_info=True) return [TextContent(type="text", text=f"오류 발생: {str(e)}")] async def search_available_rooms(args: dict) -> list[TextContent]: """예약 가능한 회의실 검색""" date = DateParser.parse_date(args.get("date", "")) office = args.get("office", "분당") floor_filter = args.get("floor", "") start_hour = args.get("start_hour", 9) duration = args.get("duration", 1) office_id = config.OFFICE_MAP.get(office, 2) try: all_rooms = rooms_api.get_all_rooms(office_id) if floor_filter: all_rooms = rooms_api.filter_by_floor(all_rooms, floor_filter) reservations = reservations_api.get_reservations(date, office_id) result = f"[{date}] {office} 회의실 현황\n" result += f"검색 조건: {start_hour}시 이후 {duration}시간 사용 가능\n\n" available_count = 0 for room in all_rooms: slots, room_reservations = TimeChecker.check_available_slots( room['id'], reservations, start_hour, duration ) if slots: available_count += 1 result += f"✅ {room['name']} ({room['floorName']}, {room['personnel']}명)\n" result += f" 가능 시간:\n" for slot in slots: result += f" • {slot}\n" if room_reservations: result += f" 기존 예약:\n" for res in room_reservations: h_start = res['start'] // 60 m_start = res['start'] % 60 h_end = res['end'] // 60 m_end = res['end'] % 60 result += f" • {h_start:02d}:{m_start:02d}~{h_end:02d}:{m_end:02d} ({res['title']})\n" result += "\n" if available_count == 0: result += "❌ 조건에 맞는 회의실이 없습니다.\n" else: result += f"\n총 {available_count}개 회의실 예약 가능!" return [TextContent(type="text", text=result)] except Exception as e: return [TextContent(type="text", text=f"검색 실패: {str(e)}")] async def get_room_details(args: dict) -> list[TextContent]: """회의실 상세 정보""" room_name = args.get("room_name", "") date = DateParser.parse_date(args.get("date", "")) try: room = rooms_api.get_room_by_name(room_name) if not room: return [TextContent(type="text", text=f"'{room_name}' 회의실을 찾을 수 없습니다.")] office_id = room.get('officeId', 2) reservations = reservations_api.get_reservations(date, office_id) room_reservations = [] for res in reservations: try: targets = res.get('reservation', {}).get('reservationTargets', []) if targets and targets[0].get('room', {}).get('id') == room['id']: room_reservations.append(res) except: continue result = f"[회의실 정보] {room['name']}\n\n" result += f"위치: {room.get('officeName', 'N/A')} {room.get('floorName', 'N/A')}\n" result += f"수용 인원: {room.get('personnel', 'N/A')}명\n\n" if room_reservations: result += f"[{date} 예약 현황]\n" for res in room_reservations: title = res.get('reservation', {}).get('title', '제목없음') start = res.get('startDateTime', 'N/A')[11:16] end = res.get('endDateTime', 'N/A')[11:16] result += f" • {start}~{end}: {title}\n" else: result += f"[{date}] 예약 없음\n" return [TextContent(type="text", text=result)] except Exception as e: return [TextContent(type="text", text=f"조회 실패: {str(e)}")] async def create_reservation(args: dict) -> list[TextContent]: """회의실 예약""" room_name = args.get("room_name", "") date = DateParser.parse_date(args.get("date", "")) start_time = DateParser.parse_time(args.get("start_time", "")) end_time = DateParser.parse_time(args.get("end_time", "")) title = args.get("title", "회의") try: room = rooms_api.get_room_by_name(room_name) if not room: return [TextContent(type="text", text=f"'{room_name}' 회의실을 찾을 수 없습니다.")] response = reservations_api.create_reservation(room, date, start_time, end_time, title) if response.status_code == 201: result_data = response.json() result = f"[예약 완료]\n\n회의실: {room['name']}\n날짜: {date}\n시간: {start_time} ~ {end_time}\n제목: {title}\n예약 ID: {result_data.get('id')}" return [TextContent(type="text", text=result)] else: return [TextContent(type="text", text=f"예약 실패: {response.status_code}")] except Exception as e: return [TextContent(type="text", text=f"예약 실패: {str(e)}")] async def main(): """MCP 서버 실행""" from mcp.server.stdio import stdio_server logger.info("MCP 서버 시작") # 서비스 초기화 try: initialize_services() except Exception as e: logger.error(f"초기화 실패로 서버 종료: {str(e)}") return async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() ) if __name__ == "__main__": import warnings warnings.filterwarnings('ignore', message='Unverified HTTPS request') asyncio.run(main())