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

from typing import Anyfrom mcp.server.fastmcp import FastMCPimport boto3import loggingfrom 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 servermcp = 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"            ]      }  }}