execute_sql
Execute read-only SQL queries on MariaDB databases to retrieve data, explore schemas, and inspect database structures securely.
Instructions
Execute a read-only SQL query and return results.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| database | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/mariadb_mcp/server.py:282-330 (handler)The primary handler for the 'execute_sql' tool, registered via @mcp.tool() decorator. Performs security validation for read-only queries, executes the SQL using MariaDBConnection, and returns formatted results as a markdown table.
@mcp.tool() async def execute_sql(query: str, database: Optional[str] = None) -> str: """Execute a read-only SQL query and return results.""" try: # Security check: only allow SELECT, SHOW, DESCRIBE, EXPLAIN queries query_upper = query.strip().upper() allowed_keywords = ['SELECT', 'SHOW', 'DESCRIBE', 'DESC', 'EXPLAIN', 'WITH'] if not any(query_upper.startswith(keyword) for keyword in allowed_keywords): return "Error: Only read-only queries (SELECT, SHOW, DESCRIBE, EXPLAIN) are allowed" # Switch database if specified if database: await db_connection.execute_query(f"USE `{database}`") results = await db_connection.execute_query(query) if not results: return "Query executed successfully. No results returned." # Format results as a table if len(results) == 0: return "No rows returned" # Get column names columns = list(results[0].keys()) # Create table header output = "Query Results:\n\n" header = " | ".join(columns) separator = " | ".join(["-" * len(col) for col in columns]) output += header + "\n" + separator + "\n" # Add data rows (limit to 100 rows for readability) for i, row in enumerate(results[:100]): row_data = " | ".join([str(row.get(col, '')) for col in columns]) output += row_data + "\n" if len(results) > 100: output += f"\n... and {len(results) - 100} more rows (truncated for display)" output += f"\nTotal rows: {len(results)}" return output except Exception as e: logger.error(f"Error executing SQL: {e}") return f"Error executing SQL: {str(e)}" - src/mariadb_mcp/server.py:135-149 (helper)Helper method in MariaDBConnection class that actually executes SQL queries using aiomysql pool, used by the execute_sql tool.
async def execute_query(self, query: str, params: Optional[tuple] = None) -> List[Dict[str, Any]]: """Execute a SQL query and return results.""" logger.debug(f"Executing query: {query[:100]}{'...' if len(query) > 100 else ''}") await self.connect() async with self.pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cursor: await cursor.execute(query, params) if cursor.description: results = await cursor.fetchall() logger.debug(f"Query returned {len(results)} rows") return [dict(row) for row in results] logger.debug("Query executed successfully, no results returned") return [] - src/mariadb_mcp/server.py:282-282 (registration)The @mcp.tool() decorator registers the execute_sql function as an MCP tool in the FastMCP server.
@mcp.tool()