使用说明
mcp mysql服务器
一个简单的Python脚本,用于通过基于FastMCP的标准输入/输出(stdio)与本地MySQL数据库进行交互。
概述
mcp-mysql-server 提供了一个命令行界面,用于查询和检查本地MySQL数据库的模式。它的设计是为了在首选轻量级、基于脚本的交互的本地开发环境中易于使用。该脚本利用命令和结果的标准输入和输出,使其易于与其他命令行工具和工作流集成。
特性
- 查询执行: 执行SQL SELECT语句,并以简单易读的格式查看结果。
- 数据操作: 使用write_table工具将数据插入表中。
- 架构检查: 检索有关数据库架构的信息,包括:
- 使用list_tables工具列出所有表。
- 使用get_table_schema显示特定表的列及其数据类型。
- 连接测试: 测试数据库连接并获取服务器信息。
- 可配置的访问控制: 通过配置定义连接参数(主机、用户、密码、数据库),允许安全访问数据库。
- stdio接口: 使用命令的标准输入与脚本交互,并通过标准输出接收输出,从而促进脚本编写和自动化。
- 本地部署重点: 针对在本地开发环境中使用进行了优化。
入门指南
先决条件
- Python 3.x已安装在您的系统上。
- 正在运行的本地MySQL服务器实例。
- 所需的Python库(使用requirements.txt文件安装):
pip install -r requirements.txt
安装
- 将存储库下载或克隆到本地计算机。
配置
-
创建配置文件(例如。,
config.ini)在与脚本相同的目录中,或在运行脚本时指定路径。配置文件应具有以下格式:[mysql] host = localhost port = 3306 # Optional, defaults to 3306 user = your_mysql_user password = your_mysql_password database = your_database_name注: 确保指定的MySQL用户具有执行您打算执行的操作所需的权限。
用法
使用以下命令运行脚本,指定配置文件的路径:
python mcp-mysql-server.py --config config.ini
该脚本公开了以下可用于与MySQL数据库交互的工具:
test_connection:测试与数据库的连接并返回服务器信息。list_tables:列出当前数据库中的所有表。read_table:读取并返回指定表中的所有数据。write_table:将新行数据插入到指定表中。get_table_schema:返回指定表的架构(列定义)。execute_sql:执行自定义SQL查询并返回结果。
项目结构
mcp-mysql-server.py:包含MySQL MCP服务器实现的主脚本config.iniMySQL连接参数的配置文件requirements.txt:Python依赖项列表changes.log:项目更改日志
Docker支持
这个项目可以使用Docker Compose在Docker容器中运行。设置包括:
- 带有示例数据的MySQL容器
- MCP MySQL服务器容器
使用Docker运行
- 确保你的系统上安装了Docker和Docker Compose。
- 克隆此存储库。
- 从存储库根目录运行:
docker-compose up -d
这将:
- 使用示例数据(用户和帖子表)启动MySQL服务器
- 启动连接到MySQL数据库的MCP MySQL服务器
Docker配置
特殊配置文件(config.docker.ini)在Docker中运行时使用,它指向MySQL容器。
测试
提供单元测试以确保功能正常工作。测试使用mocking来避免需要实际的数据库连接。
在本地运行测试
# Make a copy of the main script with underscore naming
cp mcp-mysql-server.py mcp_mysql_server.py
# Run the tests
python -m unittest tests.py
使用Docker运行测试
docker-compose -f docker-compose.test.yml up --build
这将构建一个测试容器并在其中运行单元测试。
发展
对于想要扩展此工具的开发人员,您可以通过创建其他MCP工具来添加新功能。每个工具都是一个Python函数,用 @mcp.tool() 它处理特定的数据库操作。
添加新工具的示例:
@mcp.tool()
def my_new_tool(param1: str, param2: int) -> dict:
"""
Documentation for the new tool.
"""
connection = create_mysql_connection()
cursor = connection.cursor()
try:
# Your tool implementation here
return {"result": "success"}
finally:
cursor.close()
connection.close()