MySQL-MCP

一个强大的 MySQL 数据库连接器，专为 Claude 构建，基于模型上下文协议 (MCP)

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

特性

• 查询执行: 针对您的数据库运行任何 SQL 查询
• 查询解释: 获取详细的查询执行计划
• 模式探索: 查看数据库、表和列定义
• 数据检查: 使用示例行预览表内容
• 安全第一: 控制措施可防止意外数据丢失
• 最佳实践: 内置提示，帮助 Claude 编写最佳 SQL
• 完整日志: 所有操作的全面日志记录

安装

有几种方法可以安装 MySQL-MCP：

选项 1：使用 pip 从源代码安装

# 克隆存储库
git clone https://github.com/yourusername/mysql-mcp.git
cd mysql-mcp

# 创建一个虚拟环境（推荐）
python -m venv venv
source venv/bin/activate  # 在 Windows 上：venv\Scripts\activate

# 安装依赖
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 包：

# 从本地副本
fastmcp install mysql_mcp.py

# 或直接从 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_dotenv()

# 初始化 FastMCP 应用
mcp = FastMCP(
    "MySQL MCP",
    description="MySQL database connector for Claude",
    dependencies=["mysql-connector-python", "python-dotenv"]
)

# 数据库连接配置
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"))

# 全局连接状态
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 一起运行

1. 如果您尚未安装，请在 Claude 中安装您的 MySQL MCP 应用：

fastmcp install mysql_mcp.py

2. 在 Claude 中，从工具选择器中选择 MySQL MCP 工具

3. 现在您可以要求 Claude：

   • "显示所有可用的数据库"
   • "customers 数据库中有哪些表？"
   • "查询过去一周内的所有订单"
   • "显示 users 表的模式"
   • "帮助我优化这个慢查询：SELECT * FROM orders JOIN users ON user_id WHERE status = 'pending'"

本地运行

您也可以在本地运行 MCP 服务器进行开发：

# 直接运行
python mysql_mcp.py

# 或使用 FastMCP 开发模式
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 应该有权访问的数据库的访问权限
• 考虑使用只读用户以确保安全
• 在处理生产数据时，在执行之前审查所有查询
• 根据最小权限原则限制连接权限

故障排除

如果您遇到问题：

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

贡献

欢迎贡献！请随时提交 Pull Request。

许可证

MIT

---

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