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 SearchTool(BaseTool): def __init__(self): super().__init__( name="search_objects", description="Search for database objects (tables, columns) by name pattern" ) def execute(self, **kwargs) -> Dict[str, Any]: try: search_term = kwargs.get('search_term') search_type = kwargs.get('search_type', 'tables') # tables, columns, or both # Validate required parameters if not search_term: return self.format_response(False, error="search_term is required") results = {} # Search tables if search_type in ['tables', 'both']: tables = query_executor.search_tables_by_name(search_term) results['tables'] = tables # Search columns if search_type in ['columns', 'both']: column_query = """ SELECT TABLE_NAME as table_name, COLUMN_NAME as column_name, DATA_TYPE as data_type, COLUMN_COMMENT as comment FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND COLUMN_NAME LIKE %s ORDER BY TABLE_NAME, COLUMN_NAME """ pattern = f"%{search_term}%" columns = query_executor.db.execute_query(column_query, (pattern,)) results['columns'] = columns result = { 'search_term': search_term, 'search_type': search_type, 'results': results } logger.info(f"Search completed for term: {search_term}, type: {search_type}") return self.format_response(True, result) except Exception as e: logger.error(f"Search tool failed: {str(e)}") return self.format_response(False, error=str(e))