executeScript
Run SQL scripts on a database in Alibaba Cloud DMS and retrieve structured results. Use getDatabase or searchDatabase to obtain the databaseId if unknown. Supports logical execution mode for advanced query handling.
Instructions
Execute SQL script against a database in DMS and return structured results.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 | |
| logic | No | Whether to use logical execution mode | |
| script | Yes | SQL script to execute |
Implementation Reference
- Pydantic model defining the structure of the executeScript tool response, including request ID, list of ResultSet objects, and success status. Includes __str__ method for string representation used by tool wrappers.class ExecuteScriptResult(MyBaseModel): RequestId: str = Field(description="Unique request identifier") Results: List[ResultSet] = Field(description="List of result sets from executed script") Success: bool = Field(description="Overall operation success status") def __str__(self) -> str: if self.Success and self.Results: first_result = self.Results[0] if first_result.Success and first_result.MarkdownTable: return first_result.MarkdownTable elif not first_result.Success: return first_result.Message else: return "Result data is not available in Markdown format." elif not self.Success: return "Script execution failed." else: return "Script executed successfully, but no results were returned."
- Core handler function that performs the actual SQL script execution by calling the Alibaba Cloud DMS Enterprise API's execute_script method, processes the response into structured ResultSets with Markdown tables, and handles errors.async def execute_script( 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") ) -> ExecuteScriptResult: # Return the object, __str__ will be used by wrapper if needed client = create_client() req = dms_enterprise_20181101_models.ExecuteScriptRequest(db_id=database_id, script=script, logic=logic) if mcp.state.real_login_uid: req.real_login_user_uid = mcp.state.real_login_uid try: resp = client.execute_script(req) if not resp or not resp.body: return ExecuteScriptResult(RequestId="", Results=[], Success=False) data = resp.body.to_map() processed_results = [] if data.get('Success') and data.get('Results'): for res_item in data.get('Results', []): if res_item.get('Success'): column_names = res_item.get('ColumnNames', []) rows_data = res_item.get('Rows', []) markdown_table = _format_as_markdown_table(column_names, rows_data) processed_results.append( ResultSet(ColumnNames=column_names, RowCount=res_item.get('RowCount', 0), Rows=rows_data, MarkdownTable=markdown_table, Success=True, Message='')) else: processed_results.append( ResultSet(ColumnNames=[], RowCount=0, Rows=[], MarkdownTable=None, Success=False, Message=res_item.get('Message'))) return ExecuteScriptResult(RequestId=data.get('RequestId', ""), Results=processed_results, Success=data.get('Success', False)) except Exception as e: logger.error(f"Error in execute_script: {e}") if "The instance is not in secure hosting mode" in str(e): return "当前实例尚未开启安全托管功能。您可以通过DMS控制台免费开启「安全托管模式」。请注意，该操作需要管理员或DBA身份权限。"
- MCP tool handler for executeScript when a default database is pre-configured. Calls the core execute_script with default_database_id and returns string representation.@self.mcp.tool(name="executeScript", description="Executes an SQL script against the pre-configured database.", annotations={"title": "Execute SQL (Pre-configured DB)", "readOnlyHint": False, "destructiveHint": True}) async def execute_script_configured( script: str = Field(description="SQL script to execute") ) -> str: result_obj = await execute_script(database_id=self.default_database_id, script=script, logic=False) return str(result_obj)
- Full MCP tool handler for executeScript requiring explicit database_id parameter. Calls the core execute_script and returns string representation.@self.mcp.tool(name="executeScript", description=f"Execute SQL script against a database in DMS and return structured results." f"{DATABASE_ID_DESCRIPTION}", annotations={"title": "在DMS中执行SQL脚本", "readOnlyHint": False, "destructiveHint": True}) async def execute_script_full_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 execute_script(database_id=database_id, script=script, logic=logic) return str(result_obj)