MultiMyMCP
MCP 服务配置
复制以下 JSON 到 OPClaw 或其他 MCP 客户端的配置文件中即可使用
{
"mcpServers": {
"multi-mysql-mcp": {
"args": [
"-m",
"multimymcp.mcp_server"
],
"command": "python3"
}
}
}
服务介绍
MultiMyMCP - 生产级 MySQL 多数据源 MCP 工具
项目概述
MultiMyMCP 是一个完全适配 TRAE SOLO IDE 环境的生产级 MySQL 多数据源管理工具。它提供了增强型连接池、安全配置、性能监控等核心功能,支持标准 MCP 协议,让 AI 能够直接查询多个 MySQL 数据库。
核心特性
1. 标准 MCP 协议支持
- 完全符合 MCP (Model Context Protocol) 标准
- 支持 JSON-RPC 2.0 通信协议
- 提供 4 个核心工具:query_database、list_datasources、execute_sql、get_schema
- 与 TRAE SOLO IDE 无缝集成
2. 增强型连接池
- 支持连接池动态扩容/缩容
- 记录连接使用时长和 SQL 执行耗时
- 连接池状态监控(当前连接数、空闲连接数、等待队列长度)
- 线程安全的连接管理
3. 安全与通用规范
- SQL 执行超时控制
- 数据源配置加密存储(AES-256)
- SQL 白名单/黑名单机制
- 危险 SQL 操作拦截
4. 多数据源支持
- 支持同时管理多个 MySQL 数据源
- 灵活的数据源切换
- 全局配置文件管理
- 支持主从库配置
技术栈
- Python: 3.8+
- 核心依赖: pymysql, DBUtils>=2.0, python-dotenv, cryptography, json5
- 可选依赖: psutil, pytest
项目结构
MultiMyMCP/
├── multimymcp/
│ ├── __init__.py # 模块初始化
│ ├── core.py # 核心MCP接口
│ ├── config.py # 配置管理
│ ├── encryption.py # 加密工具
│ ├── pool.py # 连接池管理
│ ├── executor.py # SQL执行器
│ ├── monitor.py # 监控模块
│ ├── exceptions.py # 异常定义
│ ├── cli.py # 命令行接口
│ ├── mcp_server.py # MCP服务器主程序
│ ├── mcp_protocol.py # MCP协议处理器
│ ├── mcp_tools.py # MCP工具实现
│ └── mcp_config_loader.py # MCP配置加载器
├── tests/ # 单元测试
├── docs/ # 文档
├── requirements.txt # 依赖配置
├── setup.py # 打包配置
├── README.md # 项目说明
├── CHANGELOG.md # 更新日志
└── .env.example # 环境变量示例
安装步骤
1. 克隆项目
git clone git@gitee.com:wanhonglei/multi-mysql-mcp.git
cd multimymcp
2. 安装依赖
pip3 install -r requirements.txt
3. 安装项目(开发模式)
pip3 install -e .
4. 验证安装
python3 -m multimymcp.mcp_server --help
应该看到帮助信息输出。
MCP 服务器使用指南
第一步:创建配置文件
1.1 创建配置目录
mkdir -p ~/.multimymcp
1.2 创建配置文件
创建 ~/.multimymcp/mcp_config.json5 文件:
cat > ~/.multimymcp/mcp_config.json5 << 'EOF'
{
"datasources": {
"master": {
"host": "localhost",
"port": 3306,
"user": "root",
"password": "your_password",
"database": "production",
"pool_min_size": 2,
"pool_max_size": 10
},
"slave": {
"host": "localhost",
"port": 3307,
"user": "readonly",
"password": "your_password",
"database": "production",
"pool_min_size": 5,
"pool_max_size": 20
}
},
"security": {
"whitelist_enabled": false,
"blacklist_enabled": true,
"blacklist": ["DROP", "TRUNCATE", "ALTER", "CREATE"]
}
}
EOF
重要提示:
- 请将
your_password替换为实际的数据库密码 - 根据实际情况修改 host、port、user、database 等参数
- 可以添加更多数据源
第二步:测试 MCP 服务器
2.1 测试初始化请求
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | python3 -m multimymcp.mcp_server
预期输出:
{"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "multimymcp", "version": "1.0.0"}}}
2.2 测试工具列表
echo '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}' | python3 -m multimymcp.mcp_server
预期输出:
{"jsonrpc": "2.0", "id": 2, "result": {"tools": [...]}}
第三步:在 TRAE SOLO IDE 中配置
3.1 打开 MCP 设置
在 TRAE SOLO IDE 中:
方式一:通过菜单
- 点击菜单栏的
Preferences(偏好设置) - 选择
MCP Servers(MCP 服务器) - 点击
Edit MCP Settings(编辑 MCP 设置)
方式二:通过命令面板
- 按
Cmd + Shift + P(macOS)打开命令面板 - 输入
MCP - 选择
MCP: Edit MCP Settings
3.2 添加 MCP 服务器配置
在打开的配置文件中,添加以下内容:
{
"mcpServers": {
"multimymcp": {
"command": "python3",
"args": ["-m", "multimymcp.mcp_server"],
"env": {}
}
}
}
完整配置示例:
如果文件中已有其他 MCP 服务器,请添加到 mcpServers 对象中:
{
"mcpServers": {
"existing-server": {
"command": "...",
"args": [...]
},
"multimymcp": {
"command": "python3",
"args": ["-m", "multimymcp.mcp_server"],
"env": {}
}
}
}
3.3 保存并重启
- 保存配置文件(
Cmd + S) - 重启 MCP 服务器或重启 TRAE SOLO IDE
第四步:验证配置成功
4.1 检查 MCP 服务器状态
在 TRAE SOLO IDE 中:
- 打开命令面板(
Cmd + Shift + P) - 输入
MCP: Show Status - 查看
multimymcp是否显示为connected(已连接)
4.2 测试 AI 调用
在 TRAE SOLO IDE 的 AI 对话框中,尝试以下对话:
请列出所有可用的 MySQL 数据源
AI 应该会调用 list_datasources 工具并返回结果。
请查询 master 数据源中的 users 表,返回前 10 条记录
AI 应该会调用 query_database 工具执行查询。
可用的 MCP 工具
1. query_database
描述:在指定的 MySQL 数据源上执行查询 SQL
参数:
datasource: 数据源名称(必需)sql: 完整的 SELECT 查询语句(必需)
示例:
请从 master 数据源查询 users 表的前 10 条记录
2. list_datasources
描述:列出所有可用的 MySQL 数据源
参数:无
示例:
请列出所有可用的 MySQL 数据源
3. execute_sql
描述:执行 SQL 语句(INSERT/UPDATE/DELETE 等)
参数:
datasource: 数据源名称(必需)sql: 完整的 SQL 语句(必需)
示例:
请在 master 数据源中执行以下 SQL:
INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com')
4. get_schema
描述:获取指定数据源的表结构信息
参数:
datasource: 数据源名称(必需)table: 表名(可选,不指定则返回所有表)
示例:
请获取 master 数据源中 users 表的结构信息
Python API 使用
基本使用
from multimymcp import MultiMyMCP
# 初始化 MCP
mcp = MultiMyMCP()
# 连接数据源
mcp.connect('default')
# 执行 SQL
result = mcp.execute('SELECT * FROM users LIMIT 10')
print(result)
# 断开连接
mcp.disconnect()
多数据源配置
from multimymcp import MultiMyMCP, DataSourceConfig
mcp = MultiMyMCP()
# 添加主库
master_config = DataSourceConfig(
name="master",
host="master_db",
port=3306,
user="root",
password="master_password",
database="production"
)
mcp.add_data_source(master_config)
# 添加从库
slave_config = DataSourceConfig(
name="slave",
host="slave_db",
port=3306,
user="readonly",
password="slave_password",
database="production"
)
mcp.add_data_source(slave_config)
# 使用不同数据源
mcp.connect('master')
mcp.execute('INSERT INTO users (name) VALUES ("test")')
mcp.connect('slave')
result = mcp.execute('SELECT * FROM users')
事务处理
from multimymcp import MultiMyMCP
mcp = MultiMyMCP()
mcp.connect('default')
# 使用事务
try:
mcp._executors['default'].begin_transaction()
mcp.execute('INSERT INTO orders (user_id, amount) VALUES (1, 100)')
mcp.execute('UPDATE users SET balance = balance - 100 WHERE id = 1')
mcp._executors['default'].commit_transaction()
except Exception as e:
mcp._executors['default'].rollback_transaction()
raise
监控与调试
1. 连接池状态
status = mcp.get_pool_status()
print(f"当前连接数: {status['current_size']}")
print(f"活跃连接数: {status['active_size']}")
print(f"空闲连接数: {status['idle_size']}")
2. 性能报告
report = mcp.get_performance_report()
print(f"总执行次数: {report['summary']['total_sql_executions']}")
print(f"成功率: {report['summary']['success_rate']:.2f}%")
print(f"平均执行时间: {report['summary']['avg_execution_time']:.3f}s")
3. 健康检查
health = mcp.get_health_status()
if health['healthy']:
print("系统健康状态良好")
else:
print("系统存在问题:")
for issue in health['issues']:
print(f"- {issue}")
命令行工具
# 查看帮助
multimymcp --help
# 连接数据源
multimymcp connect default
# 执行 SQL
multimymcp execute "SELECT * FROM users"
# 查看状态
multimymcp status
# 健康检查
multimymcp health
# 性能报告
multimymcp performance
# 列出数据源
multimymcp list
# 调整连接池大小
multimymcp resize 5 20
常见问题排查
问题 1:MCP 服务器无法启动
症状:状态显示为 disconnected 或 error
解决方法:
-
检查 Python 路径是否正确:
which python3 -
在配置中使用完整路径:
{ "mcpServers": { "multimymcp": { "command": "/usr/bin/python3", "args": ["-m", "multimymcp.mcp_server"], "env": {} } } }
问题 2:找不到模块
症状:错误信息显示 ModuleNotFoundError
解决方法:
-
确认项目已安装:
pip3 install -e /path/to/MultiMyMCP -
在配置中添加 PYTHONPATH:
{ "mcpServers": { "multimymcp": { "command": "python3", "args": ["-m", "multimymcp.mcp_server"], "env": { "PYTHONPATH": "/path/to/MultiMyMCP" } } } }
问题 3:数据库连接失败
症状:AI 调用工具时返回连接错误
解决方法:
- 检查配置文件中的数据库连接信息
- 确认数据库服务正在运行
- 测试数据库连接:
mysql -h localhost -P 3306 -u root -p
安全配置
生产环境配置
{
"security": {
"whitelist_enabled": true,
"blacklist_enabled": true,
"whitelist": ["SELECT", "INSERT", "UPDATE", "DELETE"],
"blacklist": ["DROP", "TRUNCATE", "ALTER", "CREATE", "GRANT", "REVOKE"]
}
}
性能优化建议
1. 连接池优化
# 推荐配置
config = DataSourceConfig(
# 其他配置...
pool_min_size=5, # 基础并发数
pool_max_size=15, # 最大并发数
pool_timeout=30, # 连接池超时(秒)
sql_timeout=60 # SQL执行超时(秒)
)
2. SQL 优化
- 使用参数化查询
- 批量执行操作
- 使用事务减少网络开销
- 限制结果集大小
测试
运行单元测试:
pytest tests/ -v --cov=multimymcp
版本信息
- 版本: 1.0.0
- 发布日期: 2026-03-12
- Python 版本: 3.8+
许可证
MIT License
更新日志
v1.0.0 (2026-03-12)
- 初始版本发布
- 实现标准 MCP 协议支持
- 实现 4 个核心 MCP 工具
- 实现核心连接池管理
- 实现加密存储功能
- 实现 SQL 执行器和监控功能
- 完整的单元测试覆盖
- 详细的文档和使用指南
技术支持
如有任何问题,请参考文档或提交 Issue。