#!/usr/bin/env python3
"""
MariaDB MCP Server - Test Script
Run this to verify your setup and connection
"""
import os
import sys
import mariadb
from datetime import datetime
def test_connection():
"""Test basic database connection"""
print("=" * 50)
print("MariaDB MCP Server - Connection Test")
print("=" * 50)
# Get configuration
config = {
'host': os.getenv('MARIADB_HOST', 'localhost'),
'port': int(os.getenv('MARIADB_PORT', '3306')),
'user': os.getenv('MARIADB_USER', ''),
'password': os.getenv('MARIADB_PASSWORD', ''),
'database': os.getenv('MARIADB_DATABASE', '')
}
print(f"\n📍 Configuration:")
print(f" Host: {config['host']}:{config['port']}")
print(f" User: {config['user']}")
print(f" Database: {config['database'] or '(none)'}")
print(f" Read-only mode: {os.getenv('MARIADB_READ_ONLY', 'true')}")
if not config['user']:
print("\n❌ Error: MARIADB_USER not set")
print(" Please set the environment variable:")
print(" export MARIADB_USER=your_username")
return False
print("\n🔄 Testing connection...")
try:
# Remove empty database from config
if not config['database']:
del config['database']
# Connect to MariaDB
conn = mariadb.connect(**config)
cursor = conn.cursor()
print("✅ Connection successful!")
# Get server info
cursor.execute("SELECT VERSION()")
version = cursor.fetchone()[0]
print(f"\n📊 Server Information:")
print(f" Version: {version}")
# Get current database
cursor.execute("SELECT DATABASE()")
current_db = cursor.fetchone()[0]
print(f" Current Database: {current_db or '(none)'}")
# List accessible databases
cursor.execute("SHOW DATABASES")
databases = [row[0] for row in cursor.fetchall()]
print(f"\n📚 Accessible Databases ({len(databases)}):")
for db in databases[:10]: # Show first 10
print(f" - {db}")
if len(databases) > 10:
print(f" ... and {len(databases) - 10} more")
# If a database is specified, show tables
if config.get('database'):
cursor.execute(f"USE `{config['database']}`")
cursor.execute("SHOW TABLES")
tables = [row[0] for row in cursor.fetchall()]
print(f"\n📋 Tables in '{config['database']}' ({len(tables)}):")
for table in tables[:10]: # Show first 10
print(f" - {table}")
if len(tables) > 10:
print(f" ... and {len(tables) - 10} more")
# Test query capabilities
print("\n🧪 Testing Query Capabilities:")
# Test SELECT
try:
cursor.execute("SELECT 1 + 1 AS result")
result = cursor.fetchone()[0]
print(f" ✅ SELECT queries: Working (1 + 1 = {result})")
except:
print(" ❌ SELECT queries: Failed")
# Test SHOW
try:
cursor.execute("SHOW VARIABLES LIKE 'max_connections'")
cursor.fetchone()
print(" ✅ SHOW queries: Working")
except:
print(" ❌ SHOW queries: Failed")
# Test write permissions (if not in read-only mode)
if os.getenv('MARIADB_READ_ONLY', 'true').lower() == 'false':
print("\n🔓 Testing Write Permissions:")
try:
# Try to create a temporary table
test_table = f"mcp_test_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
cursor.execute(f"CREATE TEMPORARY TABLE {test_table} (id INT)")
print(f" ✅ CREATE TABLE: Working")
cursor.execute(f"DROP TEMPORARY TABLE {test_table}")
print(f" ✅ DROP TABLE: Working")
except Exception as e:
print(f" ⚠️ Write operations: May be restricted ({str(e)[:50]}...)")
cursor.close()
conn.close()
print("\n" + "=" * 50)
print("✅ All tests passed! Your MCP server is ready to use.")
print("=" * 50)
return True
except mariadb.Error as e:
print(f"\n❌ Connection failed: {e}")
print("\n🔧 Troubleshooting tips:")
print(" 1. Check that MariaDB is running")
print(" 2. Verify your credentials")
print(" 3. Ensure the user has appropriate permissions")
print(" 4. Check firewall/network settings")
if "Can't connect to MySQL server" in str(e):
print("\n The server might not be running or accessible.")
print(" Try: sudo systemctl start mariadb")
elif "Access denied" in str(e):
print("\n The username or password might be incorrect.")
print(" Verify your MARIADB_USER and MARIADB_PASSWORD")
elif "Unknown database" in str(e):
print("\n The specified database doesn't exist.")
print(" Try without setting MARIADB_DATABASE")
return False
except Exception as e:
print(f"\n❌ Unexpected error: {e}")
return False
def test_mariadb_connector():
"""Test if MariaDB Connector/C is properly installed"""
print("\n🔍 Checking MariaDB Connector/C installation...")
try:
import mariadb
print("✅ Python mariadb module imported successfully")
# Check if we can access mariadb_config
mariadb_config = os.getenv('MARIADB_CONFIG')
if mariadb_config:
print(f"✅ MARIADB_CONFIG set to: {mariadb_config}")
else:
print("⚠️ MARIADB_CONFIG not set (might still work)")
return True
except ImportError as e:
print(f"❌ Failed to import mariadb: {e}")
print("\n🔧 Installation instructions:")
print("\nOn macOS:")
print(" brew install mariadb-connector-c")
print(" export MARIADB_CONFIG=$(brew --prefix mariadb-connector-c)/bin/mariadb_config")
print(" pip install mariadb")
print("\nOn Ubuntu/Debian:")
print(" sudo apt-get install libmariadb-dev")
print(" pip install mariadb")
print("\nOn RHEL/CentOS:")
print(" sudo yum install mariadb-connector-c-devel")
print(" pip install mariadb")
return False
def test_mcp_sdk():
"""Test if MCP SDK is installed"""
print("\n🔍 Checking MCP SDK installation...")
try:
import mcp
from mcp.server.fastmcp import FastMCP
print("✅ MCP SDK imported successfully")
return True
except ImportError as e:
print(f"❌ Failed to import MCP SDK: {e}")
print("\n🔧 To install:")
print(" pip install mcp")
return False
def main():
"""Run all tests"""
print("\n" + "🚀 MariaDB MCP Server - Setup Test" + "\n")
all_passed = True
# Test dependencies
if not test_mcp_sdk():
all_passed = False
if not test_mariadb_connector():
all_passed = False
print("\n⚠️ Cannot test connection without mariadb module")
else:
# Test connection
if not test_connection():
all_passed = False
# Final summary
print("\n" + "=" * 50)
if all_passed:
print("✅ SUCCESS: Your environment is ready!")
print("\nNext steps:")
print("1. Add the server to your Claude Desktop config")
print("2. Restart Claude Desktop")
print("3. Start using the MariaDB tools in Claude")
else:
print("⚠️ Some tests failed. Please fix the issues above.")
print("\nFor detailed setup instructions, see README.md")
print("=" * 50 + "\n")
return 0 if all_passed else 1
if __name__ == "__main__":
sys.exit(main())
