MySQL-MCP

モデルコンテキストプロトコル（MCP）で構築された、Claude 用の強力な MySQL データベースコネクタ

MySQL-MCPは、モデルコンテキストプロトコル（MCP）を介してClaudeがMySQLデータベースに直接アクセスできるようにします。シンプルな自然言語による指示で、データベースのクエリ、変更、探索を行うための包括的なツールセットを提供します。

特徴

クエリ実行: データベースに対して任意のSQLクエリを実行します
クエリの説明: 詳細なクエリ実行プランを取得する
スキーマ探索: データベース、テーブル、列定義を表示する
データ検査: サンプル行で表の内容をプレビューする
安全第一：偶発的なデータ損失を防ぐための制御
ベストプラクティス: クロードが最適なSQLを書くのに役立つ組み込みプロンプト
完全なログ記録: すべての操作の包括的なログ記録

インストール

MySQL-MCP をインストールするにはいくつかの方法があります。

オプション1: pipを使ってソースからインストールする

# Clone the repository
git clone https://github.com/yourusername/mysql-mcp.git
cd mysql-mcp

# Create a virtual environment (recommended)
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install requirements
pip install -r requirements.txt

オプション2: 開発パッケージとしてインストールする

git clone https://github.com/yourusername/mysql-mcp.git
cd mysql-mcp
pip install -e .

オプション3: FastMCPでインストールする（Claude統合用）

まず、FastMCP をインストールします。

pip install fastmcp

次に、MySQL-MCP パッケージをインストールします。

# From a local copy
fastmcp install mysql_mcp.py

# Or directly from GitHub
fastmcp install https://github.com/yourusername/mysql-mcp/mysql_mcp.py

クイックスタート

# mysql_mcp.py
from fastmcp import FastMCP
import mysql.connector
from mysql.connector import Error
import os
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any

# Load environment variables
load_dotenv()

# Initialize FastMCP app
mcp = FastMCP(
    "MySQL MCP",
    description="MySQL database connector for Claude",
    dependencies=["mysql-connector-python", "python-dotenv"]
)

# Database connection configuration
class DBConfig(BaseModel):
    host: str = Field(default=os.getenv("MYSQL_HOST", "localhost"))
    port: int = Field(default=int(os.getenv("MYSQL_PORT", "3306")))
    user: str = Field(default=os.getenv("MYSQL_USER", "root"))
    password: str = Field(default=os.getenv("MYSQL_PASSWORD", ""))
    database: Optional[str] = Field(default=os.getenv("MYSQL_DATABASE"))

# Global connection state
current_db = os.getenv("MYSQL_DATABASE", "")
config = DBConfig()

def get_connection():
    """Create a MySQL connection using the current configuration"""
    try:
        conn = mysql.connector.connect(
            host=config.host,
            port=config.port,
            user=config.user,
            password=config.password,
            database=config.database if config.database else None
        )
        return conn
    except Error as e:
        raise Exception(f"Database connection error: {e}")

