在本地部署mcp服务器实现自然语言操作mysql数据库,轻松实现数据表的增~ 删~ 改~ 查~

1.将写好的mcp_server代码放在本地任意盘!

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())

2.再cherry studio中导入json配置参数

{"mcpServers": {"mysql": {"command": "uv","args": ["--directory","path/to/mysql_mcp_server","run","mysql_mcp_server"],"env": {"MYSQL_HOST": "localhost","MYSQL_PORT": "3306","MYSQL_USER": "your_username","MYSQL_PASSWORD": "your_password","MYSQL_DATABASE": "your_database"}}}
}

其中args中的参数说明:~
“–directory”, # 目录参数
“path/to/mysql_mcp_server.py”, # mcpserver端程序放置绝对路径!
“run”, # uv启动python脚本!
“mysql_mcp_server.py” # mcpserver代码 !

其中env中为本地或者远程数据库配置参数:~
“MYSQL_HOST”: “localhost”, #mysql主机IP
“MYSQL_PORT”: “3306”, #mysql服务端口
“MYSQL_USER”: “your_username”, #数据库用户名
“MYSQL_PASSWORD”: “your_password”, #数据库密码
“MYSQL_DATABASE”: “your_database” #数据库名

3.再cherry studio中新建会话,选择mysql mcp服务器

在这里插入图片描述

即可使用自然语言对数据表进行增删改查!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如若转载,请注明出处:http://www.pswp.cn/pingmian/88175.shtml
繁体地址,请注明出处:http://hk.pswp.cn/pingmian/88175.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

2025快手创作者中心发布视频python实现

难度还行,只有一个__NS_sig3加密,流程麻烦点cookies_list cookie.split("; ")cookie_dict {}# 遍历每个 Cookie,根据等号将键值对拆分并添加到字典中for cookie in cookies_list:key_value cookie.split("")if len(ke…

Android 组件内核

