Add EXPLAIN and EXPLAIN EXTENDED query analysis tools

README.md

@@ -24,6 +24,8 @@ The MCP MariaDB Server exposes a set of tools for interacting with MariaDB datab
 - Listing databases and tables
 - Retrieving table schemas
 - Executing safe, read-only SQL queries
+- Query performance analysis with EXPLAIN and EXPLAIN EXTENDED
+- Comprehensive tool usage guide for LLM self-discovery
 - Creating and managing vector stores for embedding-based search
 - Integrating with embedding providers (currently OpenAI, Gemini, and HuggingFace) (optional)
 
@@ -63,6 +65,18 @@ The MCP MariaDB Server exposes a set of tools for interacting with MariaDB datab
   - Creates a new database if it doesn't exist.
   - Parameters: `database_name` (string, required)  
 
+### Query Performance Analysis Tools
+
+- **explain_query**
+  - Executes EXPLAIN on a SQL query to show the execution plan for performance analysis.
+  - Parameters: `sql_query` (string, required), `database_name` (string, required), `parameters` (list, optional)
+  - _Note: Helps analyze query performance and optimization opportunities. Does not execute the actual query._
+
+- **explain_query_extended**
+  - Executes EXPLAIN EXTENDED on a SQL query to show detailed execution plan with additional information.
+  - Parameters: `sql_query` (string, required), `database_name` (string, required), `parameters` (list, optional)
+  - _Note: Provides comprehensive analysis including filtered rows percentage and extra optimization details._
+
 ### Vector Store & Embedding Tools (optional)
 
 **Note**: These tools are only available when `EMBEDDING_PROVIDER` is configured. If no embedding provider is set, these tools will be disabled.

src/server.py

@@ -112,7 +112,7 @@ async def _execute_query(self, sql: str, params: Optional[tuple] = None, databas
             logger.error("Connection pool is not initialized.")
             raise RuntimeError("Database connection pool not available.")
 
-        allowed_prefixes = ('SELECT', 'SHOW', 'DESC', 'DESCRIBE', 'USE')
+        allowed_prefixes = ('SELECT', 'SHOW', 'DESC', 'DESCRIBE', 'USE', 'EXPLAIN')
         query_upper = sql.strip().upper()
         is_allowed_read_query = any(query_upper.startswith(prefix) for prefix in allowed_prefixes)
 
@@ -359,6 +359,52 @@ async def create_database(self, database_name: str) -> Dict[str, Any]:
             logger.error(f"TOOL ERROR: create_database. {error_message} Error: {e}", exc_info=True)
             raise RuntimeError(f"{error_message} Reason: {str(e)}")
 
+    async def explain_query(self, sql_query: str, database_name: str, parameters: Optional[List[Any]] = None) -> List[Dict[str, Any]]:
+        """
+        Executes EXPLAIN on a SQL query to show the execution plan.
+        This helps analyze query performance and optimization opportunities.
+        Example `parameters`: ["value1", 123] corresponding to %s placeholders in `sql_query`.
+        """
+        logger.info(f"TOOL START: explain_query called. database_name={database_name}, sql_query={sql_query[:100]}, parameters={parameters}")
+        if database_name and not database_name.isidentifier():
+            logger.warning(f"TOOL WARNING: explain_query called with invalid database_name: {database_name}")
+            raise ValueError(f"Invalid database name provided: {database_name}")
+
+        # Add EXPLAIN keyword to the query
+        explain_sql = f"EXPLAIN {sql_query.strip()}"
+        param_tuple = tuple(parameters) if parameters is not None else None
+
+        try:
+            results = await self._execute_query(explain_sql, params=param_tuple, database=database_name)
+            logger.info(f"TOOL END: explain_query completed. Execution plan rows returned: {len(results)}.")
+            return results
+        except Exception as e:
+            logger.error(f"TOOL ERROR: explain_query failed for database_name={database_name}, sql_query={sql_query[:100]}, parameters={parameters}: {e}", exc_info=True)
+            raise
+
+    async def explain_query_extended(self, sql_query: str, database_name: str, parameters: Optional[List[Any]] = None) -> List[Dict[str, Any]]:
+        """
+        Executes EXPLAIN EXTENDED on a SQL query to show detailed execution plan with additional information.
+        This provides more comprehensive analysis including filtered rows percentage and extra information.
+        Example `parameters`: ["value1", 123] corresponding to %s placeholders in `sql_query`.
+        """
+        logger.info(f"TOOL START: explain_query_extended called. database_name={database_name}, sql_query={sql_query[:100]}, parameters={parameters}")
+        if database_name and not database_name.isidentifier():
+            logger.warning(f"TOOL WARNING: explain_query_extended called with invalid database_name: {database_name}")
+            raise ValueError(f"Invalid database name provided: {database_name}")
+
+        # Add EXPLAIN EXTENDED keyword to the query
+        explain_sql = f"EXPLAIN EXTENDED {sql_query.strip()}"
+        param_tuple = tuple(parameters) if parameters is not None else None
+
+        try:
+            results = await self._execute_query(explain_sql, params=param_tuple, database=database_name)
+            logger.info(f"TOOL END: explain_query_extended completed. Extended execution plan rows returned: {len(results)}.")
+            return results
+        except Exception as e:
+            logger.error(f"TOOL ERROR: explain_query_extended failed for database_name={database_name}, sql_query={sql_query[:100]}, parameters={parameters}: {e}", exc_info=True)
+            raise
+
     async def create_vector_store_tool(self,
                                   database_name: str,
                                   vector_store_name: str,
@@ -701,6 +747,8 @@ def register_tools(self):
         self.mcp.add_tool(self.get_table_schema)
         self.mcp.add_tool(self.execute_sql)
         self.mcp.add_tool(self.create_database)
+        self.mcp.add_tool(self.explain_query)
+        self.mcp.add_tool(self.explain_query_extended)
         if EMBEDDING_PROVIDER is not None:
             self.mcp.add_tool(self.create_vector_store)
             self.mcp.add_tool(self.list_vector_stores)