executeScript

Instructions

Implementation Reference

• src/alibabacloud_dms_mcp_server/server.py:360-393 (handler)

  Core handler function `execute_script` that executes SQL scripts via the DMS client. Takes database_id, script, and logic parameters, calls the Aliyun DMS API, processes results into ExecuteScriptResult with Markdown table formatting using the _format_as_markdown_table helper.

  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身份权限。"

• src/alibabacloud_dms_mcp_server/server.py:97-114 (schema)

  ExecuteScriptResult pydantic model - the return type schema for execute_script. Contains RequestId, Results (list of ResultSet), and Success fields, with a __str__ method for markdown rendering.

  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:88-94 (schema)

  ResultSet pydantic model - schema for individual result sets within ExecuteScriptResult. Contains ColumnNames, RowCount, Rows, MarkdownTable, Success, and Message fields.

  class ResultSet(MyBaseModel):
      ColumnNames: List[str] = Field(description="Ordered list of column names")
      RowCount: int = Field(description="Number of rows returned")
      Rows: List[Dict[str, Any]] = Field(description="List of rows, where each row is a dictionary of column_name: value")
      MarkdownTable: Optional[str] = Field(default=None, description="Data formatted as a Markdown table string")
      Success: bool = Field(description="Whether this result set was successfully retrieved")
      Message: str = Field(description="Additional message returned")

• src/alibabacloud_dms_mcp_server/server.py:750-760 (registration)

  Registration of the executeScript tool in the full toolset (via _register_full_toolset). Wraps the core execute_script with database_id, script, and logic parameters.

  @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:618-626 (registration)

  Registration of the executeScript tool in the pre-configured DB toolset (via _register_configured_db_toolset). Wraps the core execute_script using the default_database_id, exposing only script parameter.

  @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)