createDataChangeOrder
Destructive
Execute SQL changes through a formal data change order in Alibaba Cloud DMS, returning an order ID for tracking and compliance purposes.
Instructions
Execute SQL changes through a data change order, and a corresponding order ID will be returned. Prefer using the executeScript tool for SQL execution;only use this tool when explicitly instructed to perform the operation via a order.If you don't know the databaseId, first use getDatabase or searchDatabase to retrieve it. (1) If you have the exact host, port, and database name, use getDatabase. (2) If you only know the database name, use searchDatabase. (3) If you don't know any information, ask the user to provide the necessary details. Note: searchDatabase may return multiple databases. In this case, let the user choose which one to use.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database_id | Yes | Required DMS databaseId. Obtained via getDatabase tool | |
| script | Yes | SQL script to execute | |
| logic | No | Whether to use logical execution mode |
Implementation Reference
- Core handler function implementing the logic to create a data change order via Alibaba Cloud DMS API. Constructs the request parameters including SQL script, database ID, and logic mode, then invokes the API.
async def create_data_change_order( database_id: str = Field(description="DMS databaseId"), script: str = Field(description="SQL script to execute"), logic: bool = Field(default=False, description="Whether to use logical execution mode") ) -> Dict[str, Any]: client = create_client() req = dms_enterprise_20181101_models.CreateDataCorrectOrderRequest() req.comment = "Data correct order submitted by MCP" param = dms_enterprise_20181101_models.CreateDataCorrectOrderRequestParam() param.estimate_affect_rows = 1 param.sql_type = "TEXT" param.exec_sql = script param.classify = "MCP" db_list = dms_enterprise_20181101_models.CreateDataCorrectOrderRequestParamDbItemList() db_list.db_id = database_id db_list.logic = logic db_items = [db_list] param.db_item_list = db_items req.param = param try: resp = client.create_data_correct_order(req) return resp.body.to_map() except Exception as e: logger.error(f"Error in create_data_change_order: {e}") raise - src/alibabacloud_dms_mcp_server/server.py:599-609 (registration)Tool registration for pre-configured database mode, using default_database_id. Wrapper calls the core handler.
@self.mcp.tool(name="createDataChangeOrder", description="Execute SQL changes through a data change order, and a corresponding order ID will be returned. " "Prefer using the executeScript tool for SQL execution; " "only use this tool when explicitly instructed to perform the operation via a order.", annotations={"title": "在DMS中创建数据变更工单", "readOnlyHint": False, "destructiveHint": True}) async def create_data_change_order_configured( script: str = Field(description="SQL script to execute") ) -> str: result_obj = await create_data_change_order(database_id=self.default_database_id, script=script, logic=False) return str(result_obj) - src/alibabacloud_dms_mcp_server/server.py:727-739 (registration)Tool registration for full toolset, requiring explicit database_id. Wrapper calls the core handler.
@self.mcp.tool(name="createDataChangeOrder", description=f"Execute SQL changes through a data change order, and a corresponding order ID will be returned. " f"Prefer using the executeScript tool for SQL execution;" f"only use this tool when explicitly instructed to perform the operation via a order." f"{DATABASE_ID_DESCRIPTION}", annotations={"title": "在DMS中创建数据变更工单", "readOnlyHint": False, "destructiveHint": True}) async def create_data_change_order_wrapper( database_id: str = Field(description="Required DMS databaseId. Obtained via getDatabase tool"), script: str = Field(description="SQL script to execute"), logic: bool = Field(description="Whether to use logical execution mode", default=False) ) -> str: # Return string representation result_obj = await create_data_change_order(database_id=database_id, script=script, logic=logic) return str(result_obj)