feat: add charset and collation settings for compatibility with older MySQL versions (#56)

Co-authored-by: azrilaiman03 <[email protected]>

Author: azrilaiman2003

README.md

@@ -33,6 +33,26 @@ MYSQL_PORT=3306         # Optional: Database port (defaults to 3306 if not speci
 MYSQL_USER=your_username
 MYSQL_PASSWORD=your_password
 MYSQL_DATABASE=your_database
+
+# Optional: Charset and collation settings for compatibility with older MySQL versions
+MYSQL_CHARSET=utf8mb4           # Optional: Character set (defaults to utf8mb4)
+MYSQL_COLLATION=utf8mb4_unicode_ci  # Optional: Collation (defaults to utf8mb4_unicode_ci)
+MYSQL_SQL_MODE=TRADITIONAL      # Optional: SQL mode (defaults to TRADITIONAL)
+```
+
+### Troubleshooting Collation Issues
+If you encounter the error "Unknown collation: 'utf8mb4_0900_ai_ci'", this typically means you're connecting to an older MySQL version (5.7 or earlier) that doesn't support the newer collation. The server now automatically uses compatible settings, but you can override them:
+
+For MySQL 5.7 and earlier:
+```bash
+MYSQL_CHARSET=utf8mb4
+MYSQL_COLLATION=utf8mb4_unicode_ci
+```
+
+For very old MySQL versions (5.6 and earlier):
+```bash
+MYSQL_CHARSET=utf8
+MYSQL_COLLATION=utf8_unicode_ci
 ```
 
 ## Usage
@@ -44,7 +64,7 @@ Add this to your `claude_desktop_config.json`:
     "mysql": {
       "command": "uv",
       "args": [
-        "--directory", 
+        "--directory",
         "path/to/mysql_mcp_server",
         "run",
         "mysql_mcp_server"
@@ -55,6 +75,9 @@ Add this to your `claude_desktop_config.json`:
         "MYSQL_USER": "your_username",
         "MYSQL_PASSWORD": "your_password",
         "MYSQL_DATABASE": "your_database"
+        // Optional: Add these if you encounter collation issues
+        // "MYSQL_CHARSET": "utf8mb4",
+        // "MYSQL_COLLATION": "utf8mb4_unicode_ci"
       }
     }
   }
@@ -80,6 +103,9 @@ Add this to your `mcp.json`:
         "MYSQL_USER": "your_username",
         "MYSQL_PASSWORD": "your_password",
         "MYSQL_DATABASE": "your_database"
+        // Optional: Add these if you encounter collation issues
+        // "MYSQL_CHARSET": "utf8mb4",
+        // "MYSQL_COLLATION": "utf8mb4_unicode_ci"
       }
   }
 }

src/mysql_mcp_server/server.py

@@ -21,14 +21,25 @@ def get_db_config():
         "port": int(os.getenv("MYSQL_PORT", "3306")),
         "user": os.getenv("MYSQL_USER"),
         "password": os.getenv("MYSQL_PASSWORD"),
-        "database": os.getenv("MYSQL_DATABASE")
+        "database": os.getenv("MYSQL_DATABASE"),
+        # Add charset and collation to avoid utf8mb4_0900_ai_ci issues with older MySQL versions
+        # These can be overridden via environment variables for specific MySQL versions
+        "charset": os.getenv("MYSQL_CHARSET", "utf8mb4"),
+        "collation": os.getenv("MYSQL_COLLATION", "utf8mb4_unicode_ci"),
+        # Disable autocommit for better transaction control
+        "autocommit": True,
+        # Set SQL mode for better compatibility - can be overridden
+        "sql_mode": os.getenv("MYSQL_SQL_MODE", "TRADITIONAL")
     }
-    
-    if not all([config["user"], config["password"], config["database"]]):
+
+    # Remove None values to let MySQL connector use defaults if not specified
+    config = {k: v for k, v in config.items() if v is not None}
+
+    if not all([config.get("user"), config.get("password"), config.get("database")]):
         logger.error("Missing required database configuration. Please check environment variables:")
         logger.error("MYSQL_USER, MYSQL_PASSWORD, and MYSQL_DATABASE are required")
         raise ValueError("Missing required database configuration")
-    
+
     return config
 
 # Initialize server
@@ -39,12 +50,14 @@ async def list_resources() -> list[Resource]:
     """List MySQL tables as resources."""
     config = get_db_config()
     try:
+        logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}")
         with connect(**config) as conn:
+            logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}")
             with conn.cursor() as cursor:
                 cursor.execute("SHOW TABLES")
                 tables = cursor.fetchall()
                 logger.info(f"Found tables: {tables}")
-                
+
                 resources = []
                 for table in tables:
                     resources.append(
@@ -58,6 +71,7 @@ async def list_resources() -> list[Resource]:
                 return resources
     except Error as e:
         logger.error(f"Failed to list resources: {str(e)}")
+        logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")
         return []
 
 @app.read_resource()
@@ -66,24 +80,27 @@ async def read_resource(uri: AnyUrl) -> str:
     config = get_db_config()
     uri_str = str(uri)
     logger.info(f"Reading resource: {uri_str}")
-    
+
     if not uri_str.startswith("mysql://"):
         raise ValueError(f"Invalid URI scheme: {uri_str}")
-        
+
     parts = uri_str[8:].split('/')
     table = parts[0]
-    
+
     try:
+        logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}")
         with connect(**config) as conn:
+            logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}")
             with conn.cursor() as cursor:
                 cursor.execute(f"SELECT * FROM {table} LIMIT 100")
                 columns = [desc[0] for desc in cursor.description]
                 rows = cursor.fetchall()
                 result = [",".join(map(str, row)) for row in rows]
                 return "\n".join([",".join(columns)] + result)
-                
+
     except Error as e:
         logger.error(f"Database error reading resource {uri}: {str(e)}")
+        logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")
         raise RuntimeError(f"Database error: {str(e)}")
 
 @app.list_tools()
@@ -112,26 +129,28 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
     """Execute SQL commands."""
     config = get_db_config()
     logger.info(f"Calling tool: {name} with arguments: {arguments}")
-    
+
     if name != "execute_sql":
         raise ValueError(f"Unknown tool: {name}")
-    
+
     query = arguments.get("query")
     if not query:
         raise ValueError("Query is required")
-    
+
     try:
+        logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}")
         with connect(**config) as conn:
+            logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}")
             with conn.cursor() as cursor:
                 cursor.execute(query)
-                
+
                 # Special handling for SHOW TABLES
                 if query.strip().upper().startswith("SHOW TABLES"):
                     tables = cursor.fetchall()
                     result = ["Tables_in_" + config["database"]]  # Header
                     result.extend([table[0] for table in tables])
                     return [TextContent(type="text", text="\n".join(result))]
-                
+
                 # Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.)
                 elif cursor.description is not None:
                     columns = [desc[0] for desc in cursor.description]
@@ -142,31 +161,32 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
                     except Error as e:
                         logger.warning(f"Error fetching results: {str(e)}")
                         return [TextContent(type="text", text=f"Query executed but error fetching results: {str(e)}")]
-                
+
                 # Non-SELECT queries
                 else:
                     conn.commit()
                     return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")]
-                
+
     except Error as e:
         logger.error(f"Error executing SQL '{query}': {e}")
+        logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")
         return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
 
 async def main():
     """Main entry point to run the MCP server."""
     from mcp.server.stdio import stdio_server
-    
+
     # Add additional debug output
     print("Starting MySQL MCP server with config:", file=sys.stderr)
     config = get_db_config()
     print(f"Host: {config['host']}", file=sys.stderr)
     print(f"Port: {config['port']}", file=sys.stderr)
     print(f"User: {config['user']}", file=sys.stderr)
     print(f"Database: {config['database']}", file=sys.stderr)
-    
+
     logger.info("Starting MySQL MCP server...")
     logger.info(f"Database config: {config['host']}/{config['database']} as {config['user']}")
-    
+
     async with stdio_server() as (read_stream, write_stream):
         try:
             await app.run(