Alibaba Cloud Operations MCP Server

Instructions

Args:
    region: 区域ID，如cn-beijing
    instance_ids: ECS实例ID列表

Implementation Reference

• complete_fastmcp_server.py:279-301 (handler)

  Handler and registration for the 'get_disk_usage_data' MCP tool. Dynamically invokes CMS_GetDiskUsageData from cms_tools by matching function name keywords.

  @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)}"

• alibaba_cloud_ops_mcp_server/tools/cms_tools.py:92-97 (helper)

  Direct helper function invoked by the handler to fetch disk usage data. Defines input schema with pydantic Fields and delegates to core metric fetcher.

  def CMS_GetDiskUsageData(
      InstanceIds: List[str] = Field(description='AlibabaCloud ECS instance ID List'),
      RegionId: str = Field(description='AlibabaCloud region ID', default='cn-hangzhou')
  ):
      """获取磁盘利用率指标数据"""
      return _get_cms_metric_data(RegionId, InstanceIds, 'diskusage_utilization')

• alibaba_cloud_ops_mcp_server/tools/cms_tools.py:23-36 (helper)

  Core utility function that performs the actual API call to Alibaba Cloud CMS to retrieve metric data (diskusage_utilization for this tool).

  def _get_cms_metric_data(region_id: str, instance_ids: List[str], metric_name: str):
      client = create_client(region_id)
      dimesion = []
      for instance_id in instance_ids:
          dimesion.append({
              'instanceId': instance_id
          })
      describe_metric_last_request = cms_20190101_models.DescribeMetricLastRequest(
          namespace='acs_ecs_dashboard',
          metric_name=metric_name,
          dimensions=json.dumps(dimesion),
      )
      describe_metric_last_resp = client.describe_metric_last(describe_metric_last_request)
      return describe_metric_last_resp.body.datapoints