MariaDB MCP Server

Instructions

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()