MCP MySQL Server

from typing import Dict, Any from .base_tool import BaseTool from src.database.query_executor import query_executor import logging logger = logging.getLogger(__name__) class TableAnalysisTool(BaseTool): def __init__(self): super().__init__( name="analyze_table", description="Analyze table structure including columns, indexes, foreign keys, and basic statistics" ) def execute(self, **kwargs) -> Dict[str, Any]: try: table_name = kwargs.get('table_name') # Validate required parameters if not table_name: return self.format_response(False, error="table_name is required") # Get table columns columns = query_executor.get_table_columns(table_name) if not columns: return self.format_response(False, error=f"Table '{table_name}' not found") # Get table indexes indexes = query_executor.get_table_indexes(table_name) # Get foreign keys foreign_keys = query_executor.get_foreign_keys(table_name) # Get table statistics stats = query_executor.get_table_stats(table_name) # Organize indexes by name indexes_grouped = {} for index in indexes: index_name = index['index_name'] if index_name not in indexes_grouped: indexes_grouped[index_name] = { 'name': index_name, 'unique': index['non_unique'] == 0, 'type': index['index_type'], 'columns': [] } indexes_grouped[index_name]['columns'].append(index['column_name']) analysis_result = { 'table_name': table_name, 'columns': columns, 'indexes': list(indexes_grouped.values()), 'foreign_keys': foreign_keys, 'statistics': stats } logger.info(f"Table analysis completed for: {table_name}") return self.format_response(True, analysis_result) except Exception as e: logger.error(f"Table analysis failed: {str(e)}") return self.format_response(False, error=str(e))