Alibaba Cloud RDS OpenAPI MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/alibabacloud_rds_openapi_mcp_server/server.py:397-517 (handler)

  Handler function for the 'create_db_instance' tool. It constructs a CreateDBInstanceRequest using the provided parameters and calls the Alibaba Cloud RDS client to create a new DB instance.
  async def create_db_instance( region_id: str, engine: str, engine_version: str, dbinstance_class: str, dbinstance_storage: int, vpc_id: str, vswitch_id: str, zone_id: str, zone_id_slave1: str = None, zone_id_slave2: str = None, security_ip_list: str = "127.0.0.1", instance_network_type: str = "VPC", pay_type: str = "Postpaid", system_db_charset: str = None, dbinstance_net_type: str = "Internet", category: str = "Basic", dbinstance_storage_type: str = None, private_ip_address: str = None, client_token: str = None, resource_group_id: str = None, tde_status: str = None, encryption_key: str = None, serverless_config: Dict[str, Any] = None, table_names_case_sensitive: bool = False, db_time_zone: str = None, connection_string: str = None, db_param_group_id: str = None, ) -> Dict[str, Any]: """Create an RDS instance. Args: region_id: Region ID. engine: Database type (MySQL, SQLServer, PostgreSQL, MariaDB). engine_version: Database version. dbinstance_class: Instance specification. Query available class_codes by `describe_available_classes`. dbinstance_storage: Storage space in GB. security_ip_list: IP whitelist, separated by commas. Default: "127.0.0.1". instance_network_type: Network type (Classic, VPC). Default: VPC. zone_id: Zone ID. Query available zones by `describe_available_zones`. zone_id_slave1: Slave Node1 Zone ID. Query available zones by `describe_available_zones`. zone_id_slave2: Slave Node2 Zone ID. Query available zones by `describe_available_zones`. pay_type: Payment type (Postpaid, Prepaid). Default: Postpaid. instance_charge_type: Instance charge type. system_db_charset: Character set. dbinstance_net_type: Network connection type (Internet, Intranet). Default: Internet. category: Instance category. Default: Basic. dbinstance_storage_type: Storage type. (e.g. local_ssd,general_essd,cloud_essd,cloud_essd2,cloud_essd3) vpc_id: VPC ID. vswitch_id: VSwitch ID. private_ip_address: Private IP address. client_token: Idempotence token. resource_group_id: Resource group ID. tde_status: TDE status (Enable, Disable). encryption_key: Custom encryption key. serverless_config: Serverless instance configuration. table_names_case_sensitive: Are table names case-sensitive. db_time_zone: the db instance time zone. connection_string: the connection string for db instance. db_param_group_id: the db param group id for db instance. Returns: Dict[str, Any]: Response containing the created instance details. """ try: client = get_rds_client(region_id) request = rds_20140815_models.CreateDBInstanceRequest( region_id=region_id, engine=engine, engine_version=engine_version, dbinstance_class=dbinstance_class, dbinstance_storage=dbinstance_storage, security_iplist=security_ip_list, instance_network_type=instance_network_type, dbis_ignore_case=str(not table_names_case_sensitive).lower(), dbtime_zone=db_time_zone, connection_string=connection_string, dbparam_group_id=db_param_group_id ) # Add optional parameters if zone_id: request.zone_id = zone_id if zone_id_slave1: request.zone_id_slave_1 = zone_id_slave1 if zone_id_slave2: request.zone_id_slave_2 = zone_id_slave2 if pay_type: request.pay_type = pay_type if system_db_charset: request.system_dbcharset = system_db_charset if dbinstance_net_type: request.dbinstance_net_type = dbinstance_net_type if category: request.category = category if dbinstance_storage_type: request.dbinstance_storage_type = dbinstance_storage_type if vpc_id: request.vpcid = vpc_id if vswitch_id: request.v_switch_id = vswitch_id if private_ip_address: request.private_ip_address = private_ip_address if client_token: request.client_token = client_token if resource_group_id: request.resource_group_id = resource_group_id if tde_status: request.tde_status = tde_status if encryption_key: request.encryption_key = encryption_key if serverless_config: request.serverless_config = json.dumps(serverless_config) response = client.create_dbinstance(request) return response.body.to_map() except Exception as e: logger.error(f"Error occurred while creating RDS instance: {str(e)}") raise e