MCP SQL Server

# MCP Tools/Services Architecture Arquitectura basada en plugins donde cada función MCP es una clase independiente. ## 📐 Estructura ``` mcp_sql/tools/ ├── __init__.py # Registry de todos los servicios ├── base.py # Clase base abstracta MCPTool ├── list_servers.py # Servicio: listar servidores ├── get_databases.py # Servicio: obtener bases de datos ├── get_tables.py # Servicio: obtener tablas ├── explore_table.py # Servicio: explorar tabla ├── describe_table.py # Servicio: describir tabla ├── query_columns.py # Servicio: consultar columnas └── README.md # Esta documentación ``` ## 🎯 Servicios Disponibles | Servicio | Clase | Descripción | |----------|-------|-------------| | `list_configured_servers` | `ListServersService` | Lista servidores configurados | | `get_databases` | `GetDatabasesService` | Obtiene lista de bases de datos | | `get_tables` | `GetTablesService` | Obtiene lista de tablas | | `explore_table` | `ExploreTableService` | Muestra datos de ejemplo | | `describe_table` | `DescribeTableService` | Descripción detallada de tabla | | `query_table_with_columns` | `QueryColumnsService` | Consulta columnas específicas | | `execute_select_query` | `ExecuteSelectService` | Ejecuta consultas SELECT seguras (bloquea ALTER, INSERT, CREATE, UPDATE, DELETE, etc.) | ## 🔧 Clase Base: MCPTool Todos los servicios heredan de `MCPTool`: ```python class MCPTool(ABC): """Abstract base class for MCP tools.""" def __init__( self, connection_manager: ConnectionManager, credentials_manager: CredentialsManager, inspector: DatabaseInspector, executor: QueryExecutor ): # Acceso a todos los componentes principales pass @property @abstractmethod def name(self) -> str: """Tool name (used for registration).""" pass @property @abstractmethod def description(self) -> str: """Tool description (used in MCP metadata).""" pass @abstractmethod async def execute(self, ctx: Context, **kwargs) -> Any: """Execute the tool logic.""" pass ``` ## ➕ Crear un Nuevo Servicio ### Paso 1: Crear el archivo del servicio Crea un nuevo archivo en `mcp_sql/tools/` (ej: `analyze_query.py`): ```python """Analyze query performance tool.""" from typing import Any, Optional, Dict from fastmcp import Context from .base import MCPTool class AnalyzeQueryService(MCPTool): """Service to analyze query performance and execution plan.""" @property def name(self) -> str: return "analyze_query" @property def description(self) -> str: return "Analyze SQL query performance and get execution plan" async def execute( self, ctx: Context, sql: str, database: Optional[str] = None, server_name: Optional[str] = None, user: Optional[str] = None, password: Optional[str] = None, driver: Optional[str] = None, port: Optional[int] = None, **kwargs ) -> Dict[str, Any]: """Analyze query execution. Args: ctx: FastMCP context sql: SQL query to analyze database: Database name server_name: Server hostname user: Username password: Password driver: ODBC driver port: Server port Returns: dict: Analysis results including execution plan """ # Obtener credenciales creds = self.creds_manager.get_from_context( ctx, user, password, server_name, database, driver, port ) if not creds.is_valid(): return {"error": "Missing credentials"} # Obtener engine engine = self.conn_manager.get_engine_with_credentials(creds) if not engine: return {"error": "Could not create connection"} # Tu lógica de análisis aquí try: with engine.connect() as conn: # Ejemplo: obtener execution plan en SQL Server result = conn.execute(f"SET SHOWPLAN_TEXT ON; {sql}") plan = [row for row in result] return { "query": sql, "execution_plan": plan, "database": creds.database, "server": creds.server } except Exception as e: return {"error": f"Analysis failed: {str(e)}"} ``` ### Paso 2: Registrar en `__init__.py` Edita `mcp_sql/tools/__init__.py`: ```python from .analyze_query import AnalyzeQueryService ALL_TOOLS = [ ListServersService, GetDatabasesService, GetTablesService, ExploreTableService, DescribeTableService, QueryColumnsService, AnalyzeQueryService, # ← Agregar aquí ] __all__ = [ # ... existentes ... 'AnalyzeQueryService', ] ``` ### Paso 3: ¡Listo! El servicio se registrará automáticamente al iniciar el servidor: ``` ✅ Initialized tool: analyze_query 🔧 Registered MCP tool: analyze_query ``` ## 🚀 Agregar Servicio Dinámicamente También puedes agregar servicios en tiempo de ejecución sin modificar archivos: ```python from mcp_sql import MCPSQLServer from mcp_sql.tools import MCPTool from fastmcp import Context class CustomReportService(MCPTool): @property def name(self) -> str: return "generate_report" @property def description(self) -> str: return "Generate custom database report" async def execute(self, ctx: Context, **kwargs): # Tu lógica aquí return {"report": "data"} # Crear servidor server = MCPSQLServer() # Agregar servicio personalizado custom_tool = CustomReportService( connection_manager=server.connection_manager, credentials_manager=server.credentials_manager, inspector=server.inspector, executor=server.executor ) server.add_custom_tool(custom_tool) # Iniciar servidor server.run(port=3939) ``` ## 💡 Ejemplos de Servicios Personalizados ### 1. Servicio de Backup ```python class BackupDatabaseService(MCPTool): @property def name(self) -> str: return "backup_database" @property def description(self) -> str: return "Create database backup" async def execute(self, ctx: Context, database: str, **kwargs): creds = self.creds_manager.get_from_context(ctx, **kwargs) engine = self.conn_manager.get_engine_with_credentials(creds) # Ejecutar BACKUP DATABASE... return {"status": "backup_created", "file": "path/to/backup"} ``` ### 2. Servicio de Estadísticas ```python class DatabaseStatsService(MCPTool): @property def name(self) -> str: return "get_database_stats" @property def description(self) -> str: return "Get comprehensive database statistics" async def execute(self, ctx: Context, database: str, **kwargs): creds = self.creds_manager.get_from_context(ctx, **kwargs) tables = self.inspector.get_tables(creds) stats = { "database": database, "table_count": len(tables), "tables": [] } for table in tables: table_info = self.inspector.describe_table(creds, table) stats["tables"].append({ "name": table, "rows": table_info.get("row_count", 0), "columns": len(table_info.get("columns", [])) }) return stats ``` ### 3. Servicio de Validación de Esquema ```python class ValidateSchemaService(MCPTool): @property def name(self) -> str: return "validate_schema" @property def description(self) -> str: return "Validate database schema against rules" async def execute( self, ctx: Context, database: str, rules: dict, **kwargs ): creds = self.creds_manager.get_from_context(ctx, **kwargs) tables = self.inspector.get_tables(creds) violations = [] for table in tables: table_info = self.inspector.describe_table(creds, table) # Validar reglas if rules.get("require_primary_key"): if not table_info.get("primary_key"): violations.append({ "table": table, "rule": "require_primary_key", "message": "Table missing primary key" }) return { "valid": len(violations) == 0, "violations": violations } ``` ## 🔍 Acceso a Componentes en Servicios Dentro de cualquier servicio, tienes acceso a: ```python # Gestión de conexiones self.conn_manager.get_engine_with_credentials(creds) self.conn_manager.get_configured_server_names() # Gestión de credenciales self.creds_manager.get_from_context(ctx, **kwargs) # Inspección de base de datos self.inspector.get_databases(creds) self.inspector.get_tables(creds) self.inspector.describe_table(creds, table) # Ejecución de queries self.executor.explore_table(creds, table, limit) self.executor.query_columns(creds, table, columns, limit) ``` ## 🎨 Ventajas de esta Arquitectura 1. ✅ **Modular**: Cada servicio es independiente 2. ✅ **Extensible**: Agregar servicios sin modificar código existente 3. ✅ **Testeable**: Cada servicio se puede probar aisladamente 4. ✅ **Organizado**: Responsabilidad única por servicio 5. ✅ **Discoverable**: Registro automático de servicios 6. ✅ **Reutilizable**: Componentes compartidos entre servicios 7. ✅ **Dinámico**: Agregar servicios en runtime ## 📊 Patrón Plugin Architecture Este diseño sigue el patrón **Plugin Architecture**: ``` MCPSQLServer │ ├─ ConnectionManager ├─ CredentialsManager ├─ Inspector ├─ Executor │ └─ Tools Registry │ ├─ Plugin 1 (ListServers) ├─ Plugin 2 (GetDatabases) ├─ Plugin 3 (GetTables) ├─ Plugin 4 (ExploreTable) ├─ Plugin 5 (DescribeTable) ├─ Plugin 6 (QueryColumns) └─ Plugin N (YourCustomTool) ``` Cada plugin: - Hereda de `MCPTool` - Implementa `name`, `description`, `execute` - Tiene acceso a todos los componentes - Se registra automáticamente ## 🧪 Testing de Servicios ```python import pytest from mcp_sql.tools.get_databases import GetDatabasesService from mcp_sql import ConnectionManager, CredentialsManager @pytest.fixture def service(): conn_mgr = ConnectionManager() creds_mgr = CredentialsManager() # ... inicializar componentes return GetDatabasesService( conn_mgr, creds_mgr, inspector, executor ) async def test_get_databases(service): result = await service.execute( ctx=mock_context, server_name="localhost", user="test", password="test" ) assert isinstance(result, list) assert len(result) > 0 ``` ## 📝 Mejores Prácticas 1. **Nombrado**: Usa sufijo `Service` en el nombre de la clase 2. **Documentación**: Documenta args y returns en `execute()` 3. **Validación**: Valida credenciales antes de ejecutar 4. **Errores**: Retorna diccionarios con `{"error": "..."}` en caso de fallo 5. **Logging**: Usa `print()` para mensajes importantes 6. **Límites**: Implementa límites en queries (max rows, timeout, etc.) 7. **Reutilización**: Usa componentes existentes en lugar de crear nuevos