mysql_mcp_server quickstart

docker启动mysql服务

mysql服务，用于mcp server连接

docker run --name mysql-mcp-new -e MYSQL_ROOT_PASSWORD=123456 -e MYSQL_DATABASE=test_db -p 13306:3306 -d mysql:8.0

环境安装

git clone https://github.com/designcomputer/mysql_mcp_server.git
uv venv
source .venv/bin/activate
uv pip install -r requirements.txt 

完整代码

import asyncio
import logging
import os
import sys
from mysql.connector import connect, Error
from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
from pydantic import AnyUrl# Configure logging
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mysql_mcp_server")def get_db_config():"""Get database configuration from environment variables."""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")}# Remove None values to let MySQL connector use defaults if not specifiedconfig = {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")raise ValueError("Missing required database configuration")return config# Initialize server
app = Server("mysql_mcp_server")@app.list_resources()
async def list_resources() -> list[Resource]:"""List MySQL tables as resources."""config = get_db_config()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()}")with conn.cursor() as cursor:cursor.execute("SHOW TABLES")tables = cursor.fetchall()logger.info(f"Found tables: {tables}")resources = []for table in tables:resources.append(Resource(uri=f"mysql://{table[0]}/data",name=f"Table: {table[0]}",mimeType="text/plain",description=f"Data in table: {table[0]}"))return resourcesexcept Error as e:logger.error(f"Failed to list resources: {str(e)}")logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")return []@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:"""Read table contents."""config = get_db_config()uri_str = str(uri)logger.info(f"Reading resource: {uri_str}")if not uri_str.startswith("mysql://"):raise ValueError(f"Invalid URI scheme: {uri_str}")parts = uri_str[8:].split('/')table = parts[0]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()}")with conn.cursor() as cursor:cursor.execute(f"SELECT * FROM {table} LIMIT 100")columns = [desc[0] for desc in cursor.description]rows = cursor.fetchall()result = [",".join(map(str, row)) for row in rows]return "\n".join([",".join(columns)] + result)except Error as e:logger.error(f"Database error reading resource {uri}: {str(e)}")logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")raise RuntimeError(f"Database error: {str(e)}")@app.list_tools()
async def list_tools() -> list[Tool]:"""List available MySQL tools."""logger.info("Listing tools...")return [Tool(name="execute_sql",description="Execute an SQL query on the MySQL server",inputSchema={"type": "object","properties": {"query": {"type": "string","description": "The SQL query to execute"}},"required": ["query"]})]@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:"""Execute SQL commands."""config = get_db_config()logger.info(f"Calling tool: {name} with arguments: {arguments}")if name != "execute_sql":raise ValueError(f"Unknown tool: {name}")query = arguments.get("query")if not query:raise ValueError("Query is required")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()}")with conn.cursor() as cursor:cursor.execute(query)# Special handling for SHOW TABLESif query.strip().upper().startswith("SHOW TABLES"):tables = cursor.fetchall()result = ["Tables_in_" + config["database"]]  # Headerresult.extend([table[0] for table in tables])return [TextContent(type="text", text="\n".join(result))]# Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.)elif cursor.description is not None:columns = [desc[0] for desc in cursor.description]try:rows = cursor.fetchall()result = [",".join(map(str, row)) for row in rows]return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]except Error as e:logger.warning(f"Error fetching results: {str(e)}")return [TextContent(type="text", text=f"Query executed but error fetching results: {str(e)}")]# Non-SELECT querieselse:conn.commit()return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")]except Error as e:logger.error(f"Error executing SQL '{query}': {e}")logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")return [TextContent(type="text", text=f"Error executing query: {str(e)}")]async def main():"""Main entry point to run the MCP server."""from mcp.server.stdio import stdio_server# Add additional debug outputprint("Starting MySQL MCP server with config:", file=sys.stderr)config = get_db_config()print(f"Host: {config['host']}", file=sys.stderr)print(f"Port: {config['port']}", file=sys.stderr)print(f"User: {config['user']}", file=sys.stderr)print(f"Database: {config['database']}", file=sys.stderr)logger.info("Starting MySQL MCP server...")logger.info(f"Database config: {config['host']}/{config['database']} as {config['user']}")async with stdio_server() as (read_stream, write_stream):try:await app.run(read_stream,write_stream,app.create_initialization_options())except Exception as e:logger.error(f"Server error: {str(e)}", exc_info=True)raiseif __name__ == "__main__":asyncio.run(main())

代码解释

导入模块和配置日志

import asyncio
import logging
import os
import sys
from mysql.connector import connect, Error
from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
from pydantic import AnyUrl# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mysql_mcp_server")

这部分导入了必要的模块：

• asyncio：用于异步编程
• mysql.connector：MySQL数据库连接器
• mcp.server和mcp.types：MCP协议相关的类和类型
• 配置了日志系统，便于调试和监控

数据库配置函数

def get_db_config():"""从环境变量获取数据库配置。"""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"),# 添加字符集和排序规则以避免utf8mb4_0900_ai_ci在旧版MySQL中的问题"charset": os.getenv("MYSQL_CHARSET", "utf8mb4"),"collation": os.getenv("MYSQL_COLLATION", "utf8mb4_unicode_ci"),# 启用自动提交以便更好地控制事务"autocommit": True,# 设置SQL模式以提高兼容性"sql_mode": os.getenv("MYSQL_SQL_MODE", "TRADITIONAL")}# 移除None值，让MySQL连接器使用默认值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("缺少必要的数据库配置。请检查环境变量：")logger.error("MYSQL_USER, MYSQL_PASSWORD, 和 MYSQL_DATABASE 是必需的")raise ValueError("缺少必要的数据库配置")return config

这个函数从环境变量中读取MySQL连接配置：

• 支持配置主机、端口、用户名、密码、数据库名等
• 设置了默认的字符集和排序规则，解决兼容性问题
• 验证必要的配置项（用户名、密码、数据库名）是否存在

初始化MCP服务器

# 初始化服务器
app = Server("mysql_mcp_server")

创建了一个名为"mysql_mcp_server"的MCP服务器实例。

资源列表功能

@app.list_resources()
async def list_resources() -> list[Resource]:"""列出MySQL表作为资源。"""# 实现代码...

这个函数实现了MCP的资源列表功能：

• 连接到MySQL数据库
• 执行SHOW TABLES命令获取所有表
• 将每个表转换为MCP资源格式返回
• 每个资源包含URI、名称、MIME类型和描述

读取资源功能

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:"""读取表内容。"""# 实现代码...

这个函数实现了读取资源的功能：

• 解析URI获取表名
• 连接到MySQL数据库
• 执行SELECT * FROM {table} LIMIT 100查询
• 将结果格式化为CSV格式的字符串返回

工具列表功能

@app.list_tools()
async def list_tools() -> list[Tool]:"""列出可用的MySQL工具。"""# 实现代码...

这个函数定义了MCP服务器提供的工具：

• 返回一个名为execute_sql的工具
• 该工具允许执行SQL查询
• 定义了工具的输入模式（需要一个query参数）

工具调用功能

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:"""执行SQL命令。"""# 实现代码...

这个函数处理工具调用：

• 验证工具名称是否为execute_sql
• 从参数中获取SQL查询
• 连接到MySQL数据库并执行查询
• 根据查询类型处理结果：
  • 对于SHOW TABLES有特殊处理
  • 对于返回结果集的查询（如SELECT），格式化为CSV
  • 对于非查询操作（如INSERT），返回受影响的行数

主函数

async def main():"""运行MCP服务器的主入口点。"""# 实现代码...if __name__ == "__main__":asyncio.run(main())

主函数负责：

• 输出调试信息，显示数据库配置
• 初始化stdio服务器（标准输入输出通信）
• 启动MCP服务器
• 处理异常并记录错误

总结

这个代码实现了一个完整的MySQL MCP服务器，它允许AI模型通过标准化的MCP协议与MySQL数据库交互。服务器提供了以下功能：

1. 列出数据库中的表作为资源
2. 读取表的内容
3. 执行任意SQL查询

inspector测试

npx @modelcontextprotocol/inspector

inspector配置

参数配置

环境变量配置

基础测试语句

SHOW TABLES;

这个语句会列出数据库中的所有表，是最基本的测试，可以验证连接是否正常。

创建表测试

CREATE TABLE test_users (id INT AUTO_INCREMENT PRIMARY KEY,username VARCHAR(50) NOT NULL,email VARCHAR(100) UNIQUE,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

插入数据测试

INSERT INTO test_users (username, email) VALUES 
('user1', 'user1@example.com'),
('user2', 'user2@example.com'),
('user3', 'user3@example.com');

查询数据测试

SELECT * FROM test_users;

更复杂的查询测试

SELECT username, email, DATE_FORMAT(created_at, '%Y-%m-%d') AS registration_date 
FROM test_users 
WHERE id > 1 
ORDER BY created_at DESC;

表结构查询测试

DESCRIBE test_users;