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

Python开发者实战指南基于MCP Server的MySQL数据统计系统构建在数据驱动的时代快速获取和分析数据库信息已成为开发者日常工作的核心需求。对于Python开发者而言如何高效连接MySQL并实现灵活的数据统计同时又能与现代开发工具无缝集成是一个值得深入探讨的话题。本文将带你从零开始构建一个基于MCP Server的MySQL数据统计系统并集成到Cursor开发环境中。1. 环境准备与基础配置在开始项目前我们需要确保开发环境已正确配置。以下是所需的软件和工具清单Python 3.10MCP Server开发需要较新的Python版本MySQL数据库本地或远程实例均可Cursor编辑器作为集成开发环境基础Python包mcp[cli]pymysql安装基础依赖只需执行以下命令pip install mcp[cli] pymysql提示建议使用虚拟环境来隔离项目依赖避免与其他项目产生冲突对于MySQL数据库我们需要预先创建测试数据。以下SQL脚本将创建一个简单的用户表并插入示例数据CREATE DATABASE mcp_demo; USE mcp_demo; CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50) NOT NULL, age INT NOT NULL, department VARCHAR(50), salary DECIMAL(10,2) ); -- 插入示例数据 INSERT INTO users (name, age, department, salary) VALUES (张伟, 28, 技术部, 8500.00), (王芳, 32, 市场部, 9200.00), (李强, 25, 技术部, 7800.00), (赵敏, 30, 人事部, 8800.00), (刘洋, 35, 市场部, 9500.00);2. MCP Server核心开发MCP Server作为连接MySQL和Cursor的桥梁其核心功能是提供数据查询接口。我们将创建一个完整的统计服务包含多种查询功能。首先创建mysql_stats.py文件构建基础服务框架from mcp.server.fastmcp import FastMCP import pymysql from typing import Dict, List, Any mcp FastMCP(MySQLStatsServer) def get_db_connection(): 建立MySQL数据库连接 return pymysql.connect( hostlocalhost, port3306, userroot, passwordyourpassword, databasemcp_demo, cursorclasspymysql.cursors.DictCursor )接下来我们添加几个实用的统计函数mcp.tool() def get_department_stats(department: str) - Dict[str, Any]: 获取部门统计信息 conn get_db_connection() try: with conn.cursor() as cursor: # 部门人数统计 cursor.execute( SELECT COUNT(*) as count, AVG(age) as avg_age, AVG(salary) as avg_salary FROM users WHERE department%s, (department,) ) result cursor.fetchone() # 部门薪资分布 cursor.execute( SELECT salary FROM users WHERE department%s ORDER BY salary, (department,) ) salaries [row[salary] for row in cursor.fetchall()] return { department: department, employee_count: result[count], average_age: float(result[avg_age]), average_salary: float(result[avg_salary]), salary_distribution: salaries } finally: conn.close() mcp.tool() def age_group_analysis() - List[Dict[str, Any]]: 年龄分组分析 conn get_db_connection() try: with conn.cursor() as cursor: cursor.execute( SELECT CASE WHEN age 25 THEN Under 25 WHEN age BETWEEN 25 AND 30 THEN 25-30 WHEN age BETWEEN 31 AND 35 THEN 31-35 ELSE Over 35 END as age_group, COUNT(*) as count, AVG(salary) as avg_salary FROM users GROUP BY age_group ORDER BY age_group ) return cursor.fetchall() finally: conn.close() if __name__ __main__: mcp.run()3. 服务调试与验证开发完成后我们需要验证MCP Server是否正常工作。使用以下命令启动服务mcp dev mysql_stats.py服务启动后可以通过浏览器访问http://localhost:6274打开MCP Inspector进行测试在Tools标签页可以看到我们定义的两个工具点击Run Tool可以测试每个功能输入参数并查看返回结果对于get_department_stats工具可以尝试输入技术部作为参数应该返回类似以下结果{ department: 技术部, employee_count: 2, average_age: 26.5, average_salary: 8150.0, salary_distribution: [7800.0, 8500.0] }4. Cursor集成配置将MCP Server集成到Cursor中需要创建配置文件。在Cursor设置中找到MCP部分添加新的全局MCP Server配置。创建mcp.json文件内容如下{ mcpServers: { mysql_stats: { command: python, args: [/path/to/your/mysql_stats.py], environment: { PYTHONPATH: /path/to/your/project } } } }配置完成后在Cursor的聊天界面中就可以直接使用自然语言查询数据了。例如请统计市场部的员工数据分析公司员工的年龄分布情况比较各部门的平均薪资Cursor会自动调用对应的MCP工具并返回格式化结果。5. 高级功能扩展基础统计功能实现后我们可以进一步扩展服务能力5.1 添加缓存机制频繁查询相同数据会影响性能可以添加简单的缓存from functools import lru_cache mcp.tool() lru_cache(maxsize32) def cached_department_stats(department: str) - Dict[str, Any]: 带缓存的部门统计 return get_department_stats(department)5.2 支持复杂查询条件增加灵活查询接口支持多条件组合mcp.tool() def flexible_query( min_age: int None, max_age: int None, department: str None, min_salary: float None ) - List[Dict[str, Any]]: 灵活查询员工信息 conn get_db_connection() try: with conn.cursor() as cursor: query SELECT * FROM users WHERE 11 params [] if min_age is not None: query AND age %s params.append(min_age) if max_age is not None: query AND age %s params.append(max_age) if department is not None: query AND department %s params.append(department) if min_salary is not None: query AND salary %s params.append(min_salary) cursor.execute(query, params) return cursor.fetchall() finally: conn.close()5.3 性能监控端点添加服务健康检查接口mcp.tool() def service_health() - Dict[str, Any]: 服务健康状态检查 try: conn get_db_connection() with conn.cursor() as cursor: cursor.execute(SELECT 1) db_status OK except Exception as e: db_status fError: {str(e)} finally: conn.close() return { status: running, database: db_status, tools: [get_department_stats, age_group_analysis, flexible_query] }6. 实际应用中的优化建议在项目开发过程中有几个关键点值得注意连接管理确保每次查询后正确关闭数据库连接避免连接泄漏错误处理为每个工具添加适当的异常捕获和错误返回性能考量对于大数据量表考虑添加分页或限制返回结果数量安全防护使用参数化查询防止SQL注入避免直接拼接SQL字符串一个更健壮的工具实现示例mcp.tool() def safe_department_query(department: str, limit: int 100) - Dict[str, Any]: 安全的部门查询带结果限制 if not department or not isinstance(department, str): return {error: Invalid department parameter} if limit 500: return {error: Limit too high, maximum is 500} conn get_db_connection() try: with conn.cursor() as cursor: cursor.execute( SELECT id, name, age, salary FROM users WHERE department%s LIMIT %s, (department, limit) ) employees cursor.fetchall() cursor.execute( SELECT COUNT(*) as total FROM users WHERE department%s, (department,) ) total cursor.fetchone()[total] return { department: department, employees: employees, total_count: total, returned_count: len(employees), has_more: total len(employees) } except Exception as e: return {error: str(e)} finally: conn.close()7. 项目部署与团队协作当项目需要团队共享或部署到生产环境时考虑以下实践配置外部化将数据库连接信息移到环境变量或配置文件中依赖管理使用requirements.txt或pyproject.toml明确项目依赖文档生成为每个工具添加详细的文档字符串方便团队成员理解版本控制使用Git等工具管理代码变更示例配置分离后的代码import os from dotenv import load_dotenv load_dotenv() # 从.env文件加载环境变量 def get_db_connection(): 从环境变量获取数据库配置 return pymysql.connect( hostos.getenv(DB_HOST, localhost), portint(os.getenv(DB_PORT, 3306)), useros.getenv(DB_USER, root), passwordos.getenv(DB_PASSWORD), databaseos.getenv(DB_NAME, mcp_demo), cursorclasspymysql.cursors.DictCursor )对应的.env文件示例DB_HOSTlocalhost DB_PORT3306 DB_USERstats_user DB_PASSWORDsecure_password DB_NAMEmcp_production