executeScript
Execute SQL scripts on Alibaba Cloud DMS databases to query, update, or manage data with structured results.
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 | |
| script | Yes | SQL script to execute | |
| logic | No | Whether to use logical execution mode |
Implementation Reference
- Core handler function that executes the SQL script using the Alibaba Cloud DMS Enterprise API client, processes the response into structured ResultSet objects with Markdown tables, and returns an ExecuteScriptResult object.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身份权限。"
- Pydantic model defining the output schema for the executeScript tool, including request ID, list of ResultSets, success flag, and a __str__ method for string representation.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."
- src/alibabacloud_dms_mcp_server/server.py:715-725 (registration)Registration of the executeScript tool in the full toolset mode, wrapping the core execute_script function to require database_id parameter and return 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)
- src/alibabacloud_dms_mcp_server/server.py:589-598 (registration)Registration of the executeScript tool in the configured database mode (when default_database_id is set), wrapping the core function without requiring database_id.@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)
- Helper function used by execute_script to format query results as Markdown tables.def _format_as_markdown_table(column_names: List[str], rows: List[Dict[str, Any]]) -> str: if column_names: header = "| " + " | ".join(str(c) for c in column_names) + " |" separator = "| " + " | ".join(["---"] * len(column_names)) + " |" table_rows_str = [header, separator] for row_data in rows: row_values = [str(row_data.get(col, "")) for col in column_names] table_rows_str.append("| " + " | ".join(row_values) + " |") return "\n".join(table_rows_str) if not rows: return "Success." else: all_keys = set() for r in rows: all_keys.update(r.keys()) if all_keys: fallback_columns = sorted(all_keys) return _format_as_markdown_table(fallback_columns, rows) else: return "No data returned."