diff --git a/README.md b/README.md index f517802..b5c3ab1 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,7 @@ npx -y @smithery/cli install mysql-mcp-server --client claude ``` ## Configuration +### Environment Variables Set the following environment variables: ```bash MYSQL_HOST=localhost # Database host @@ -36,6 +37,27 @@ MYSQL_PASSWORD=your_password MYSQL_DATABASE=your_database ``` +### Command Line Arguments +Alternatively, you can provide database configuration via command line arguments, which take precedence over environment variables: + +```bash +python server.py [options] + +Options: + -h, --host HOST MySQL host (default: localhost) + -p, --port PORT MySQL port (default: 3306) + -u, --user USER MySQL username (required) + -P, --password PASS MySQL password (required) + -d, --database DB MySQL database name (required) + --charset CHARSET MySQL charset (default: utf8mb4) + --collation COLLATION MySQL collation (default: utf8mb4_unicode_ci) +``` + +Example: +```bash +python server.py -h localhost -p 3306 -u myuser -P mypassword -d mydatabase +``` + ## Usage ### With Claude Desktop Add this to your `claude_desktop_config.json`: diff --git a/src/mysql_mcp_server/server.py b/src/mysql_mcp_server/server.py index 15fcdbd..ca03163 100644 --- a/src/mysql_mcp_server/server.py +++ b/src/mysql_mcp_server/server.py @@ -1,4 +1,5 @@ import asyncio +import getopt import logging import os import sys @@ -15,33 +16,56 @@ logger = logging.getLogger("mysql_mcp_server") def get_db_config(): - """Get database configuration from environment variables.""" + """Get database configuration from command line arguments or environment variables.""" + # Parse command line arguments + try: + opts, args = getopt.getopt(sys.argv[1:], "h:p:u:d:P:", + ["host=", "port=", "user=", "database=", "password=", "charset=", "collation="]) + except getopt.GetoptError as err: + logger.error(f"Command line error: {err}") + sys.exit(2) + + # Initialize config with environment variables as defaults config = { "host": os.getenv("MYSQL_HOST", "localhost"), "port": int(os.getenv("MYSQL_PORT", "3306")), "user": os.getenv("MYSQL_USER"), "password": os.getenv("MYSQL_PASSWORD"), "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") } + + # Override with command line arguments + for opt, arg in opts: + if opt in ("-h", "--host"): + config["host"] = arg + elif opt in ("-p", "--port"): + config["port"] = int(arg) + elif opt in ("-u", "--user"): + config["user"] = arg + elif opt in ("-P", "--password"): + config["password"] = arg + elif opt in ("-d", "--database"): + config["database"] = arg + elif opt == "--charset": + config["charset"] = arg + elif opt == "--collation": + config["collation"] = arg # 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") + logger.error("Missing required database configuration. Please provide via command line or environment variables:") + logger.error("Required: user (-u), password (-P), database (-d)") raise ValueError("Missing required database configuration") return config + # Initialize server app = Server("mysql_mcp_server") @@ -52,7 +76,7 @@ async def list_resources() -> list[Resource]: 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()}") + logger.info(f"Successfully connected to MySQL server version: {conn.server_info}") with conn.cursor() as cursor: cursor.execute("SHOW TABLES") tables = cursor.fetchall() @@ -90,7 +114,7 @@ async def read_resource(uri: AnyUrl) -> str: 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()}") + logger.info(f"Successfully connected to MySQL server version: {conn.server_info}") with conn.cursor() as cursor: cursor.execute(f"SELECT * FROM {table} LIMIT 100") columns = [desc[0] for desc in cursor.description] @@ -140,7 +164,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: 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()}") + logger.info(f"Successfully connected to MySQL server version: {conn.server_info}") with conn.cursor() as cursor: cursor.execute(query) diff --git a/tests/conftest.py b/tests/conftest.py index 868acb5..befc4b0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,9 +1,19 @@ # tests/conftest.py import pytest import os +import sys import mysql.connector from mysql.connector import Error +@pytest.fixture(scope="session", autouse=True) +def mock_argv(): + """Mock sys.argv to prevent getopt from parsing test runner arguments.""" + original_argv = sys.argv.copy() + # Set minimal argv to avoid getopt parsing issues during tests + sys.argv = ['server.py'] + yield + sys.argv = original_argv + @pytest.fixture(scope="session") def mysql_connection(): """Create a test database connection.""" @@ -43,4 +53,20 @@ def mysql_cursor(mysql_connection): """Create a test cursor.""" cursor = mysql_connection.cursor() yield cursor - cursor.close() \ No newline at end of file + cursor.close() + +@pytest.fixture +def mock_argv_with_db_args(): + """Mock sys.argv with database connection arguments for testing.""" + original_argv = sys.argv.copy() + # Set argv with database arguments + sys.argv = [ + 'server.py', + '-h', os.getenv("MYSQL_HOST", "127.0.0.1"), + '-p', os.getenv("MYSQL_PORT", "3306"), + '-u', os.getenv("MYSQL_USER", "root"), + '-P', os.getenv("MYSQL_PASSWORD", "testpassword"), + '-d', os.getenv("MYSQL_DATABASE", "test_db") + ] + yield sys.argv + sys.argv = original_argv \ No newline at end of file diff --git a/tests/test_server.py b/tests/test_server.py index 3247468..5e2d589 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,5 +1,6 @@ import pytest -from mysql_mcp_server.server import app, list_tools, list_resources, read_resource, call_tool +import sys +from mysql_mcp_server.server import app, list_tools, list_resources, read_resource, call_tool, get_db_config from pydantic import AnyUrl def test_server_initialization(): @@ -43,4 +44,47 @@ async def test_list_resources(): except ValueError as e: if "Missing required database configuration" in str(e): pytest.skip("Database configuration not available") - raise \ No newline at end of file + raise + +def test_get_db_config_with_command_line_args(mock_argv_with_db_args): + """Test that get_db_config works with command line arguments.""" + config = get_db_config() + + # Verify that config contains expected values from command line args + assert config["host"] in ["127.0.0.1", "localhost"] + assert config["port"] == 3306 + assert config["user"] in ["root", "testuser"] + assert "password" in config + assert "database" in config + assert config["charset"] == "utf8mb4" + assert config["collation"] == "utf8mb4_unicode_ci" + +@pytest.mark.asyncio +async def test_mysql_connection_with_args(mock_argv_with_db_args): + """Test MySQL connection using command line arguments.""" + try: + import mysql.connector + from mysql.connector import Error + + config = get_db_config() + + # Test the connection with the parsed config + connection = mysql.connector.connect(**config) + assert connection.is_connected() + + # Test a simple query + cursor = connection.cursor() + cursor.execute("SELECT 1 as test") + result = cursor.fetchone() + assert result[0] == 1 + + cursor.close() + connection.close() + + except ImportError: + pytest.skip("mysql-connector-python not available") + except Error as e: + if "Access denied" in str(e) or "Unknown database" in str(e): + pytest.skip(f"Database not configured for testing: {e}") + else: + raise \ No newline at end of file