Python开发者必看：如何用MCP Server在Cursor中快速实现MySQL数据分析（附完整代码）

在数据驱动的时代，Python开发者经常需要与MySQL数据库打交道。无论是日常开发中的数据查询需求，还是自动化报表生成，高效的数据处理能力都至关重要。本文将带你探索如何利用MCP Server在Cursor中构建一个强大的MySQL数据分析工具链，让数据查询变得前所未有的简单。

1. MCP Server与Cursor的协同工作原理

MCP Server（Model Control Protocol Server）是一种轻量级的服务框架，它允许开发者将自定义功能封装为可被外部系统调用的"工具"。而Cursor作为一款智能编程助手，能够无缝集成这些工具，实现自然语言到数据库查询的转换。

这种组合的核心优势在于：

• 自然语言交互：开发者可以用日常语言描述查询需求
• 功能复用：一次开发，多次调用
• 性能优化：查询逻辑在服务端执行，减轻客户端负担

典型的应用场景包括：

• 快速数据探查
• 定期报表生成
• 业务指标监控
• 数据质量检查

2. 环境准备与依赖安装

在开始之前，确保你的开发环境满足以下要求：

系统要求：

• Python 3.10或更高版本
• MySQL 5.7+/8.0数据库服务
• Cursor IDE（最新版本）

依赖安装：

pip install "mcp[cli]" pymysql

这两个核心包的作用分别是：

• mcp[cli]：提供MCP Server开发框架
• pymysql：Python连接MySQL的驱动程序

提示：建议使用虚拟环境管理项目依赖，避免包冲突

3. MySQL数据准备与模型设计

为了演示完整的流程，我们先在MySQL中创建测试数据。以下SQL脚本将创建一个用户表并插入示例数据：

CREATE DATABASE IF NOT EXISTS mcp_analysis;
USE mcp_analysis;

CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    age INT NOT NULL,
    registration_date DATE NOT NULL,
    account_balance DECIMAL(10,2) DEFAULT 0.00,
    is_active BOOLEAN DEFAULT TRUE
);

-- 插入示例数据
INSERT INTO users (username, age, registration_date, account_balance, is_active) VALUES
('Alice', 28, '2022-01-15', 1250.50, TRUE),
('Bob', 35, '2021-11-22', 3200.75, TRUE),
('Charlie', 42, '2020-05-30', 850.00, FALSE),
('David', 31, '2023-02-18', 2100.00, TRUE),
('Eve', 26, '2023-04-05', 500.25, TRUE);

这个数据结构包含了：

• 基础用户信息
• 账户状态
• 财务数据
• 时间维度信息

4. 开发MCP Server核心功能

创建一个名为mysql_analytics.py的文件，实现以下核心功能：

from mcp.server.fastmcp import FastMCP
import pymysql
from typing import Dict, Any
import json

mcp = FastMCP("MySQLAnalytics")

def get_db_connection():
    """建立MySQL数据库连接"""
    return pymysql.connect(
        host="localhost",
        port=3306,
        user="your_username",
        password="your_password",
        database="mcp_analysis",
        cursorclass=pymysql.cursors.DictCursor
    )

@mcp.tool()
def user_stats(condition: str = "") -> Dict[str, Any]:
    """获取用户统计信息"""
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            base_query = "SELECT COUNT(*) as total, AVG(age) as avg_age, SUM(account_balance) as total_balance FROM users"
            if condition:
                cursor.execute(f"{base_query} WHERE {condition}")
            else:
                cursor.execute(base_query)
            return cursor.fetchone()
    finally:
        conn.close()

@mcp.tool()
def query_users(query: str) -> list:
    """执行自定义查询"""
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(query)
            return cursor.fetchall()
    finally:
        conn.close()

if __name__ == "__main__":
    mcp.run()

这个实现包含两个主要工具：

1. user_stats：提供预定义的统计查询
2. query_users：支持执行原始SQL查询

注意：实际生产环境中应该使用环境变量管理数据库凭证，而非硬编码在代码中

5. 配置Cursor使用MCP Server

要让Cursor识别并使用我们的MCP Server，需要创建配置文件mcp.json：

{
    "mcpServers": {
        "mysql_analytics": {
            "command": "python",
            "args": ["/path/to/your/mysql_analytics.py"],
            "description": "MySQL数据分析服务"
        }
    }
}

配置步骤：

1. 在Cursor中打开设置
2. 导航到MCP设置部分
3. 添加新的全局MCP Server
4. 将上述JSON内容粘贴到配置文件中
5. 保存并重启Cursor

验证配置是否成功：

• 在Cursor聊天框中输入/mcp list
• 应该能看到你定义的MySQL分析服务

6. 实际应用案例与高级技巧

现在，你可以通过自然语言与数据库交互了。以下是一些实用场景：

基础查询：

"查询年龄大于30岁的用户数量"

Cursor会自动转换为：

result = user_stats("age > 30")
print(f"符合条件的用户数量: {result['total']}")

复杂分析：

"计算活跃用户的平均账户余额，按年龄段分组"

可以扩展MCP Server添加专门的方法：

@mcp.tool()
def age_group_analysis() -> list:
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            sql = """
                SELECT 
                    FLOOR(age/10)*10 as age_group,
                    AVG(account_balance) as avg_balance,
                    COUNT(*) as user_count
                FROM users
                WHERE is_active = TRUE
                GROUP BY age_group
                ORDER BY age_group
            """
            cursor.execute(sql)
            return cursor.fetchall()
    finally:
        conn.close()

性能优化技巧：

• 使用连接池管理数据库连接
• 为常用查询添加缓存层
• 实现分页查询支持
• 添加查询超时机制

7. 安全最佳实践

在开发生产级MCP Server时，安全至关重要：

输入验证：

from pymysql import escape_string

@mcp.tool()
def safe_query(condition: str) -> list:
    # 验证输入只包含字母、数字和特定符号
    if not re.match(r'^[a-zA-Z0-9_ =<>]+$', condition):
        raise ValueError("Invalid condition format")
    
    escaped_cond = escape_string(condition)
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(f"SELECT * FROM users WHERE {escaped_cond}")
            return cursor.fetchall()
    finally:
        conn.close()

其他安全措施：

• 实现基于令牌的认证
• 限制查询返回的行数
• 记录所有查询操作
• 使用最小权限的数据库账户

8. 扩展思路与进阶应用

MCP Server的潜力不仅限于简单查询，还可以：

1. 集成数据可视化：

@mcp.tool()
def generate_age_distribution_chart():
    data = age_group_analysis()
    # 使用matplotlib生成图表
    # 返回图表文件路径或Base64编码

2. 实现业务逻辑：

@mcp.tool()
def apply_discount(min_balance: float, discount_rate: float):
    """为高净值用户应用折扣"""
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            sql = """
                UPDATE users 
                SET account_balance = account_balance * %s
                WHERE account_balance >= %s AND is_active = TRUE
            """
            cursor.execute(sql, (1 - discount_rate, min_balance))
            conn.commit()
            return {"affected_rows": cursor.rowcount}
    finally:
        conn.close()

3. 多数据源联合查询：

@mcp.tool()
def cross_database_report():
    """从多个数据库生成合并报表"""
    # 连接MySQL
    # 连接PostgreSQL
    # 执行联合分析

在实际项目中，我发现最实用的技巧是为每个业务领域创建专门的MCP Server，而不是把所有功能塞进一个服务。这样既保持了模块化，又便于团队协作开发。