使用 Amazon Q Developer CLI 调用 MCP Server 实现 AWS Support 案例自动创建

from typing import Any
from mcp.server.fastmcp import FastMCP
import boto3
import logging
from typing import Dict, Any, List, Optional

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

region_name = 'us-east-1'

support_client = boto3.client('support', region_name=region_name)
logger.info(f"已初始化AWS Support客户端，区域: {region_name}")

# Initialize FastMCP server
mcp = FastMCP("awssupportcase")

@mcp.tool()
def create_case( 
                   service: str,
                   category: str, 
                   severity: str,
                   subject: str,
                   description: str,
                   language: str = 'zh',
                   issue_type: str = 'technical') -> Dict[str, Any]:
        """
        创建AWS技术支持案例
        
        Args:
            service: 服务名称，如'EC2', 'S3'等
            category: 问题类别
            severity: 严重程度，如'low', 'normal', 'high', 'urgent', 'critical'
            subject: 案例主题
            description: 案例详细描述
            language: 语言代码，默认为'zh'（中文）
            issue_type: 问题类型，默认为'technical'（技术问题）
            
        Returns:
            包含案例ID和其他信息的字典
        """
        try:
            response = support_client.create_case(
                subject=subject,
                serviceCode=service,
                severityCode=severity,
                categoryCode=category,
                communicationBody=description,
                language=language,
                issueType=issue_type
            )
            
            case_id = response.get('caseId')
            logger.info(f"成功创建案例，案例ID: {case_id}")
            return response
        except Exception as e:
            logger.error(f"创建案例失败: {str(e)}")
            raise

@mcp.tool()
def add_communication(case_id: str, body: str) -> Dict[str, Any]:
        """
        向案例添加通信（更新）
        
        Args:
            case_id: 案例ID
            body: 通信内容
            
        Returns:
            API响应字典
        """
        try:
            response = support_client.add_communication_to_case(
                caseId=case_id,
                communicationBody=body
            )
            
            logger.info(f"成功向案例 {case_id} 添加通信")
            return response
        except Exception as e:
            logger.error(f"添加案例通信失败: {str(e)}")
            raise
@mcp.tool()
def get_case_details(case_id: str) -> Dict[str, Any]:
        """
        获取案例详情
        
        Args:
            case_id: 案例ID
            
        Returns:
            案例详情字典
        """
        try:
            response = support_client.describe_cases(
                caseIdList=[case_id],
                includeResolvedCases=True
            )
            
            if response.get('cases'):
                logger.info(f"成功获取案例 {case_id} 详情")
                return response['cases'][0]
            else:
                logger.warning(f"未找到案例 {case_id}")
                return {}
        except Exception as e:
            logger.error(f"获取案例详情失败: {str(e)}")
            raise
@mcp.tool()
def update_case_status(case_id: str, status: str) -> Dict[str, Any]:
        """
        更新案例状态（关闭或重新打开）
        
        Args:
            case_id: 案例ID
            status: 状态，可选值为'resolved'（关闭）或'reopened'（重新打开）
            
        Returns:
            API响应字典
        """
        if status.lower() not in ['resolved', 'reopened']:
            raise ValueError("状态必须是'resolved'（关闭）或'reopened'（重新打开）")
        
        try:
            if status.lower() == 'resolved':
                response = support_client.resolve_case(
                    caseId=case_id
                )
            else:
                # 使用add_communication_to_case方法重新打开案例
                response = support_client.add_communication_to_case(
                  caseId=case_id,
                  communicationBody="重新打开案例"
            )
            
            logger.info(f"成功将案例 {case_id} 状态更新为 {status}")
            return response
        except Exception as e:
            logger.error(f"更新案例状态失败: {str(e)}")
            raise

if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport='stdio')

{
  "mcpServers": {
    "manage-aws-support-case": {
            "command": "uv",
            "args": [
                "--directory",
                "/Users/XXXXXXXX/qdemo/manage-aws-support-case",
                "run",
                "awscase.py"
            ]
      }
  }
}