MySQL-MCP

# MySQL-MCP **基于模型上下文协议(MCP)构建的强大MySQL数据库连接器**    MySQL-MCP通过模型上下文协议(Model Context Protocol)为Claude提供直接访问MySQL数据库的能力。它提供了一套完整的工具，可以通过简单的自然语言指令来查询、修改和探索数据库。 ## 功能特点 - **SQL查询执行**：对数据库执行任何SQL查询 - **查询计划解析**：获取详细的查询执行计划 - **数据库结构探索**：查看数据库、表和列定义 - **数据检查**：通过样本行预览表内容 - **安全优先**：内置控制措施防止意外数据丢失 - **最佳实践**：内置提示帮助Claude编写最优SQL - **完整日志**：所有操作的全面日志记录 ## 安装方法 MySQL-MCP提供了多种安装方式： ### 方式一：通过源码用pip安装 ```bash # 克隆仓库 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 ``` ### 方式二：作为开发包安装 ```bash git clone https://github.com/yourusername/mysql-mcp.git cd mysql-mcp pip install -e . ``` ### 方式三：使用FastMCP安装（用于Claude集成） 首先，安装FastMCP： ```bash pip install fastmcp ``` 然后安装MySQL-MCP包： ```bash # 从本地副本安装 fastmcp install mysql_mcp.py # 或直接从GitHub安装 fastmcp install https://github.com/yourusername/mysql-mcp/mysql_mcp.py ``` ## 快速开始 ```python # 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数据库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(): """创建MySQL连接，使用当前配置""" 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"数据库连接错误: {e}") @mcp.tool() def query_sql(query: str) -> Dict[str, Any]: """执行SELECT查询并返回结果""" conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(query) results = cursor.fetchall() return { "rows": results[:100], # 为安全起见限制100行 "row_count": cursor.rowcount, "column_names": [desc[0] for desc in cursor.description] if cursor.description else [] } except Error as e: raise Exception(f"查询错误: {e}") finally: cursor.close() conn.close() @mcp.tool() def execute_sql(query: str) -> Dict[str, Any]: """执行非SELECT查询(INSERT, UPDATE, DELETE等)""" 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"查询错误: {e}") finally: cursor.close() conn.close() @mcp.tool() def explain_sql(query: str) -> Dict[str, Any]: """获取查询的执行计划""" 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错误: {e}") finally: cursor.close() conn.close() @mcp.tool() def show_databases() -> Dict[str, Any]: """列出所有可用的数据库""" 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"列出数据库错误: {e}") finally: cursor.close() conn.close() @mcp.tool() def use_database(database: str) -> Dict[str, Any]: """切换到不同的数据库""" global config, current_db # 验证数据库是否存在 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}'不存在") # 更新配置 config.database = database current_db = database return { "current_database": database, "status": "success" } except Error as e: raise Exception(f"切换数据库错误: {e}") finally: cursor.close() conn.close() @mcp.tool() def show_tables() -> Dict[str, Any]: """列出当前数据库中的所有表""" if not config.database: raise ValueError("未选择数据库。请先使用'use_database'。") 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"列出表错误: {e}") finally: cursor.close() conn.close() @mcp.tool() def describe_table(table: str) -> Dict[str, Any]: """获取表的列定义""" if not config.database: raise ValueError("未选择数据库。请先使用'use_database'。") conn = get_connection() cursor = conn.cursor(dictionary=True) try: cursor.execute(f"DESCRIBE {table}") columns = cursor.fetchall() # 获取索引信息 cursor.execute(f"SHOW INDEX FROM {table}") indexes = cursor.fetchall() return { "table": table, "columns": columns, "indexes": indexes } except Error as e: raise Exception(f"描述表错误: {e}") finally: cursor.close() conn.close() @mcp.resource(f"schema://{'{database}'}") def get_database_schema(database: Optional[str] = None) -> str: """获取数据库的完整模式作为资源""" db_to_use = database or config.database if not db_to_use: raise ValueError("未指定或选择数据库") conn = get_connection() cursor = conn.cursor() schema = [] try: # 切换到指定数据库 cursor.execute(f"USE {db_to_use}") # 获取所有表 cursor.execute("SHOW TABLES") tables = [table[0] for table in cursor.fetchall()] # 获取每个表的CREATE 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"获取模式错误: {e}") finally: cursor.close() conn.close() @mcp.prompt() def write_query_for_task(task: str) -> str: """帮助Claude为给定任务编写最优SQL查询""" return f"""任务：{task} 请编写一个能够高效完成此任务的SQL查询。 一些指导原则： 1. 根据数据关系使用适当的连接(INNER, LEFT, RIGHT) 2. 在WHERE子句中过滤数据以减少数据处理 3. 考虑使用索引以提高性能 4. 在需要时使用适当的聚合函数 5. 使用清晰的缩进格式化查询，提高可读性 如果您需要先查看数据库模式，可以使用schema://资源访问。 """ @mcp.prompt() def analyze_query_performance(query: str) -> str: """帮助Claude分析查询性能""" return f"""查询：{query} 请分析此查询的性能问题： 1. 首先使用explain_sql工具获取执行计划 2. 查找表扫描而非索引使用的情况 3. 检查连接是否高效 4. 确定查询是否可以通过更好的索引进行优化 5. 提出具体的改进建议，使查询更高效 """ if __name__ == "__main__": # 直接运行服务器 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应用： ```bash 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服务器进行开发： ```bash # 直接运行 python mysql_mcp.py # 或使用FastMCP开发模式 fastmcp dev mysql_mcp.py ``` ## 高级用法 ### 连接到多个数据库 您可以为多个数据库创建配置并在它们之间切换： ```python @mcp.tool() def save_connection(name: str, host: str, port: int, user: str, password: str, database: Optional[str] = None) -> Dict[str, Any]: """保存命名的数据库连接配置""" # 实现细节... ``` ### 事务支持 对需要原子操作的场景： ```python @mcp.tool() def run_transaction(queries: List[str]) -> Dict[str, Any]: """在单个事务中运行多个查询""" # 实现细节... ``` ### 架构分析 提供关于数据库结构的更深入见解： ```python @mcp.tool() def analyze_table_relationships() -> Dict[str, Any]: """分析表之间的外键关系""" # 实现细节... ``` ## 安全考虑 - MySQL-MCP直接对您的数据库执行SQL - 仅提供Claude应该可以访问的数据库访问权限 - 考虑使用只读用户以确保安全 - 在处理生产数据时，执行前审查所有查询 - 基于最小权限原则限制连接权限 ## 故障排除 如果遇到问题： 1. 检查您的`.env`文件中的数据库凭据是否正确 2. 验证MySQL服务器是否正在运行并可访问 3. 查看日志文件(`mysql_mcp_zh.log`)以获取详细的错误信息 4. 确保所有依赖项都已安装：`pip install -r requirements.txt` ## 贡献 欢迎贡献！请随时提交Pull Request。 ## 许可证 MIT --- 使用[FastMCP](https://github.com/jlowin/fastmcp)构建，这是构建模型上下文协议服务器的快速、Pythonic方式。