文章目录什么是binder1. 什么是Binder?2. Binder架构组成3. 工作原理与通信流程1)服务注册2)服务查询3)通信过程4)核心数据结构4. 关键技术点5. 常见面试考点1)Binder与传统IPC(Socket、管道、共…

java类加载机制:Tomcat的类加载机制

Tomcat类加载机制深度解析:打破双亲委派的Web容器实现 Tomcat作为Java Web容器,其类加载机制为满足Web应用的隔离性、热部署和兼容性需求,对标准Java类加载机制进行了定制化扩展,核心是打破双亲委派模型并引入多层级类加载器。以下…

【PTA数据结构 | C语言版】从顺序表 list 中删除第 i 个元素

本专栏持续输出数据结构题目集,欢迎订阅。 文章目录题目代码题目 请编写程序,将 n 个整数存入顺序表,对任一指定的第 i 个位置,将这个位置上的元素从顺序表中删除。注意:i 代表位序,从 1 开始,…

VS2022 C++ EasyX库 扫雷游戏项目开发:打造经典游戏的详细之旅

老样子,先上效果 视频演示 C经典扫雷-介绍一、引言 在这篇博客中,我将详细介绍扫雷游戏项目的开发过程。扫雷作为一款经典的游戏,其规则简单但富有挑战性。通过开发这个项目,我不仅加深了对 C 编程的理解,还提升了自己…

Go语言网络游戏服务器模块化编程

本文以使用origin框架(一款使用Go语言写的开源游戏服务器框架)为例进行说明,当然也可以使用其它的框架或者自己写。 在框架中PBProcessor用来处理Protobuf消息,在使用之前,需要使用Register函数注册网络消息&#xff…

【机器人】Aether 多任务世界模型 | 4D动态重建 | 视频预测 | 视觉规划

Aether 是一个的世界模型,整合几何重建与生成建模的统一框架,实现类人空间推理能力。 来自ICCV 2025,该框架具有三大核心功能: (1) 4D动态重建,(2) 动作条件视频预测, (3) 目标条件视觉规划。 代码地址&…

MiniMind:3小时训练26MB微型语言模型,开源项目助力AI初学者快速入门

开发|界面|引擎|交付|副驾——重写全栈法则:AI原生的倍速造应用流来自全栈程序员 nine 的探索与实践,持续迭代中。 欢迎关注评论私信交流~ 在大型语言模型(LLaMA、GPT等)日益流行的今天,一个名为…

相机Camera日志实例分析之五:相机Camx【萌拍闪光灯后置拍照】单帧流程日志详解

【关注我,后续持续新增专题博文,谢谢!!!】 上一篇我们讲了: 这一篇我们开始讲: 目录 一、场景操作步骤 二、日志基础关键字分级如下 三、场景日志如下: 一、场景操作步骤 操作步…

[2-02-02].第03节:环境搭建 - Win10搭建ES集群环境

ElasticSearch学习大纲 基于ElasticSearch7.8版本 一、ElasticStack下载: 1.Elasticsearch 的官方地址 2.Elasticsearch 下载地址: 二、集群搭建: 第1步:创建es目录: 1.创建 elasticsearch-cluster 文件夹,在内部…

操作系统核心技术剖析:从Android驱动模型到鸿蒙微内核的国产化实践

目录 一、移动端操作系统技术细节 1. Android 内核版本 核心模块 驱动架构 国内定制案例 2. iOS XNU内核关键模块 安全机制 3. HarmonyOS 多内核架构 驱动隔离 二、PC端操作系统技术细节 1. Windows NT内核 模块分层 驱动模型 国内适配 2. macOS(X…

整合Spring、Spring MVC与MyBatis:构建高效Java Web应用

本文将详细讲解如何整合Spring、Spring MVC和MyBatis(SSM框架),通过一个人员信息查询案例展示完整开发流程。所有代码基于提供的文件实现。一、项目结构src/ ├── main/ │ ├── java/ │ │ └── com/ │ │ └── qcb…

视频插帧技术:从流畅观影到AI创作的革命

一、起源:为什么需要视频插帧? 视频的本质是连续播放的静态帧序列,帧率(FPS) 决定了流畅度。早期电影受限于拍摄技术和存储成本,普遍采用24FPS,而现代显示设备(如120Hz屏幕&#xf…

【一起来学AI大模型】PyTorch 实战示例:使用 BatchNorm 处理张量(Tensor)

PyTorch 实战示例 演示如何在神经网络中使用 BatchNorm 处理张量(Tensor),涵盖关键实现细节和常见陷阱。示例包含数据准备、模型构建、训练/推理模式切换及结果分析。示例场景:在 CIFAR-10 数据集上实现带 BatchNorm 的 CNNimport…

第8章:应用层协议HTTP、SDN软件定义网络、组播技术、QoS

应用层协议HTTP 应用层协议概述 应用层协议非常多,我们重点熟悉以下常见协议功能即可。 Telnet:远程登录协议,基于TCP 23端口,用于远程管理设备,采用明文传输。安全外壳协议 (SecureShell,SSH) ,基于TCP 22端口,用于…

uniapp页面间通信

uniapp中通过eventChannel实现页面间通信的方法,这是一种官方推荐的高效传参方式。我来解释下这种方式的完整实现和注意事项:‌发送页面(父页面)‌:uni.navigateTo({url: /pages/detail/detail,success: (res) > {/…

Android ViewModel机制与底层原理详解

Android 的 ViewModel 是 Jetpack 架构组件库的核心部分,旨在以生命周期感知的方式存储和管理与 UI 相关的数据。它的核心目标是解决两大痛点: 数据持久化: 在配置变更(如屏幕旋转、语言切换、多窗口模式切换)时保留数…

双倍硬件=双倍性能?TDengine线性扩展能力深度实测验证!

软件扩展能力是软件架构设计中的一个关键要素,具有良好扩展能力的软件能够充分利用新增的硬件资源。当软件性能与硬件增加保持同步比例增长时,我们称这种现象为软件具有线性扩展能力。要实现这种线性扩展并不简单,它要求软件架构精心设计&…

频繁迭代下完成iOS App应用上架App Store:一次快速交付项目的完整回顾

在一次面向商户的会员系统App开发中,客户要求每周至少更新一次版本,涉及功能迭代、UI微调和部分支付方案的更新。团队使用Flutter进行跨平台开发,但大部分成员日常都在Windows或Linux环境,只有一台云Mac用于打包。如何在高频率发布…

springsecurity03--异常拦截处理(认证异常、权限异常)

目录 Spingsecurity异常拦截处理 认证异常拦截 权限异常拦截 注册异常拦截器 设置跨域访问 Spingsecurity异常拦截处理 认证异常拦截 /*自定义认证异常处理器类*/ Component public class MyAuthenticationExceptionHandler implements AuthenticationEntryPoint {Overr…