MCP MySQL Server

from typing import Dict, Any, List from .query_tool import QueryTool from .table_analysis_tool import TableAnalysisTool from .sample_data_tool import SampleDataTool from .list_tables_tool import ListTablesTool from .search_tool import SearchTool from .security_tool import SecurityStatusTool from .insert_tool import InsertTool from .update_tool import UpdateTool from .delete_tool import DeleteTool from .procedure_tool import ProcedureTool import logging logger = logging.getLogger(__name__) class ToolRegistry: def __init__(self): self.tools = {} self.register_tools() def register_tools(self): """Register all available tools""" tools = [ QueryTool(), TableAnalysisTool(), SampleDataTool(), ListTablesTool(), SearchTool(), SecurityStatusTool(), InsertTool(), UpdateTool(), DeleteTool(), ProcedureTool() ] for tool in tools: self.tools[tool.name] = tool logger.info(f"Registered tool: {tool.name}") def get_tool(self, tool_name: str): """Get a tool by name""" return self.tools.get(tool_name) def list_tools(self) -> List[Dict[str, str]]: """List all available tools""" return [ { 'name': tool.name, 'description': tool.description } for tool in self.tools.values() ] def execute_tool(self, tool_name: str, **kwargs) -> Dict[str, Any]: """Execute a tool by name""" tool = self.get_tool(tool_name) if not tool: return { 'success': False, 'error': f'Tool "{tool_name}" not found' } try: result = tool.execute(**kwargs) # Add timestamp from datetime import datetime result['timestamp'] = datetime.now().isoformat() return result except Exception as e: logger.error(f"Tool execution failed for {tool_name}: {str(e)}") return { 'success': False, 'error': str(e), 'tool': tool_name } # Global tool registry instance tool_registry = ToolRegistry()