"""예약 플랫폼 통합 관리 MCP 서버"""
from datetime import datetime, date, timedelta
from typing import List, Optional
from fastmcp import FastMCP
from .models import Reservation, Customer, Platform, RevenueData
from .storage import Storage
from .ical_parser import ICalParser
# MCP 서버 초기화
mcp = FastMCP("reservation-platform-mcp")
# 저장소 초기화
storage = Storage()
# ===== 예약 조회 Tools =====
@mcp.tool()
def get_today_reservations() -> str:
"""오늘 예약 현황을 조회합니다.
Returns:
오늘의 모든 예약 목록 (시간순 정렬)
"""
today = date.today()
reservations = storage.get_reservations_by_date(today)
if not reservations:
return "오늘 예약이 없습니다."
# 시간순 정렬
reservations.sort(key=lambda r: r.start_time)
result = f"📅 오늘({today.strftime('%Y-%m-%d')}) 예약 현황 ({len(reservations)}건)\n\n"
for r in reservations:
result += f"• [{r.platform}] {r.start_time.strftime('%H:%M')}-{r.end_time.strftime('%H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
return result
@mcp.tool()
def get_reservations_by_date(target_date: str) -> str:
"""특정 날짜의 예약 현황을 조회합니다.
Args:
target_date: 조회할 날짜 (YYYY-MM-DD 형식)
Returns:
해당 날짜의 모든 예약 목록
"""
try:
target = datetime.strptime(target_date, "%Y-%m-%d").date()
except ValueError:
return "날짜 형식이 올바르지 않습니다. YYYY-MM-DD 형식으로 입력해주세요."
reservations = storage.get_reservations_by_date(target)
if not reservations:
return f"{target_date}에는 예약이 없습니다."
reservations.sort(key=lambda r: r.start_time)
result = f"📅 {target_date} 예약 현황 ({len(reservations)}건)\n\n"
for r in reservations:
result += f"• [{r.platform}] {r.start_time.strftime('%H:%M')}-{r.end_time.strftime('%H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
return result
@mcp.tool()
def get_reservations_by_platform(platform: str) -> str:
"""특정 플랫폼의 모든 예약을 조회합니다.
Args:
platform: 플랫폼 이름 (airbnb, spacecloud, naver, yanolja, kakao)
Returns:
해당 플랫폼의 모든 예약 목록
"""
if platform not in ["airbnb", "spacecloud", "naver", "yanolja", "kakao"]:
return "올바른 플랫폼 이름을 입력해주세요. (airbnb, spacecloud, naver, yanolja, kakao)"
reservations = storage.get_reservations_by_platform(platform)
if not reservations:
return f"{platform}에는 예약이 없습니다."
reservations.sort(key=lambda r: r.start_time, reverse=True)
result = f"🏢 {platform.upper()} 예약 목록 ({len(reservations)}건)\n\n"
for r in reservations[:20]: # 최근 20건만
result += f"• {r.start_time.strftime('%Y-%m-%d %H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
if len(reservations) > 20:
result += f"\n... 외 {len(reservations) - 20}건"
return result
@mcp.tool()
def get_week_reservations() -> str:
"""이번 주 예약 현황을 조회합니다.
Returns:
이번 주(월~일)의 모든 예약 목록
"""
today = date.today()
# 이번 주 월요일
monday = today - timedelta(days=today.weekday())
# 이번 주 일요일
sunday = monday + timedelta(days=6)
all_reservations = storage.get_all_reservations()
week_reservations = [
r for r in all_reservations
if monday <= r.start_time.date() <= sunday
]
if not week_reservations:
return "이번 주에는 예약이 없습니다."
week_reservations.sort(key=lambda r: r.start_time)
result = f"📅 이번 주({monday} ~ {sunday}) 예약 현황 ({len(week_reservations)}건)\n\n"
current_date = None
for r in week_reservations:
r_date = r.start_time.date()
if r_date != current_date:
current_date = r_date
weekday = ['월', '화', '수', '목', '금', '토', '일'][r_date.weekday()]
result += f"\n{r_date.strftime('%m/%d')}({weekday})\n"
result += f" • [{r.platform}] {r.start_time.strftime('%H:%M')}-{r.end_time.strftime('%H:%M')} | "
result += f"{r.customer_name} | {r.price:,.0f}원\n"
return result
# ===== 중복 예약 체크 Tools =====
@mcp.tool()
def check_duplicate_reservations() -> str:
"""현재 등록된 예약 중 시간이 겹치는 중복 예약이 있는지 확인합니다.
Returns:
중복 예약 목록 (없으면 "중복 없음" 메시지)
"""
all_reservations = storage.get_all_reservations()
duplicates = []
for i, r1 in enumerate(all_reservations):
for r2 in all_reservations[i+1:]:
if r1.is_overlapping(r2):
duplicates.append((r1, r2))
if not duplicates:
return "✅ 중복 예약이 없습니다."
result = f"⚠️ 중복 예약 발견! ({len(duplicates)}건)\n\n"
for r1, r2 in duplicates:
result += f"• {r1.start_time.strftime('%Y-%m-%d %H:%M')} - {r1.end_time.strftime('%H:%M')}\n"
result += f" - [{r1.platform}] {r1.customer_name}\n"
result += f" - [{r2.platform}] {r2.customer_name}\n\n"
return result
@mcp.tool()
def check_time_slot(start_time: str, end_time: str) -> str:
"""특정 시간대에 예약 가능한지 확인합니다.
Args:
start_time: 시작 시간 (YYYY-MM-DD HH:MM 형식)
end_time: 종료 시간 (YYYY-MM-DD HH:MM 형식)
Returns:
예약 가능 여부 및 충돌하는 예약 정보
"""
try:
start = datetime.strptime(start_time, "%Y-%m-%d %H:%M")
end = datetime.strptime(end_time, "%Y-%m-%d %H:%M")
except ValueError:
return "시간 형식이 올바르지 않습니다. YYYY-MM-DD HH:MM 형식으로 입력해주세요."
# 임시 예약 객체 생성
temp_reservation = Reservation(
id="temp",
platform="airbnb",
customer_name="temp",
start_time=start,
end_time=end,
price=0
)
conflicts = storage.find_overlapping_reservations(temp_reservation)
if not conflicts:
return f"✅ {start_time} ~ {end_time} 예약 가능합니다!"
result = f"❌ {start_time} ~ {end_time} 예약 불가 (충돌 {len(conflicts)}건)\n\n"
result += "충돌하는 예약:\n"
for c in conflicts:
result += f"• [{c.platform}] {c.start_time.strftime('%H:%M')}-{c.end_time.strftime('%H:%M')} | {c.customer_name}\n"
return result
# ===== 매출/정산 Tools =====
@mcp.tool()
def get_revenue_summary(period: str = "month") -> str:
"""매출 요약 정보를 조회합니다.
Args:
period: 기간 (today, week, month)
Returns:
기간별 매출 요약 (총액, 건수, 플랫폼별)
"""
today = date.today()
if period == "today":
start_date = today
end_date = today
period_name = "오늘"
elif period == "week":
start_date = today - timedelta(days=today.weekday())
end_date = start_date + timedelta(days=6)
period_name = "이번 주"
else: # month
start_date = today.replace(day=1)
next_month = today.replace(day=28) + timedelta(days=4)
end_date = next_month.replace(day=1) - timedelta(days=1)
period_name = f"{today.year}년 {today.month}월"
all_reservations = storage.get_all_reservations()
period_reservations = [
r for r in all_reservations
if start_date <= r.start_time.date() <= end_date
]
if not period_reservations:
return f"{period_name}에는 예약이 없습니다."
# 플랫폼별 매출 계산
platform_revenue = {}
total_amount = 0
for r in period_reservations:
if r.platform not in platform_revenue:
platform_revenue[r.platform] = {"amount": 0, "count": 0}
platform_revenue[r.platform]["amount"] += r.price
platform_revenue[r.platform]["count"] += 1
total_amount += r.price
result = f"💰 {period_name} 매출 요약\n\n"
result += f"총 매출: {total_amount:,.0f}원\n"
result += f"총 예약: {len(period_reservations)}건\n"
result += f"평균 금액: {total_amount / len(period_reservations):,.0f}원\n\n"
result += "플랫폼별 매출:\n"
for platform, data in sorted(platform_revenue.items(), key=lambda x: x[1]["amount"], reverse=True):
result += f"• {platform}: {data['amount']:,.0f}원 ({data['count']}건)\n"
return result
@mcp.tool()
def calculate_platform_fees() -> str:
"""각 플랫폼의 수수료를 계산합니다.
Returns:
플랫폼별 총 매출, 수수료, 순수익
"""
all_reservations = storage.get_all_reservations()
platforms = storage.get_all_platforms()
# 플랫폼별 설정을 딕셔너리로 변환
platform_config = {p.platform: p for p in platforms}
# 플랫폼별 매출 집계
platform_stats = {}
for r in all_reservations:
if r.platform not in platform_stats:
platform_stats[r.platform] = {
"total": 0,
"count": 0,
"fee_rate": platform_config.get(r.platform, None).fee_rate if r.platform in platform_config else 0
}
platform_stats[r.platform]["total"] += r.price
platform_stats[r.platform]["count"] += 1
if not platform_stats:
return "매출 데이터가 없습니다."
result = "💳 플랫폼 수수료 계산\n\n"
total_revenue = 0
total_fees = 0
total_net = 0
for platform, stats in sorted(platform_stats.items()):
revenue = stats["total"]
fee_rate = stats["fee_rate"]
fee = revenue * fee_rate
net = revenue - fee
total_revenue += revenue
total_fees += fee
total_net += net
result += f"[{platform.upper()}]\n"
result += f" 매출: {revenue:,.0f}원 ({stats['count']}건)\n"
result += f" 수수료({fee_rate*100:.1f}%): {fee:,.0f}원\n"
result += f" 순수익: {net:,.0f}원\n\n"
result += "─" * 40 + "\n"
result += f"총 매출: {total_revenue:,.0f}원\n"
result += f"총 수수료: {total_fees:,.0f}원\n"
result += f"총 순수익: {total_net:,.0f}원\n"
return result
# ===== 고객 관리 Tools =====
@mcp.tool()
def get_blacklist() -> str:
"""블랙리스트에 등록된 고객 목록을 조회합니다.
Returns:
블랙리스트 고객 목록
"""
blacklist = storage.get_blacklist()
if not blacklist:
return "블랙리스트에 등록된 고객이 없습니다."
result = f"🚫 블랙리스트 ({len(blacklist)}명)\n\n"
for customer in blacklist:
result += f"• {customer.name} ({customer.phone})\n"
result += f" 노쇼 {customer.noshow_count}회 | 총 예약 {customer.total_reservations}회\n"
if customer.notes:
result += f" 메모: {customer.notes}\n"
result += "\n"
return result
@mcp.tool()
def add_to_blacklist(phone: str, name: str, reason: str = "") -> str:
"""고객을 블랙리스트에 추가합니다.
Args:
phone: 고객 연락처
name: 고객 이름
reason: 블랙리스트 등록 사유
Returns:
등록 완료 메시지
"""
customer = storage.get_customer(phone)
if customer:
customer.is_blacklisted = True
if reason:
customer.notes = reason
else:
customer = Customer(
phone=phone,
name=name,
is_blacklisted=True,
notes=reason
)
storage.save_customer(customer)
return f"✅ {name}({phone})님을 블랙리스트에 추가했습니다."
@mcp.tool()
def get_noshow_customers(min_count: int = 1) -> str:
"""노쇼 이력이 있는 고객 목록을 조회합니다.
Args:
min_count: 최소 노쇼 횟수 (기본값: 1)
Returns:
노쇼 고객 목록
"""
noshow_customers = storage.get_noshow_customers(min_count)
if not noshow_customers:
return f"노쇼 {min_count}회 이상인 고객이 없습니다."
result = f"⚠️ 노쇼 {min_count}회 이상 고객 ({len(noshow_customers)}명)\n\n"
for customer in sorted(noshow_customers, key=lambda c: c.noshow_count, reverse=True):
status = "🚫 블랙리스트" if customer.is_blacklisted else ""
result += f"• {customer.name} ({customer.phone}) {status}\n"
result += f" 노쇼 {customer.noshow_count}회 | 총 예약 {customer.total_reservations}회\n\n"
return result
# ===== 메시지 생성 Tools =====
@mcp.tool()
def generate_reminder_message(reservation_date: str) -> str:
"""특정 날짜 예약 고객에게 보낼 안내 메시지를 생성합니다.
Args:
reservation_date: 예약 날짜 (YYYY-MM-DD 형식)
Returns:
고객별 안내 메시지
"""
try:
target = datetime.strptime(reservation_date, "%Y-%m-%d").date()
except ValueError:
return "날짜 형식이 올바르지 않습니다. YYYY-MM-DD 형식으로 입력해주세요."
reservations = storage.get_reservations_by_date(target)
if not reservations:
return f"{reservation_date}에는 예약이 없습니다."
result = f"📨 {reservation_date} 예약 안내 메시지\n\n"
for r in sorted(reservations, key=lambda x: x.start_time):
message = f"[{r.customer_name}님 - {r.start_time.strftime('%H:%M')}]\n"
message += f"안녕하세요, {r.customer_name}님!\n"
message += f"{r.start_time.strftime('%Y년 %m월 %d일 %H:%M')} 예약 안내드립니다.\n\n"
message += f"예약 시간: {r.start_time.strftime('%H:%M')} ~ {r.end_time.strftime('%H:%M')}\n"
message += f"예약 금액: {r.price:,.0f}원\n\n"
message += "즐거운 시간 되시길 바랍니다!\n"
result += message + "\n" + "─" * 40 + "\n\n"
return result
# ===== 플랫폼 동기화 Tools =====
@mcp.tool()
def sync_platform_ical(platform: str, ical_url: str) -> str:
"""플랫폼의 iCal URL에서 예약 데이터를 동기화합니다.
Args:
platform: 플랫폼 이름 (airbnb, spacecloud, naver, yanolja, kakao)
ical_url: iCal 캘린더 URL
Returns:
동기화 결과 메시지
"""
if platform not in ["airbnb", "spacecloud", "naver", "yanolja", "kakao"]:
return "올바른 플랫폼 이름을 입력해주세요."
try:
reservations = ICalParser.sync_platform(platform, ical_url)
# 기존 예약 삭제 후 새로 추가 (간단한 구현)
for r in reservations:
storage.add_reservation(r)
# 플랫폼 설정 업데이트
config = storage.get_platform_config(platform)
if config:
config.ical_url = ical_url
storage.update_platform_config(config)
return f"✅ {platform}에서 {len(reservations)}건의 예약을 동기화했습니다."
except Exception as e:
return f"❌ 동기화 실패: {str(e)}"
if __name__ == "__main__":
import os
# 환경 변수에서 포트 읽기 (기본값: 8000)
port = int(os.environ.get("PORT", 8000))
# Transport 선택 (기본값: stdio, 환경 변수로 변경 가능)
transport = os.environ.get("TRANSPORT", "stdio")
if transport == "http":
# HTTP transport (Remote MCP 서버용)
print(f"🚀 Starting Remote MCP Server on http://0.0.0.0:{port}/mcp")
mcp.run(
transport="http",
host="0.0.0.0",
port=port,
path="/mcp"
)
elif transport == "sse":
# SSE transport (레거시 지원)
print(f"🚀 Starting SSE MCP Server on http://0.0.0.0:{port}/sse")
mcp.run(transport="sse", host="0.0.0.0", port=port)
else:
# STDIO transport (로컬 개발용, 기본값)
print("🚀 Starting Local MCP Server (STDIO)")
mcp.run()
