execute_sql
Executes read-only SQL queries to retrieve data from MariaDB databases. Supports database exploration and schema inspection with secure connection pooling and logging.
Instructions
Execute a read-only SQL query and return results.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database | No | ||
| query | Yes |
Implementation Reference
- src/mariadb_mcp/server.py:282-329 (handler)Main handler for 'execute_sql' MCP tool. Registers via @mcp.tool(), validates read-only queries, executes via helper, formats results as 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)Core helper method in MariaDBConnection that handles actual DB query execution and result fetching as dicts. Called by execute_sql.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 execute_sql as an MCP tool.@mcp.tool()
- src/mariadb_mcp/server.py:283-283 (schema)Function signature defines input schema (query: str, optional database: str) and output str.async def execute_sql(query: str, database: Optional[str] = None) -> str: