MCP SQL Server

"""MCP SQL Server main orchestrator.""" from typing import Any import inspect from functools import wraps from .credentials import CredentialsManager from .connection import ConnectionManager from .inspector import DatabaseInspector from .executor import QueryExecutor from .dependencies import ensure_deps_once from .tools import ALL_TOOLS, MCPTool try: from fastmcp import FastMCP, Context except ImportError: ensure_deps_once() from fastmcp import FastMCP, Context class MCPSQLServer: """Main MCP SQL Server orchestrator with plugin-based tool architecture.""" def __init__(self): # Initialize core components self.connection_manager = ConnectionManager() self.credentials_manager = CredentialsManager() self.inspector = DatabaseInspector(self.connection_manager) self.executor = QueryExecutor(self.connection_manager) self.mcp = FastMCP("mcp-sql-explicit") # Initialize and register all tools self.tools: list[MCPTool] = [] self._initialize_tools() self._register_tools() def _initialize_tools(self): """Initialize all tool instances.""" for tool_class in ALL_TOOLS: tool_instance = tool_class( connection_manager=self.connection_manager, credentials_manager=self.credentials_manager, inspector=self.inspector, executor=self.executor ) self.tools.append(tool_instance) print(f"✅ Initialized tool: {tool_instance.name}") def _register_tools(self): """Register all tools with FastMCP.""" for tool in self.tools: self._register_single_tool(tool) def _register_single_tool(self, tool: MCPTool): """Register a single tool with FastMCP. Args: tool: The tool instance to register """ # Get the signature of the execute method execute_method = tool.execute sig = inspect.signature(execute_method) # Create a wrapper function that preserves the exact signature # but excludes 'self' (FastMCP will inject ctx automatically) params = [] for param_name, param in sig.parameters.items(): if param_name == 'self': continue params.append(param) # Build the new signature new_sig = sig.replace(parameters=params) # Create the wrapper function dynamically @wraps(execute_method) async def wrapper(*args, **kwargs): return await tool.execute(*args, **kwargs) # Apply the signature to the wrapper wrapper.__signature__ = new_sig wrapper.__name__ = tool.name wrapper.__doc__ = tool.description # Register with FastMCP self.mcp.tool()(wrapper) print(f"🔧 Registered MCP tool: {tool.name}") def add_custom_tool(self, tool: MCPTool): """Add a custom tool dynamically. Args: tool: Custom tool instance that inherits from MCPTool Example: >>> class MyCustomTool(MCPTool): ... @property ... def name(self): return "my_custom_tool" ... # ... implement other methods >>> >>> server = MCPSQLServer() >>> server.add_custom_tool(MyCustomTool( ... server.connection_manager, ... server.credentials_manager, ... server.inspector, ... server.executor ... )) """ self.tools.append(tool) self._register_single_tool(tool) print(f"➕ Added custom tool: {tool.name}") def run(self, port: int): """Start the MCP server. Args: port: Port number to bind the server """ print(f"\n{'='*60}") print(f"🚀 Starting MCP SQL Server") print(f"{'='*60}") print(f"📍 Port: {port}") print(f"🖥️ Configured servers: {self.connection_manager.get_configured_server_names()}") print(f"🔧 Registered tools: {len(self.tools)}") for tool in self.tools: print(f" • {tool.name}") print(f"{'='*60}\n") self.mcp.run(transport="streamable-http", host="0.0.0.0", port=port)