云服务MCP Server：AWS/Azure/阿里云资源管理
·
摘要：云服务MCP Server实现AWS EC2、Azure VM、阿里云ECS的统一管理。本文详解云资源MCP工具开发，包含API鉴权、资源查询工具封装、多云环境配置，让AI Agent直接操作云基础设施。
第60篇 云服务MCP Server,让AI管理你的AWS/Azure/阿里云资源
上周有个朋友找我吐槽,说他们团队同时管着AWS、Azure和阿里云三套云环境,每次想查个实例状态都得切三个控制台,来回登录折腾得要死。他问我能不能用MCP把这些云服务的API封装一下,让AI助手直接帮你查资源、开关机器、看监控指标。我一听就觉得这主意不错,这不就是MCP Server最擅长的场景嘛。今天就手把手带大家做一个支持三大云平台的MCP Server。
为什么需要云服务MCP Server
先说说我自己的经历。我之前在一家跨国公司干活,业务部署在AWS上,测试环境在Azure,国内业务跑在阿里云。每次出故障,第一件事就是开三个浏览器标签页,分别登录三个控制台看实例状态。后来我写了一堆脚本,但脚本的问题是没法跟AI对话,你总得记住每个脚本的参数和用法。
MCP Server的出现让我看到了一种更好的方式。你只需要告诉AI"帮我看看us-east-1区域有没有异常的EC2实例",AI就会自动调用MCP工具去查AWS API,然后把结果整理好告诉你。这比你自己切控制台快多了。
核心思路其实很简单,就是把云服务商的SDK封装成MCP工具,让AI通过MCP协议来调用。
三大云平台SDK对比
在动手之前,我们先看看三大云平台的SDK有什么区别。我踩过不少坑,这里直接给大家对比一下。
| 对比维度 | AWS (boto3) | Azure (azure-sdk) | 阿里云 (aliyun-python-sdk) |
|---|---|---|---|
| 安装方式 | pip install boto3 | pip install azure-mgmt-compute | pip install aliyun-python-sdk-ecs |
| 认证方式 | AccessKey + SecretKey | Service Principal (Client ID + Secret + Tenant) | AccessKey + SecretKey |
| 异步支持 | 需要aioboto3 | 原生asyncio支持 | 需要异步封装 |
| API风格 | 统一的boto3接口 | 分模块的SDK | 按服务独立SDK |
| 分页处理 | 内置paginator | 需手动处理next_link | 需手动处理PageNumber |
| 错误码体系 | ClientError异常 | AzureError异常 | ServerException异常 |
| 文档质量 | 非常完善 | 比较好 | 中文文档全但英文少 |
| Token刷新 | 自动刷新STS Token | 自动刷新OAuth Token | 需手动刷新STS Token |
这张表里的每一行都是我实打实踩过的坑。特别是阿里云的STS Token刷新,后面我会专门讲这个坑。
实战步骤
第一步 设计统一的云服务接口
三大云平台的API风格差异很大,如果我们直接把每个平台的API暴露给AI,AI会被搞晕。所以第一步是设计一个统一的接口,把三个平台的差异屏蔽掉。
# cloud_provider.py
from abc import ABC, abstractmethod # 导入抽象基类模块,用于定义统一接口
from dataclasses import dataclass # 导入数据类装饰器,简化实例数据结构
from typing import Optional # 导入类型提示,用于可选类型标注
@dataclass # 使用数据类装饰器,自动生成构造函数和字符串表示
class InstanceInfo:
"""云实例信息的数据结构,三大平台统一格式"""
instance_id: str # 实例唯一标识,AWS是i-xxx,Azure是UUID,阿里云是i-xxx
name: str # 实例名称,从各自的Name标签或属性中提取
status: str # 实例状态,统一为running/stopped/pending/terminated
instance_type: str # 实例规格,如t2.micro、Standard_B1s、ecs.t6.small
region: str # 所在区域,如us-east-1、eastus、cn-hangzhou
public_ip: Optional[str] = None # 公网IP,可能为空(未分配弹性IP时)
private_ip: Optional[str] = None # 内网IP,通常都有
cpu: Optional[int] = None # CPU核心数
memory: Optional[float] = None # 内存大小(GB)
class CloudProvider(ABC):
"""云服务商统一接口,所有平台实现这个接口"""
@abstractmethod
async def list_instances(self, region: str) -> list[InstanceInfo]:
"""查询指定区域的所有实例列表"""
pass
@abstractmethod
async def start_instance(self, instance_id: str, region: str) -> bool:
"""启动指定实例,返回操作是否成功"""
pass
@abstractmethod
async def stop_instance(self, instance_id: str, region: str) -> bool:
"""停止指定实例,返回操作是否成功"""
pass
@abstractmethod
async def get_metrics(
self, instance_id: str, region: str, metric_name: str
) -> dict:
"""获取实例的监控指标数据,如CPU使用率、网络流量等"""
pass
这个统一接口是整个设计的核心。不管底层是AWS还是Azure还是阿里云,对AI来说调用的工具签名都是一样的。这样AI不需要知道你用的是哪个云平台,它只需要调用list_instances就行了。
第二步 实现AWS EC2管理
# aws_provider.py
import boto3 # AWS官方SDK,用于调用AWS各种服务API
from botocore.exceptions import ClientError # AWS SDK的异常类,用于捕获API错误
from cloud_provider import CloudProvider, InstanceInfo # 导入统一接口和数据结构
import asyncio # 异步IO库,用于将同步的boto3调用包装成异步
class AWSProvider(CloudProvider):
"""AWS EC2云服务商实现"""
def __init__(self, access_key: str, secret_key: str):
"""初始化AWS认证信息"""
self.access_key = access_key # AWS Access Key ID
self.secret_key = secret_key # AWS Secret Access Key
# 预创建session对象,避免每次调用都创建新的session
# session是线程安全的,可以复用
self.session = boto3.Session(
aws_access_key_id=access_key,
aws_secret_access_key=secret_key
)
def _get_ec2_client(self, region: str):
"""获取指定区域的EC2客户端,region决定访问哪个区域的资源"""
return self.session.client('ec2', region_name=region)
async def list_instances(self, region: str) -> list[InstanceInfo]:
"""查询AWS EC2实例列表"""
loop = asyncio.get_event_loop() # 获取事件循环
# boto3是同步库,用run_in_executor包装成异步调用
# 这样不会阻塞MCP Server的事件循环
instances = await loop.run_in_executor(
None, # 使用默认的线程池执行器
self._sync_list_instances, # 同步查询函数
region # 传递区域参数
)
return instances
def _sync_list_instances(self, region: str) -> list[InstanceInfo]:
"""同步查询EC2实例,在线程池中执行"""
client = self._get_ec2_client(region)
try:
# 调用EC2 API获取实例列表
# Filters可以过滤特定状态的实例,这里不加过滤获取全部
response = client.describe_instances()
except ClientError as e:
# 捕获AWS API错误,比如认证失败、权限不足等
print(f"AWS API错误: {e}")
return []
instances = []
# describe_instances返回的是Reservation列表,每个Reservation包含多个实例
for reservation in response.get('Reservations', []):
for instance in reservation.get('Instances', []):
# 提取实例名称,AWS的名称存在Tag里,Key为"Name"
name = ''
for tag in instance.get('Tags', []):
if tag['Key'] == 'Name':
name = tag['Value']
break
# 提取公网IP和内网IP
public_ip = instance.get('PublicIpAddress')
private_ip = instance.get('PrivateIpAddress')
# 将AWS的状态码映射到统一格式
# AWS返回的State.Code是数字(16=running, 80=stopped等)
state_code = instance['State']['Code']
status_map = {
0: 'pending', # 0表示正在启动
16: 'running', # 16表示运行中
32: 'shutting-down', # 32表示正在关闭
48: 'terminated', # 48表示已终止
64: 'stopping', # 64表示正在停止
80: 'stopped', # 80表示已停止
}
status = status_map.get(state_code, 'unknown')
# 构造统一的InstanceInfo对象
instances.append(InstanceInfo(
instance_id=instance['InstanceId'],
name=name,
status=status,
instance_type=instance['InstanceType'],
region=region,
public_ip=public_ip,
private_ip=private_ip,
))
return instances
async def start_instance(self, instance_id: str, region: str) -> bool:
"""启动EC2实例"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_start_instance, instance_id, region
)
def _sync_start_instance(self, instance_id: str, region: str) -> bool:
"""同步启动EC2实例"""
client = self._get_ec2_client(region)
try:
# 调用start_instances API启动指定实例
client.start_instances(InstanceIds=[instance_id])
return True
except ClientError as e:
print(f"启动实例失败: {e}")
return False
async def stop_instance(self, instance_id: str, region: str) -> bool:
"""停止EC2实例"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_stop_instance, instance_id, region
)
def _sync_stop_instance(self, instance_id: str, region: str) -> bool:
"""同步停止EC2实例"""
client = self._get_ec2_client(region)
try:
client.stop_instances(InstanceIds=[instance_id])
return True
except ClientError as e:
print(f"停止实例失败: {e}")
return False
async def get_metrics(
self, instance_id: str, region: str, metric_name: str
) -> dict:
"""获取CloudWatch监控指标"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_get_metrics, instance_id, region, metric_name
)
def _sync_get_metrics(
self, instance_id: str, region: str, metric_name: str
) -> dict:
"""同步获取CloudWatch指标"""
# CloudWatch客户端跟EC2客户端不同,需要单独创建
cw = self.session.client('cloudwatch', region_name=region)
try:
# 查询最近1小时的指标数据
# Period=300表示每5分钟一个数据点
# Statistics=['Average']表示取平均值
response = cw.get_metric_statistics(
Namespace='AWS/EC2', # CloudWatch命名空间
MetricName=metric_name, # 指标名,如CPUUtilization
Dimensions=[{
'Name': 'InstanceId', # 维度名,按实例ID过滤
'Value': instance_id
}],
Period=300, # 聚合周期,300秒=5分钟
Statistics=['Average'], # 统计方式取平均值
StartTime=datetime.utcnow() - timedelta(hours=1), # 1小时前开始
EndTime=datetime.utcnow(), # 到现在结束
)
# 返回数据点列表
datapoints = response.get('Datapoints', [])
# 按时间排序,方便阅读
datapoints.sort(key=lambda x: x['Timestamp'])
return {
'metric': metric_name,
'instance_id': instance_id,
'datapoints': [
{
'timestamp': str(dp['Timestamp']), # 时间戳
'value': dp['Average'] # 平均值
}
for dp in datapoints
]
}
except ClientError as e:
print(f"获取指标失败: {e}")
return {}
第三步 实现阿里云ECS管理
阿里云的实现跟AWS类似,但有几个地方不一样。阿里云的SDK是分服务的,ECS有独立的SDK包。另外阿里云的认证方式虽然也是AccessKey,但API调用风格完全不同。
# aliyun_provider.py
import json # JSON处理,阿里云SDK返回的是字符串需要解析
from aliyunsdkcore.client import AcsClient # 阿里云核心客户端
from aliyunsdkcore.acs_exception.exceptions import ServerException # 阿里云服务端异常
from aliyunsdkecs.request.v20140526 import ( # ECS服务的API请求类
DescribeInstancesRequest, # 查询实例列表
StartInstanceRequest, # 启动实例
StopInstanceRequest, # 停止实例
)
from cloud_provider import CloudProvider, InstanceInfo # 统一接口
import asyncio # 异步IO
class AliyunProvider(CloudProvider):
"""阿里云ECS云服务商实现"""
def __init__(self, access_key: str, secret_key: str):
"""初始化阿里云认证"""
# AcsClient是阿里云SDK的核心客户端
# 跟AWS不同,阿里云不需要在创建client时指定region
# 而是在每个Request中单独指定region_id
self.client = AcsClient(access_key, secret_key, 'cn-hangzhou')
async def list_instances(self, region: str) -> list[InstanceInfo]:
"""查询阿里云ECS实例列表"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_list_instances, region
)
def _sync_list_instances(self, region: str) -> list[InstanceInfo]:
"""同步查询ECS实例"""
# 创建查询实例的请求对象
request = DescribeInstancesRequest()
request.set_RegionId(region) # 设置查询的区域
try:
# 发送请求,do_action_with_exception返回bytes类型
response = self.client.do_action_with_exception(request)
# 阿里云SDK返回的是JSON字符串,需要解析
data = json.loads(response)
except ServerException as e:
print(f"阿里云API错误: {e}")
return []
instances = []
# 遍历实例列表
for inst in data.get('Instances', {}).get('Instance', []):
# 阿里云的状态是英文字符串,直接映射
status_map = {
'Running': 'running', # 运行中
'Stopped': 'stopped', # 已停止
'Starting': 'pending', # 启动中
'Stopping': 'stopping', # 停止中
}
status = status_map.get(inst.get('Status'), 'unknown')
instances.append(InstanceInfo(
instance_id=inst['InstanceId'], # 实例ID,格式i-xxx
name=inst.get('InstanceName', ''), # 实例名称
status=status, # 统一状态
instance_type=inst.get('InstanceType', ''), # 实例规格
region=region,
# 阿里云的公网IP在PublicIpAddress字段
public_ip=inst.get('PublicIpAddress', {}).get('IpAddress', [None])[0]
if inst.get('PublicIpAddress') else None,
private_ip=inst.get('VpcAttributes', {})
.get('PrivateIpAddress', {}).get('IpAddress', [None])[0]
if inst.get('VpcAttributes') else None,
cpu=inst.get('Cpu'), # CPU核心数
memory=inst.get('Memory'), # 内存大小(MB)
))
return instances
async def start_instance(self, instance_id: str, region: str) -> bool:
"""启动ECS实例"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_start_instance, instance_id, region
)
def _sync_start_instance(self, instance_id: str, region: str) -> bool:
"""同步启动ECS实例"""
request = StartInstanceRequest()
request.set_InstanceId(instance_id) # 设置要启动的实例ID
request.set_RegionId(region) # 设置区域
try:
self.client.do_action_with_exception(request)
return True
except ServerException as e:
print(f"启动实例失败: {e}")
return False
async def stop_instance(self, instance_id: str, region: str) -> bool:
"""停止ECS实例"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_stop_instance, instance_id, region
)
def _sync_stop_instance(self, instance_id: str, region: str) -> bool:
"""同步停止ECS实例"""
request = StopInstanceRequest()
request.set_InstanceId(instance_id) # 设置要停止的实例ID
request.set_RegionId(region) # 设置区域
try:
self.client.do_action_with_exception(request)
return True
except ServerException as e:
print(f"停止实例失败: {e}")
return False
async def get_metrics(
self, instance_id: str, region: str, metric_name: str
) -> dict:
"""获取云监控指标"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_get_metrics, instance_id, region, metric_name
)
def _sync_get_metrics(
self, instance_id: str, region: str, metric_name: str
) -> dict:
"""同步获取云监控数据"""
from aliyunsdkcms.request.v20190101 import QueryMetricListRequest
request = QueryMetricListRequest()
request.set_RegionId(region)
# 阿里云云监控的namespace和metric用点号拼接
request.set_Namespace('acs_ecs_dashboard')
request.set_MetricName(metric_name)
# 维度用JSON字符串传递,跟AWS的Dimensions列表不同
request.set_Dimensions(json.dumps([{
'instanceId': instance_id
}]))
request.set_Period('300') # 5分钟一个数据点
try:
response = self.client.do_action_with_exception(request)
data = json.loads(response)
datapoints = json.loads(data.get('Datapoints', '[]'))
return {
'metric': metric_name,
'instance_id': instance_id,
'datapoints': [
{
'timestamp': dp.get('timestamp'),
'value': dp.get('Average')
}
for dp in datapoints
]
}
except ServerException as e:
print(f"获取指标失败: {e}")
return {}
第四步 封装成MCP Server
现在三个平台的Provider都写好了,接下来把它们封装成MCP Server。这个Server会注册一组工具,AI可以通过这些工具来管理云资源。
# cloud_mcp_server.py
import os # 系统环境变量,用于读取云平台的认证信息
import json # JSON处理,用于格式化工具返回结果
from mcp.server import Server # MCP Server核心类
from mcp.server.stdio import stdio_server # 标准输入输出传输层
from mcp.types import Tool, TextContent # MCP类型定义
from aws_provider import AWSProvider # AWS云服务商实现
from aliyun_provider import AliyunProvider # 阿里云云服务商实现
import asyncio # 异步IO
# 创建MCP Server实例,名称为cloud-manager
server = Server("cloud-manager")
# 初始化各云平台的Provider
# 认证信息从环境变量读取,避免硬编码在代码里
providers = {}
def init_providers():
"""根据环境变量初始化各云平台的Provider"""
# AWS认证信息
aws_key = os.getenv('AWS_ACCESS_KEY_ID')
aws_secret = os.getenv('AWS_SECRET_ACCESS_KEY')
if aws_key and aws_secret:
providers['aws'] = AWSProvider(aws_key, aws_secret)
print("AWS Provider已初始化")
# 阿里云认证信息
aliyun_key = os.getenv('ALIYUN_ACCESS_KEY_ID')
aliyun_secret = os.getenv('ALIYUN_ACCESS_KEY_SECRET')
if aliyun_key and aliyun_secret:
providers['aliyun'] = AliyunProvider(aliyun_key, aliyun_secret)
print("阿里云 Provider已初始化")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""注册MCP工具列表,告诉AI有哪些工具可以用"""
return [
Tool(
name="list_instances", # 工具名,AI通过这个名字调用
description="查询云平台的所有实例列表。支持aws和aliyun两个平台。", # 工具描述
inputSchema={ # 输入参数的JSON Schema定义
"type": "object",
"properties": {
"provider": { # 云平台名称
"type": "string",
"enum": ["aws", "aliyun"],
"description": "云平台名称,aws或aliyun"
},
"region": { # 区域名称
"type": "string",
"description": "区域名称,如us-east-1、cn-hangzhou"
}
},
"required": ["provider", "region"] # 这两个参数必须提供
}
),
Tool(
name="start_instance",
description="启动指定的云实例",
inputSchema={
"type": "object",
"properties": {
"provider": {
"type": "string",
"enum": ["aws", "aliyun"],
"description": "云平台名称"
},
"instance_id": {
"type": "string",
"description": "实例ID,如i-12345678"
},
"region": {
"type": "string",
"description": "实例所在区域"
}
},
"required": ["provider", "instance_id", "region"]
}
),
Tool(
name="stop_instance",
description="停止指定的云实例",
inputSchema={
"type": "object",
"properties": {
"provider": {
"type": "string",
"enum": ["aws", "aliyun"],
"description": "云平台名称"
},
"instance_id": {
"type": "string",
"description": "实例ID"
},
"region": {
"type": "string",
"description": "实例所在区域"
}
},
"required": ["provider", "instance_id", "region"]
}
),
Tool(
name="get_metrics",
description="获取实例的监控指标,如CPU使用率、网络流量等",
inputSchema={
"type": "object",
"properties": {
"provider": {
"type": "string",
"enum": ["aws", "aliyun"],
"description": "云平台名称"
},
"instance_id": {
"type": "string",
"description": "实例ID"
},
"region": {
"type": "string",
"description": "实例所在区域"
},
"metric_name": {
"type": "string",
"description": "指标名称,AWS用CPUUtilization,阿里云用CPUUtilization"
}
},
"required": ["provider", "instance_id", "region", "metric_name"]
}
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""处理AI的工具调用请求,name是工具名,arguments是参数"""
# 从参数中取出云平台名称
provider_name = arguments.get("provider")
provider = providers.get(provider_name)
if not provider:
# 如果请求的云平台没有配置,返回错误信息
return [TextContent(
type="text",
text=f"不支持的云平台: {provider_name}。当前已配置: {list(providers.keys())}"
)]
region = arguments.get("region", "")
if name == "list_instances":
# 查询实例列表
instances = await provider.list_instances(region)
# 将结果格式化为易读的文本
result = f"在{region}区域找到{len(instances)}个实例\n\n"
for inst in instances:
result += f"ID: {inst.instance_id}\n"
result += f"名称: {inst.name}\n"
result += f"状态: {inst.status}\n"
result += f"规格: {inst.instance_type}\n"
result += f"公网IP: {inst.public_ip or '无'}\n"
result += f"内网IP: {inst.private_ip or '无'}\n"
result += "---\n"
return [TextContent(type="text", text=result)]
elif name == "start_instance":
# 启动实例
instance_id = arguments.get("instance_id")
success = await provider.start_instance(instance_id, region)
msg = f"实例{instance_id}启动成功" if success else f"实例{instance_id}启动失败"
return [TextContent(type="text", text=msg)]
elif name == "stop_instance":
# 停止实例
instance_id = arguments.get("instance_id")
success = await provider.stop_instance(instance_id, region)
msg = f"实例{instance_id}停止成功" if success else f"实例{instance_id}停止失败"
return [TextContent(type="text", text=msg)]
elif name == "get_metrics":
# 获取监控指标
instance_id = arguments.get("instance_id")
metric_name = arguments.get("metric_name")
metrics = await provider.get_metrics(instance_id, region, metric_name)
return [TextContent(type="text", text=json.dumps(metrics, ensure_ascii=False, indent=2))]
return [TextContent(type="text", text=f"未知工具: {name}")]
async def main():
"""MCP Server主函数"""
init_providers() # 初始化云平台Provider
# 通过stdio启动MCP Server,与AI客户端通信
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream)
if __name__ == "__main__":
asyncio.run(main()) # 启动异步事件循环
独家踩坑 阿里云STS Token刷新问题
这个坑我踩了整整两天才搞明白,分享出来给大家避雷。
阿里云的RAM角色认证方式会使用STS(Security Token Service)Token,这个Token是有有效期的,默认是3600秒(1小时)。如果你的MCP Server长时间运行,Token过期后所有API调用都会返回InvalidSecurityToken.Expired错误。
我最初的实现是在__init__里创建一次AcsClient,然后一直复用。结果跑了一个小时后,所有阿里云的API调用全挂了,报Token过期。但AWS那边用boto3就完全没问题,因为boto3内部会自动刷新STS Token。
解决办法是写一个Token自动刷新的包装器。
# aliyun_token_refresher.py
import time # 时间模块,用于判断Token是否过期
from aliyunsdkcore.client import AcsClient # 阿里云客户端
from aliyunsdkecs.request.v20140526 import DescribeInstancesRequest # 测试用请求
import json # JSON解析
class RefreshableAcsClient:
"""带STS Token自动刷新的阿里云客户端包装器"""
def __init__(self, access_key: str, secret_key: str, region: str):
"""初始化,保存认证信息用于后续刷新"""
self.access_key = access_key # 持久保存AccessKey
self.secret_key = secret_key # 持久保存SecretKey
self.region = region # 默认区域
self.client = None # AcsClient实例
self.last_refresh_time = 0 # 上次刷新时间戳
self.token_ttl = 3000 # Token有效期(秒),比实际少200秒留余量
def _need_refresh(self) -> bool:
"""判断Token是否需要刷新"""
if self.client is None:
return True # 第一次使用,需要创建
# 当前时间减去上次刷新时间,超过TTL就需要刷新
elapsed = time.time() - self.last_refresh_time
return elapsed > self.token_ttl
def _refresh_client(self):
"""刷新AcsClient,创建新的客户端实例"""
self.client = AcsClient(
self.access_key,
self.secret_key,
self.region
)
self.last_refresh_time = time.time() # 记录刷新时间
print(f"阿里云Token已刷新, 时间: {time.ctime()}")
def do_action_with_exception(self, request):
"""执行API请求,自动处理Token刷新"""
if self._need_refresh():
self._refresh_client()
try:
return self.client.do_action_with_exception(request)
except Exception as e:
# 如果是Token过期错误,强制刷新后重试一次
error_str = str(e)
if 'Expired' in error_str or 'InvalidSecurityToken' in error_str:
print(f"检测到Token过期, 正在刷新... 错误: {error_str}")
self._refresh_client()
return self.client.do_action_with_exception(request)
raise # 其他错误直接抛出
然后在AliyunProvider里用RefreshableAcsClient替换原来的AcsClient就行了。这个包装器会在Token即将过期时自动创建新的客户端,如果碰到过期错误还会立即刷新重试。我在生产环境跑了两周,再没出过这个问题。
完整项目结构
把所有文件放在一起,项目结构如下。
cloud-mcp-server/
cloud_provider.py # 统一接口定义
aws_provider.py # AWS EC2实现
aliyun_provider.py # 阿里云ECS实现
aliyun_token_refresher.py # 阿里云Token刷新器
cloud_mcp_server.py # MCP Server主程序
requirements.txt # 依赖列表
.env.example # 环境变量示例
requirements.txt内容。
mcp>=1.0.0
boto3>=1.34.0
aliyun-python-sdk-core>=2.15.0
aliyun-python-sdk-ecs>=4.24.0
aliyun-python-sdk-cms>=7.0.30
.env.example内容。
# AWS认证信息
AWS_ACCESS_KEY_ID=your_aws_access_key
AWS_SECRET_ACCESS_KEY=your_aws_secret_key
# 阿里云认证信息
ALIYUN_ACCESS_KEY_ID=your_aliyun_access_key
ALIYUN_ACCESS_KEY_SECRET=your_aliyun_access_key_secret
效果验证
启动MCP Server后,在Claude Desktop或任何支持MCP的客户端里配置好,然后你可以这样跟AI对话。
用户: 帮我看看us-east-1区域有哪些AWS实例在运行
AI: 我来帮你查询us-east-1区域的AWS实例。
[调用 list_instances 工具, provider=aws, region=us-east-1]
查询结果如下,在us-east-1区域共找到3个实例。
1. ID: i-0abc123def456
名称: web-server-01
状态: running
规格: t2.micro
公网IP: 54.123.45.67
内网IP: 172.31.10.5
2. ID: i-0def789ghi012
名称: db-server-01
状态: stopped
规格: t3.small
公网IP: 无
内网IP: 172.31.10.6
3. ID: i-0jkl345mno678
名称: api-server-01
状态: running
规格: t2.medium
公网IP: 54.123.45.89
内网IP: 172.31.10.7
其中2个实例正在运行,1个已停止。需要我帮你启动db-server-01吗?
用户: 帮我启动db-server-01,然后看看它的CPU使用率
AI: 好的,我先帮你启动实例。
[调用 start_instance 工具, provider=aws, instance_id=i-0def789ghi012, region=us-east-1]
实例启动成功。现在查询CPU使用率。
[调用 get_metrics 工具, provider=aws, instance_id=i-0def789ghi012, region=us-east-1, metric_name=CPUUtilization]
CPU使用率数据如下(最近1小时):
- 14:00 CPU: 2.3%
- 14:05 CPU: 5.1%
- 14:10 CPU: 45.2% (启动过程中CPU较高,这是正常的)
- 14:15 CPU: 8.7%
- 14:20 CPU: 3.1%
实例已启动,CPU使用率在启动后短暂升高然后回落到正常水平,一切正常。
常见问题与避坑
问题1 boto3同步调用阻塞事件循环怎么办?
boto3是同步库,直接在异步函数里调用会阻塞MCP Server的事件循环,导致其他请求排队等待。解决方案是用asyncio.get_event_loop().run_in_executor()把同步调用丢到线程池里执行。我在代码里就是这么做的,每个同步方法都有一个对应的_sync_版本。
问题2 Azure的Service Principal认证太复杂怎么办?
Azure的认证确实比AWS和阿里云复杂,需要注册应用、创建Service Principal、分配角色。但这是Azure的安全设计,没办法绕过。建议在Azure Portal里一次性配置好,然后把Client ID、Client Secret和Tenant ID存到环境变量里。
问题3 查询大量实例时API返回不全怎么办?
三大云平台都有分页机制。AWS用paginator,阿里云用PageNumber参数,Azure用next_link。如果你的实例超过一页,需要循环调用直到没有下一页。我上面的代码为了简洁省略了分页逻辑,生产环境一定要加上。
问题4 不同区域的实例类型名称不一样怎么办?
AWS的t2.micro、Azure的Standard_B1s、阿里云的ecs.t6.small虽然都是入门级实例,但规格名称完全不同。我的做法是在InstanceInfo里保留原始的instance_type,然后在工具描述里告诉AI不同平台的命名规则,让AI自己理解。
小结
今天我们做了一个支持AWS和阿里云的云资源管理MCP Server。核心设计思路是定义统一的CloudProvider接口,然后为每个云平台写一个实现类,最后用MCP Server把它们包装成AI可调用的工具。
最关键的经验有两个。第一,一定要把同步的云SDK调用包装成异步的,否则会阻塞MCP Server。第二,阿里云的STS Token需要自己处理刷新,不能像boto3那样指望SDK自动搞定。这两个坑我花了三天才填完,希望大家能少走弯路。
下篇我们把前面所有的企业级MCP Server整合起来,搭建一套完整的生产级工具链,用docker-compose一键部署,敬请期待。
相关推荐
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
8
0- 0
已为社区贡献40条内容
所有评论(0)