"""FastMCP 기반 MCP 서버
AWS EBS CloudWatch 지표 계산 MCP 서버입니다.
FastMCP를 사용하여 더 간결한 데코레이터 기반 API로 구현합니다.
"""
import json
import logging
from datetime import datetime
from typing import Optional
from mcp.server.fastmcp import FastMCP
from .advanced_calculator import AdvancedCalculator
from .cloudwatch_client import CloudWatchClient, SUPPORTED_EBS_METRICS
from .iops_calculator import IOPSCalculator
from .snapshot_calculator import SnapshotCalculator
from .throughput_calculator import ThroughputCalculator
# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# FastMCP 서버 초기화
mcp = FastMCP("ebs-cloudwatch-metrics")
def _parse_datetime(dt_str: str) -> datetime:
"""ISO 형식 문자열을 datetime으로 변환"""
return datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
def _dataclass_to_dict(obj) -> dict:
"""데이터클래스를 딕셔너리로 변환"""
if hasattr(obj, "__dict__"):
result = {}
for key, value in obj.__dict__.items():
if isinstance(value, list):
result[key] = [_dataclass_to_dict(item) for item in value]
elif hasattr(value, "__dict__"):
result[key] = _dataclass_to_dict(value)
else:
result[key] = value
return result
return obj
@mcp.tool()
async def calculate_iops(
volume_id: str,
start_time: str,
end_time: str,
period: int = 300,
region: Optional[str] = None
) -> str:
"""EBS 볼륨의 IOPS(초당 I/O 작업 수)를 계산합니다.
VolumeReadOps, VolumeWriteOps 지표를 조회하여 평균, 최대, 최소 IOPS를 반환합니다.
Args:
volume_id: EBS 볼륨 ID (예: vol-1234567890abcdef0)
start_time: 조회 시작 시간 (ISO 8601 형식, 예: 2024-01-01T00:00:00Z)
end_time: 조회 종료 시간 (ISO 8601 형식, 예: 2024-01-02T00:00:00Z)
period: 지표 수집 간격 (초 단위, 기본값: 300)
region: AWS 리전 (선택적, 기본값: 환경 변수에서 로드)
"""
calculator = IOPSCalculator(region=region)
result = await calculator.calculate_iops(
volume_id=volume_id,
start_time=_parse_datetime(start_time),
end_time=_parse_datetime(end_time),
period=period
)
return json.dumps(_dataclass_to_dict(result), indent=2, default=str)
@mcp.tool()
async def calculate_throughput(
volume_id: str,
start_time: str,
end_time: str,
period: int = 300,
region: Optional[str] = None
) -> str:
"""EBS 볼륨의 처리량(Throughput)을 계산합니다.
VolumeReadBytes, VolumeWriteBytes 지표를 조회하여 MB/s 단위로 반환합니다.
Args:
volume_id: EBS 볼륨 ID (예: vol-1234567890abcdef0)
start_time: 조회 시작 시간 (ISO 8601 형식)
end_time: 조회 종료 시간 (ISO 8601 형식)
period: 지표 수집 간격 (초 단위, 기본값: 300)
region: AWS 리전 (선택적)
"""
calculator = ThroughputCalculator(region=region)
result = await calculator.calculate_throughput(
volume_id=volume_id,
start_time=_parse_datetime(start_time),
end_time=_parse_datetime(end_time),
period=period
)
return json.dumps(_dataclass_to_dict(result), indent=2, default=str)
@mcp.tool()
async def get_snapshot_size(
snapshot_id: str,
region: Optional[str] = None
) -> str:
"""EBS 스냅샷의 크기를 조회합니다.
스냅샷 ID, 볼륨 ID, 크기(GB), 상태 등의 정보를 반환합니다.
Args:
snapshot_id: EBS 스냅샷 ID (예: snap-1234567890abcdef0)
region: AWS 리전 (선택적)
"""
calculator = SnapshotCalculator(region=region)
result = await calculator.get_snapshot_size(snapshot_id=snapshot_id)
return json.dumps(_dataclass_to_dict(result), indent=2, default=str)
@mcp.tool()
async def list_volume_snapshots(
volume_id: str,
max_results: int = 100,
region: Optional[str] = None
) -> str:
"""EBS 볼륨의 모든 스냅샷 목록을 조회합니다.
각 스냅샷의 ID, 크기, 상태, 생성 시간 등을 반환합니다.
Args:
volume_id: EBS 볼륨 ID (예: vol-1234567890abcdef0)
max_results: 최대 결과 수 (기본값: 100)
region: AWS 리전 (선택적)
"""
calculator = SnapshotCalculator(region=region)
snapshots = await calculator.list_volume_snapshots(
volume_id=volume_id,
max_results=max_results
)
total_size_gb = sum(s.size_gb for s in snapshots)
result = {
"volume_id": volume_id,
"snapshot_count": len(snapshots),
"total_size_gb": total_size_gb,
"snapshots": [_dataclass_to_dict(s) for s in snapshots]
}
return json.dumps(result, indent=2, default=str)
@mcp.tool()
async def get_ebs_metric(
volume_id: str,
metric_name: str,
start_time: str,
end_time: str,
period: int = 300,
statistics: list[str] = None,
region: Optional[str] = None
) -> str:
"""특정 EBS CloudWatch 지표를 조회합니다.
지표 이름, 볼륨 ID, 시간 범위를 지정하여 통계 데이터를 반환합니다.
Args:
volume_id: EBS 볼륨 ID
metric_name: CloudWatch 지표 이름. 지원되는 지표: VolumeReadOps, VolumeWriteOps,
VolumeReadBytes, VolumeWriteBytes, VolumeTotalReadTime, VolumeTotalWriteTime,
VolumeIdleTime, VolumeQueueLength, VolumeThroughputPercentage,
VolumeConsumedReadWriteOps, BurstBalance
start_time: 조회 시작 시간 (ISO 8601 형식)
end_time: 조회 종료 시간 (ISO 8601 형식)
period: 지표 수집 간격 (초 단위, 기본값: 300)
statistics: 조회할 통계 유형 (Average, Sum, Minimum, Maximum, SampleCount)
region: AWS 리전 (선택적)
"""
if statistics is None:
statistics = ["Average"]
client = CloudWatchClient(region=region)
result = await client.get_metric_statistics(
volume_id=volume_id,
metric_name=metric_name,
start_time=_parse_datetime(start_time),
end_time=_parse_datetime(end_time),
period=period,
statistics=statistics
)
return json.dumps(_dataclass_to_dict(result), indent=2, default=str)
@mcp.tool()
def list_ebs_metrics() -> str:
"""사용 가능한 EBS CloudWatch 지표 목록을 반환합니다."""
client = CloudWatchClient()
metrics = client.list_available_metrics()
result = {
"metrics": metrics,
"count": len(metrics),
"description": "AWS EBS CloudWatch에서 지원하는 지표 목록입니다."
}
return json.dumps(result, indent=2)
@mcp.tool()
async def get_advanced_metrics(
volume_id: str,
start_time: str,
end_time: str,
period: int = 300,
region: Optional[str] = None
) -> str:
"""EBS 볼륨의 고급 성능 지표를 계산합니다.
I/O 사용률, 지연 시간, I/O 크기, 평균/버스팅 IOPS 및 처리량 등을 반환합니다.
Args:
volume_id: EBS 볼륨 ID
start_time: 조회 시작 시간 (ISO 8601 형식)
end_time: 조회 종료 시간 (ISO 8601 형식)
period: 지표 수집 간격 (초 단위, 기본값: 300)
region: AWS 리전 (선택적)
"""
calculator = AdvancedCalculator(region=region)
result = await calculator.calculate_advanced_metrics(
volume_id=volume_id,
start_time=_parse_datetime(start_time),
end_time=_parse_datetime(end_time),
period=period
)
return json.dumps(_dataclass_to_dict(result), indent=2, default=str)
def main():
"""MCP 서버 메인 함수"""
mcp.run()
if __name__ == "__main__":
main()