@mcp.tool()
def query_sql(query: str) -> Dict[str, Any]:
    """Execute a SELECT query and return the results"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute(query)
        results = cursor.fetchall()
        return {
            "rows": results[:100],  # Limit to 100 rows for safety
            "row_count": cursor.rowcount,
            "column_names": [desc[0] for desc in cursor.description] if cursor.description else []
        }
    except Error as e:
        raise Exception(f"Query error: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def execute_sql(query: str) -> Dict[str, Any]:
    """Execute a non-SELECT query (INSERT, UPDATE, DELETE, etc.)"""
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute(query)
        conn.commit()
        return {
            "affected_rows": cursor.rowcount,
            "last_insert_id": cursor.lastrowid if cursor.lastrowid else None
        }
    except Error as e:
        conn.rollback()
        raise Exception(f"Query error: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def explain_sql(query: str) -> Dict[str, Any]:
    """Get the execution plan for a query"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute(f"EXPLAIN {query}")
        results = cursor.fetchall()
        return {
            "plan": results
        }
    except Error as e:
        raise Exception(f"EXPLAIN error: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def show_databases() -> Dict[str, Any]:
    """List all available databases"""
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("SHOW DATABASES")
        results = cursor.fetchall()
        return {
            "databases": [db[0] for db in results]
        }
    except Error as e:
        raise Exception(f"Error listing databases: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def use_database(database: str) -> Dict[str, Any]:
    """Switch to a different database"""
    global config, current_db
    # Verify database exists
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("SHOW DATABASES")
        dbs = [db[0] for db in cursor.fetchall()]
        if database not in dbs:
            raise ValueError(f"Database '{database}' does not exist")
        
        # Update configuration
        config.database = database
        current_db = database
        return {
            "current_database": database,
            "status": "success"
        }
    except Error as e:
        raise Exception(f"Error changing database: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def show_tables() -> Dict[str, Any]:
    """List all tables in the current database"""
    if not config.database:
        raise ValueError("No database selected. Use 'use_database' first.")
    
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("SHOW TABLES")
        results = cursor.fetchall()
        return {
            "database": config.database,
            "tables": [table[0] for table in results]
        }
    except Error as e:
        raise Exception(f"Error listing tables: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.tool()
def describe_table(table: str) -> Dict[str, Any]:
    """Get column definitions for a table"""
    if not config.database:
        raise ValueError("No database selected. Use 'use_database' first.")
    
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute(f"DESCRIBE {table}")
        columns = cursor.fetchall()
        
        # Get index information
        cursor.execute(f"SHOW INDEX FROM {table}")
        indexes = cursor.fetchall()
        
        return {
            "table": table,
            "columns": columns,
            "indexes": indexes
        }
    except Error as e:
        raise Exception(f"Error describing table: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.resource(f"schema://{'{database}'}")
def get_database_schema(database: Optional[str] = None) -> str:
    """Get the full schema of a database as a resource"""
    db_to_use = database or config.database
    if not db_to_use:
        raise ValueError("No database specified or selected")
    
    conn = get_connection()
    cursor = conn.cursor()
    schema = []
    
    try:
        # Switch to the specified database
        cursor.execute(f"USE {db_to_use}")
        
        # Get all tables
        cursor.execute("SHOW TABLES")
        tables = [table[0] for table in cursor.fetchall()]
        
        # Get CREATE TABLE statements for each table
        for table in tables:
            cursor.execute(f"SHOW CREATE TABLE {table}")
            create_stmt = cursor.fetchone()[1]
            schema.append(create_stmt)
        
        return "\n\n".join(schema)
    except Error as e:
        raise Exception(f"Error getting schema: {e}")
    finally:
        cursor.close()
        conn.close()

@mcp.prompt()
def write_query_for_task(task: str) -> str:
    """Help Claude write an optimal SQL query for a given task"""
    return f"""Task: {task}

Please write an SQL query that accomplishes this task efficiently.

Some guidelines:
1. Use appropriate JOINs (INNER, LEFT, RIGHT) based on the data relationships
2. Filter data in the WHERE clause to minimize data processing
3. Consider using indexes for better performance
4. Use appropriate aggregation functions when needed
5. Format the query with clear indentation for readability

If you need to see the database schema first, you can access it using the schema:// resource.
"""

@mcp.prompt()
def analyze_query_performance(query: str) -> str:
    """Help Claude analyze the performance of a query"""
    return f"""Query: {query}

Please analyze this query for performance issues:

1. First, use the explain_sql tool to get the execution plan
2. Look for table scans instead of index usage
3. Check if the joins are efficient
4. Identify if the query can be optimized with better indexes
5. Suggest concrete improvements to make the query more efficient
"""

if __name__ == "__main__":
    # Run the server directly
    mcp.run()

環境設定

MySQL 接続の詳細を含む.envファイルを作成します。

MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=your_password
MYSQL_DATABASE=initial_database

クロードと走る

まだインストールしていない場合は、Claude に MySQL MCP アプリをインストールします。

fastmcp install mysql_mcp.py

Claudeでは、ツールセレクターからMySQL MCPツールを選択します。
Claude に次のことを依頼できます。
- 「利用可能なすべてのデータベースを表示」
- 「顧客データベースにはどのようなテーブルがありますか？」
- 「先週行われたすべての注文を照会する」
- 「ユーザーテーブルのスキーマを表示してください」
- 「この遅いクエリの最適化を手伝ってください: SELECT * FROM orders JOIN users ON user_id WHERE status = 'pending'」

ローカルで実行

開発のために MCP サーバーをローカルで実行することもできます。

# Run directly
python mysql_mcp.py

# Or use FastMCP development mode
fastmcp dev mysql_mcp.py

高度な使用法

複数のデータベースへの接続

複数のデータベースの構成を作成し、それらを切り替えることができます。

@mcp.tool()
def save_connection(name: str, host: str, port: int, user: str, password: str, database: Optional[str] = None) -> Dict[str, Any]:
    """Save a named database connection configuration"""
    # Implementation details...

トランザクションサポート

アトミックである必要がある操作の場合:

@mcp.tool()
def run_transaction(queries: List[str]) -> Dict[str, Any]:
    """Run multiple queries in a single transaction"""
    # Implementation details...

スキーマ分析

データベース構造に関するより深い洞察を提供します。

@mcp.tool()
def analyze_table_relationships() -> Dict[str, Any]:
    """Analyze foreign key relationships between tables"""
    # Implementation details...

セキュリティに関する考慮事項

MySQL-MCPはデータベースに対して直接SQLを実行します
クロードがアクセスする必要があるデータベースへのアクセスのみを許可する
安全のため、読み取り専用ユーザーの使用を検討してください
本番データを扱うときは、実行前にすべてのクエリを確認してください。
最小権限の原則に基づいて接続権限を制限する

トラブルシューティング

問題が発生した場合:

.envファイルで正しいデータベース資格情報を確認してください
MySQLサーバーが実行中でアクセス可能であることを確認する
詳細なエラー情報については、ログファイル（ mysql_mcp.log ）を参照してください。
すべての依存関係がインストールされていることを確認します: pip install -r requirements.txt

貢献

貢献を歓迎します！お気軽にプルリクエストを送信してください。

ライセンス

マサチューセッツ工科大学

モデルコンテキストプロトコルサーバーを高速に Python で構築する方法であるFastMCPを使用して構築されています。