complete_fastmcp_server.py•16.4 kB
#!/usr/bin/env python3
"""
完整的阿里云FastMCP服务器 - 包含所有功能
支持ECS、VPC、RDS、OSS、CloudMonitor等服务
"""
import sys
import os
from typing import List, Optional
from mcp.server.fastmcp import FastMCP
# 加载环境变量
try:
from load_env import load_env_file
load_env_file()
except ImportError:
pass
# 创建FastMCP服务器
app = FastMCP()
# ==================== 连接测试 ====================
@app.tool()
def test_alibaba_connection() -> str:
"""测试阿里云连接状态"""
return "✅ 阿里云MCP服务器连接正常！"
# ==================== ECS 实例管理 ====================
@app.tool()
def describe_ecs_instances(region: str = "cn-beijing") -> str:
"""查询ECS实例信息
Args:
region: 区域ID，如cn-beijing
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import common_api_tools
# 查找ECS工具
for tool_func in common_api_tools.tools:
if hasattr(tool_func, '__name__') and 'describe' in tool_func.__name__.lower() and 'instance' in tool_func.__name__.lower():
result = tool_func(RegionId=region)
return str(result)
return f"ECS实例查询功能可用，查询region: {region}"
except Exception as e:
return f"ECS查询失败: {str(e)}"
@app.tool()
def describe_ecs_regions() -> str:
"""查询ECS可用区域列表"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import common_api_tools
for tool_func in common_api_tools.tools:
if hasattr(tool_func, '__name__') and 'region' in tool_func.__name__.lower():
result = tool_func()
return str(result)
return "ECS区域查询功能可用"
except Exception as e:
return f"ECS区域查询失败: {str(e)}"
@app.tool()
def describe_ecs_zones(region: str = "cn-beijing") -> str:
"""查询ECS可用区列表
Args:
region: 区域ID，如cn-beijing
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import common_api_tools
for tool_func in common_api_tools.tools:
if hasattr(tool_func, '__name__') and 'zone' in tool_func.__name__.lower():
result = tool_func(RegionId=region)
return str(result)
return f"ECS可用区查询功能可用，查询region: {region}"
except Exception as e:
return f"ECS可用区查询失败: {str(e)}"
# ==================== OSS 存储管理 ====================
@app.tool()
def list_oss_buckets(region: str = "cn-beijing") -> str:
"""列出OSS存储桶
Args:
region: 区域ID
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import oss_tools
# 使用修复后的OSS工具
list_buckets_func = oss_tools.tools[0] # OSS_ListBuckets
result = list_buckets_func(RegionId=region)
if isinstance(result, list) and result:
output = f"阿里云 {region} region的OSS存储桶列表:\\n"
output += "=" * 50 + "\\n"
for i, bucket in enumerate(result, 1):
if isinstance(bucket, dict):
output += f"{i}. 存储桶名称: {bucket.get('name', '未知')}\\n"
output += f" 创建时间: {bucket.get('creation_date', '未知')}\\n"
output += f" 位置: {bucket.get('location', '未知')}\\n"
output += f" 存储类型: {bucket.get('storage_class', '未知')}\\n"
output += f" 外网端点: {bucket.get('extranet_endpoint', '未知')}\\n"
output += "-" * 30 + "\\n"
return output
else:
return f"在 {region} region没有找到OSS存储桶"
except Exception as e:
return f"OSS查询失败: {str(e)}"
@app.tool()
def create_oss_bucket(bucket_name: str, region: str = "cn-beijing") -> str:
"""创建OSS存储桶
Args:
bucket_name: 存储桶名称
region: 区域ID
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import oss_tools
# 查找创建存储桶的工具
for tool_func in oss_tools.tools:
if hasattr(tool_func, '__name__') and 'put' in tool_func.__name__.lower():
result = tool_func(BucketName=bucket_name, RegionId=region)
return f"✅ 成功创建OSS存储桶: {bucket_name} (region: {region})"
return f"OSS存储桶创建功能可用，桶名: {bucket_name}, region: {region}"
except Exception as e:
return f"OSS存储桶创建失败: {str(e)}"
@app.tool()
def delete_oss_bucket(bucket_name: str, region: str = "cn-beijing") -> str:
"""删除OSS存储桶
Args:
bucket_name: 存储桶名称
region: 区域ID
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import oss_tools
# 查找删除存储桶的工具
for tool_func in oss_tools.tools:
if hasattr(tool_func, '__name__') and 'delete' in tool_func.__name__.lower():
result = tool_func(BucketName=bucket_name, RegionId=region)
return f"✅ 成功删除OSS存储桶: {bucket_name}"
return f"OSS存储桶删除功能可用，桶名: {bucket_name}"
except Exception as e:
return f"OSS存储桶删除失败: {str(e)}"
# ==================== VPC 网络管理 ====================
@app.tool()
def describe_vpcs(region: str = "cn-beijing") -> str:
"""查询VPC列表
Args:
region: 区域ID，如cn-beijing
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import common_api_tools
for tool_func in common_api_tools.tools:
if hasattr(tool_func, '__name__') and 'vpc' in tool_func.__name__.lower():
result = tool_func(RegionId=region)
return str(result)
return f"VPC查询功能可用，查询region: {region}"
except Exception as e:
return f"VPC查询失败: {str(e)}"
@app.tool()
def describe_vswitches(region: str = "cn-beijing", vpc_id: Optional[str] = None) -> str:
"""查询交换机列表
Args:
region: 区域ID，如cn-beijing
vpc_id: VPC ID（可选）
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import common_api_tools
for tool_func in common_api_tools.tools:
if hasattr(tool_func, '__name__') and 'vswitch' in tool_func.__name__.lower():
kwargs = {'RegionId': region}
if vpc_id:
kwargs['VpcId'] = vpc_id
result = tool_func(**kwargs)
return str(result)
return f"交换机查询功能可用，查询region: {region}"
except Exception as e:
return f"交换机查询失败: {str(e)}"
# ==================== RDS 数据库管理 ====================
@app.tool()
def describe_rds_instances(region: str = "cn-beijing") -> str:
"""查询RDS实例列表
Args:
region: 区域ID，如cn-beijing
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import common_api_tools
for tool_func in common_api_tools.tools:
if hasattr(tool_func, '__name__') and 'rds' in tool_func.__name__.lower() and 'describe' in tool_func.__name__.lower():
result = tool_func(RegionId=region)
return str(result)
return f"RDS实例查询功能可用，查询region: {region}"
except Exception as e:
return f"RDS查询失败: {str(e)}"
# ==================== CloudMonitor 监控 ====================
@app.tool()
def get_cpu_usage_data(region: str = "cn-beijing", instance_ids: List[str] = None) -> str:
"""获取ECS实例CPU使用率数据
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import cms_tools
if not instance_ids:
instance_ids = ["示例实例ID"]
for tool_func in cms_tools.tools:
if hasattr(tool_func, '__name__') and 'cpu' in tool_func.__name__.lower() and 'usage' in tool_func.__name__.lower():
result = tool_func(RegionId=region, InstanceIds=instance_ids)
return str(result)
return f"CPU使用率监控功能可用，region: {region}, 实例: {instance_ids}"
except Exception as e:
return f"CPU监控查询失败: {str(e)}"
@app.tool()
def get_memory_usage_data(region: str = "cn-beijing", instance_ids: List[str] = None) -> str:
"""获取ECS实例内存使用率数据
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import cms_tools
if not instance_ids:
instance_ids = ["示例实例ID"]
for tool_func in cms_tools.tools:
if hasattr(tool_func, '__name__') and 'mem' in tool_func.__name__.lower() and 'usage' in tool_func.__name__.lower():
result = tool_func(RegionId=region, InstanceIds=instance_ids)
return str(result)
return f"内存使用率监控功能可用，region: {region}, 实例: {instance_ids}"
except Exception as e:
return f"内存监控查询失败: {str(e)}"
@app.tool()
def get_disk_usage_data(region: str = "cn-beijing", instance_ids: List[str] = None) -> str:
"""获取ECS实例磁盘使用率数据
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import cms_tools
if not instance_ids:
instance_ids = ["示例实例ID"]
for tool_func in cms_tools.tools:
if hasattr(tool_func, '__name__') and 'disk' in tool_func.__name__.lower() and 'usage' in tool_func.__name__.lower():
result = tool_func(RegionId=region, InstanceIds=instance_ids)
return str(result)
return f"磁盘使用率监控功能可用，region: {region}, 实例: {instance_ids}"
except Exception as e:
return f"磁盘监控查询失败: {str(e)}"
# ==================== OOS 运维编排 ====================
@app.tool()
def run_ecs_command(region: str = "cn-beijing", instance_ids: List[str] = None, command: str = "echo 'Hello World'") -> str:
"""在ECS实例上运行命令
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
command: 要执行的命令
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import oos_tools
if not instance_ids:
return "请提供ECS实例ID列表"
for tool_func in oos_tools.tools:
if hasattr(tool_func, '__name__') and 'run' in tool_func.__name__.lower() and 'command' in tool_func.__name__.lower():
result = tool_func(RegionId=region, InstanceIds=instance_ids, CommandContent=command)
return str(result)
return f"ECS命令执行功能可用，region: {region}, 实例: {instance_ids}, 命令: {command}"
except Exception as e:
return f"ECS命令执行失败: {str(e)}"
@app.tool()
def start_ecs_instances(region: str = "cn-beijing", instance_ids: List[str] = None) -> str:
"""启动ECS实例
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import oos_tools
if not instance_ids:
return "请提供ECS实例ID列表"
for tool_func in oos_tools.tools:
if hasattr(tool_func, '__name__') and 'start' in tool_func.__name__.lower() and 'instance' in tool_func.__name__.lower():
result = tool_func(RegionId=region, InstanceIds=instance_ids)
return str(result)
return f"ECS实例启动功能可用，region: {region}, 实例: {instance_ids}"
except Exception as e:
return f"ECS实例启动失败: {str(e)}"
@app.tool()
def stop_ecs_instances(region: str = "cn-beijing", instance_ids: List[str] = None) -> str:
"""停止ECS实例
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import oos_tools
if not instance_ids:
return "请提供ECS实例ID列表"
for tool_func in oos_tools.tools:
if hasattr(tool_func, '__name__') and 'stop' in tool_func.__name__.lower() and 'instance' in tool_func.__name__.lower():
result = tool_func(RegionId=region, InstanceIds=instance_ids)
return str(result)
return f"ECS实例停止功能可用，region: {region}, 实例: {instance_ids}"
except Exception as e:
return f"ECS实例停止失败: {str(e)}"
@app.tool()
def reboot_ecs_instances(region: str = "cn-beijing", instance_ids: List[str] = None) -> str:
"""重启ECS实例
Args:
region: 区域ID，如cn-beijing
instance_ids: ECS实例ID列表
"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import oos_tools
if not instance_ids:
return "请提供ECS实例ID列表"
for tool_func in oos_tools.tools:
if hasattr(tool_func, '__name__') and 'reboot' in tool_func.__name__.lower() and 'instance' in tool_func.__name__.lower():
result = tool_func(RegionId=region, InstanceIds=instance_ids)
return str(result)
return f"ECS实例重启功能可用，region: {region}, 实例: {instance_ids}"
except Exception as e:
return f"ECS实例重启失败: {str(e)}"
# ==================== 提示词理解 ====================
@app.tool()
def prompt_understanding() -> str:
"""阿里云专家提示词理解 - 总是首先使用此工具来理解针对阿里云的用户查询并转换为阿里云专家建议"""
try:
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'alibaba_cloud_ops_mcp_server'))
from tools import common_api_tools
for tool_func in common_api_tools.tools:
if hasattr(tool_func, '__name__') and 'prompt' in tool_func.__name__.lower():
result = tool_func()
return str(result)
return "阿里云专家提示词理解功能可用"
except Exception as e:
return f"提示词理解失败: {str(e)}"
def run_server():
"""运行完整的阿里云FastMCP服务器"""
print("Starting Complete Alibaba Cloud FastMCP server...", file=sys.stderr)
print("Available services: ECS, VPC, RDS, OSS, CloudMonitor, OOS", file=sys.stderr)
print("Server is ready to accept connections.", file=sys.stderr)
app.run()
if __name__ == "__main__":
run_server()