MySQL-MCP

使用模型上下文协议 (MCP) 为 Claude 构建的强大的 MySQL 数据库连接器

MySQL-MCP 允许 Claude 通过模型上下文协议直接访问您的 MySQL 数据库。它提供了一整套工具，可以通过简单的自然语言指令来查询、修改和探索您的数据库。

特征

查询执行：对数据库运行任何 SQL 查询
查询解释：获取详细的查询执行计划
模式探索：查看数据库、表和列定义
数据检查：使用样本行预览表格内容
安全第一：控制以防止意外数据丢失
最佳实践：内置提示，帮助 Claude 编写最佳 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

与克劳德一起奔跑

如果您还没有安装 MySQL MCP 应用，请在 Claude 中安装它：

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
仅提供 Claude 有权访问的数据库的访问权限
考虑使用只读用户以确保安全
处理生产数据时，在执行之前检查所有查询
根据最小特权原则限制连接权限

故障排除

如果您遇到问题：

检查你的.env文件是否有正确的数据库凭证
验证 MySQL 服务器是否正在运行且可访问
查看日志文件（ mysql_mcp.log ）以获取详细的错误信息
确保所有依赖项都已安装： pip install -r requirements.txt

贡献

欢迎贡献代码！欢迎提交 Pull 请求。

执照

麻省理工学院

使用FastMCP构建，这是构建模型上下文协议服务器的快速、Pythonic 方式。