Alibaba Cloud RDS OpenAPI MCP Server

Instructions

Implementation Reference

• src/alibabacloud_rds_openapi_mcp_server/server.py:1396-1476 (handler)

  The core handler function for the 'describe_monitor_metrics' tool. It queries performance metrics from Alibaba Cloud DAS (Database Autonomy Service) using the GetPerformanceMetrics API, processes the data by grouping metrics by timestamp, and formats the output as a markdown table.
  async def describe_monitor_metrics( dbinstance_id: str, metrics_list: list[str], db_type: str, start_time: str, end_time: str, ): """ Queries performance and diagnostic metrics for an instance using the DAS (Database Autonomy Service) API. This method provides extra monitoring and diagnostic data which cannot be queried by describe_db_instance_performance, such as IOPSUsage, MdlLockSession, etc. Args: dbinstance_id (str): The ID of the RDS instance. metrics_list (list[str]): The metrics to query. (e.g. ["IOBytesPS", "IOPSUsage", "MdlLockSession", "DiskUsage"]) db_type (str): The type of the database. (e.g. "mysql") start_time(str): the start time. e.g. 2025-06-06 20:00:00 end_time(str): the end time. e.g. 2025-06-06 20:10:00 Returns: the monitor metrics information. """ try: # Initialize client client = get_das_client() metrics = transform_das_key(db_type, metrics_list) if not metrics: raise OpenAPIError(f"Unsupported das_metric_key: {metrics_list}") start_time = convert_datetime_to_timestamp(start_time) end_time = convert_datetime_to_timestamp(end_time) # 通过 interval 控制查询粒度 interval = int(max((end_time - start_time) / 30000, 5)) body = { "InstanceId": dbinstance_id, "Metrics": ",".join(metrics), "StartTime": start_time, "EndTime": end_time, "Interval": interval } req = open_api_models.OpenApiRequest( query=OpenApiUtilClient.query({}), body=OpenApiUtilClient.parse_to_map(body) ) params = open_api_models.Params( action='GetPerformanceMetrics', version='2020-01-16', protocol='HTTPS', pathname='/', method='POST', auth_type='AK', style='RPC', req_body_type='formData', body_type='json' ) response = client.call_api(params, req, util_models.RuntimeOptions()) response_data = response['body']['Data'] timestamp_map = {} resp_metrics_list = set() for metric in response_data: name = metric["Name"] values = metric["Value"] timestamps = metric["Timestamp"] resp_metrics_list.add(name) for timestamp, value in zip(timestamps, values): dt = transform_timestamp_to_datetime(timestamp) if dt not in timestamp_map: timestamp_map[dt] = {} timestamp_map[dt][name] = value headers = sorted(list(resp_metrics_list)) datas = [] for dt in sorted(timestamp_map.keys()): value_map = timestamp_map[dt] value_map["datetime"] = dt datas.append(value_map) headers.insert(0, "datetime") return json_array_to_markdown(headers, datas) except Exception as e: logger.error(f"Error occurred: {str(e)}") raise e
• src/alibabacloud_rds_openapi_mcp_server/utils.py:54-62 (helper)

  Defines the mapping of high-level metric names to actual DAS metric keys for MySQL, used by transform_das_key in describe_monitor_metrics.
  DAS_KEYS = { "mysql": { "DiskUsage": ["disk_usage"], "IOPSUsage": ["data_iops_usage"], "IOBytesPS": ["data_io_bytes_ps"], "MdlLockSession": ["mdl_lock_session"] } }
• src/alibabacloud_rds_openapi_mcp_server/utils.py:118-126 (helper)

  Helper function that transforms user-provided metrics_list into the actual DAS metric names using the DAS_KEYS mapping, specific to the db_type.
  def transform_das_key(db_type: str, das_keys: list[str]): das_key_after_transform = [] for key in das_keys: if key in DAS_KEYS[db_type.lower()]: das_key_after_transform.extend(DAS_KEYS[db_type.lower()][key]) else: das_key_after_transform.append(key) return das_key_after_transform
• src/alibabacloud_rds_openapi_mcp_server/utils.py:248-261 (helper)

  Helper function that creates and returns the DAS (Database Autonomy Service) client instance used for API calls in describe_monitor_metrics.
  def get_das_client(): ak, sk, sts = get_aksk() config = Config( access_key_id=ak, access_key_secret=sk, security_token=sts, region_id='cn-shanghai', protocol="https", connect_timeout=10 * 1000, read_timeout=300 * 1000 ) client = DAS20200116Client(config) return client