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 SampleDataTool(BaseTool): def __init__(self): super().__init__( name="get_sample_data", description="Get sample data from a table with configurable limit" ) def execute(self, **kwargs) -> Dict[str, Any]: try: table_name = kwargs.get('table_name') limit = kwargs.get('limit', 10) # Validate required parameters if not table_name: return self.format_response(False, error="table_name is required") # Validate limit try: limit = int(limit) if limit <= 0 or limit > 1000: limit = 10 except (ValueError, TypeError): limit = 10 # Get sample data sample_data = query_executor.get_sample_data(table_name, limit) # Get column information for context columns = query_executor.get_table_columns(table_name) result = { 'table_name': table_name, 'sample_data': sample_data, 'sample_size': len(sample_data), 'requested_limit': limit, 'columns_info': columns } logger.info(f"Sample data retrieved for table: {table_name}, size: {len(sample_data)}") return self.format_response(True, result) except Exception as e: logger.error(f"Sample data tool failed: {str(e)}") return self.format_response(False, error=str(e))