# MCP MySQL Server 代码架构文档
本文档详细介绍 MCP MySQL Server 工具的代码结构和各功能模块的作用。
## 📁 项目结构
```
mcp_mysql_server/
├── main.go # 主程序文件
├── go.mod # Go模块依赖
├── go.sum # 依赖校验文件
├── README.md # 项目说明文档
└── mcp-mysql-server # 编译后的可执行文件
```
## 🏗️ 代码架构概览
整个项目采用模块化设计，主要包含以下几个核心模块：
1.
**配置管理模块** - 处理环境变量和配置加载
2.
**数据库连接模块** - 管理MySQL数据库连接
3.
**安全控制模块** - SQL安全检查和防护
4.
**查询执行模块** - SQL查询执行和结果处理
5.
**表管理模块** - 数据库表相关操作
6.
**MCP服务模块** - MCP协议服务器和工具定义
7.
**主程序模块** - 程序入口和服务启动
---
## 🔧 详细模块分析
### 1. 配置管理模块
#### 📍
`Config`
结构体 (第19-29行)
```go
type
Config
struct {
Host
string
Port
string
User
string
Password
string
Database
string
AllowInsertOperation
bool
AllowUpdateOperation
bool
AllowDeleteOperation
bool
}
```
**作用：**
- 定义了MySQL数据库连接的所有配置参数
- 包含安全控制开关，决定是否允许写操作
- 为后续的数据库连接和安全检查提供配置基础
#### 📍 `loadConfig()` 函数
// 从环境变量加载配置
func loadConfig() *Config {
config := &Config{
Host: getEnv("MYSQL_HOST", "127.0.0.1"),
Port: getEnv("MYSQL_PORT", "3306"),
User: getEnv("MYSQL_USER", "root"),
Password: getEnv("MYSQL_PASS", ""),
Database: getEnv("MYSQL_DB", "mysql"),
AllowInsertOperation: getEnvBool("ALLOW_INSERT_OPERATION", false),
AllowUpdateOperation: getEnvBool("ALLOW_UPDATE_OPERATION", false),
AllowDeleteOperation: getEnvBool("ALLOW_DELETE_OPERATION", false),
}
return config
}
**作用：**
- 从环境变量中加载配置信息
- 为每个配置项提供默认值
- 确保程序在缺少环境变量时仍能正常运行
#### 📍 `getEnv()` 和 `getEnvBool()` 辅助函数
func getEnv(key, defaultValue string) string {
if value := os.Getenv(key); value != "" {
return value
}
return defaultValue
}
func getEnvBool(key string, defaultValue bool) bool {
if value := os.Getenv(key); value != "" {
result, err := strconv.ParseBool(value)
if err != nil {
return defaultValue
}
return result
}
return defaultValue
}
**作用：**
-
`getEnv()`: 安全地读取字符串类型环境变量
-
`getEnvBool()`: 安全地读取布尔类型环境变量，包含错误处理
- 提供环境变量读取的统一接口
---
### 2. 数据库连接模块
#### 📍 `createMySQLConnection()` 函数
// 创建 MySQL 连接
func createMySQLConnection(config *Config) (*sql.DB, error) {
dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local",
config.User, config.Password, config.Host, config.Port, config.Database)
db, err := sql.Open("mysql", dsn)
if err != nil {
return nil, fmt.Errorf("连接数据库失败: %v", err)
}
// 测试连接
if err := db.Ping(); err != nil {
return nil, fmt.Errorf("数据库连接测试失败: %v", err)
}
log.Printf("成功连接到 MySQL 数据库: %s:%s/%s", config.Host, config.Port, config.Database)
return db, nil
}
**作用：**
- 根据配置创建MySQL数据库连接
- 构建标准的MySQL DSN连接字符串
- 执行连接测试确保数据库可用
- 返回可用的数据库连接对象
**技术特点：**
- 使用UTF8MB4字符集支持emoji和特殊字符
- 启用时间解析和本地时区设置
- 包含连接失败的详细错误信息
---
### 3. 安全控制模块
#### 📍 `isSafeQuery()` 函数
// 检查 SQL 语句是否安全
func isSafeQuery(query string, config *Config) error {
query = strings.TrimSpace(strings.ToUpper(query))
// 检查是否包含危险的操作
dangerousOperations := []string{
"INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER",
"TRUNCATE", "GRANT", "REVOKE", "SET", "LOAD", "OUTFILE",
}
for _, op := range dangerousOperations {
if strings.Contains(query, op) {
switch op {
case "INSERT":
if !config.AllowInsertOperation {
return fmt.Errorf("INSERT 操作被禁用")
}
case "UPDATE":
if !config.AllowUpdateOperation {
return fmt.Errorf("UPDATE 操作被禁用")
}
case "DELETE":
if !config.AllowDeleteOperation {
return fmt.Errorf("DELETE 操作被禁用")
}
default:
return fmt.Errorf("危险操作 '%s' 被禁用", op)
}
}
}
// 确保是 SELECT 语句或被允许的操作
if !strings.HasPrefix(query, "SELECT") &&
!strings.HasPrefix(query, "SHOW") &&
!strings.HasPrefix(query, "DESCRIBE") &&
!strings.HasPrefix(query, "DESC") &&
!strings.HasPrefix(query, "EXPLAIN") {
return fmt.Errorf("只允许执行 SELECT、SHOW、DESCRIBE、EXPLAIN 查询语句")
}
return nil
}
**作用：**
- 检查SQL语句是否安全可执行
- 防止危险操作（DROP、CREATE、ALTER等）
- 根据配置允许特定的写操作（INSERT、UPDATE、DELETE）
- 确保只执行查询类语句
**安全机制：**
```go
dangerousOperations
:= []
string{
"INSERT",
"UPDATE",
"DELETE",
"DROP",
"CREATE",
"ALTER",
"TRUNCATE",
"GRANT",
"REVOKE",
"SET",
"LOAD",
"OUTFILE",
}
```
#### 📍 `isValidTableName()` 函数
// 验证表名是否安全
func isValidTableName(tableName string) bool {
// 只允许字母、数字、下划线
matched, _ := regexp.MatchString("^[a-zA-Z0-9_]+$", tableName)
return matched
}
**作用：**
- 验证表名格式防止SQL注入
- 只允许字母、数字、下划线组合
- 为表相关操作提供额外安全保障
---
### 4. 查询执行模块
#### 📍 `executeQuery()` 函数
// 执行 MySQL 查询
func executeQuery(ctx context.Context, db *sql.DB, query string, config *Config) ([]map[string]interface{}, error) {
if err := isSafeQuery(query, config); err != nil {
return nil, err
}
rows, err := db.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("查询执行失败: %v", err)
}
defer rows.Close()
columns, err := rows.Columns()
if err != nil {
return nil, fmt.Errorf("获取列信息失败: %v", err)
}
var results []map[string]interface{}
for rows.Next() {
values := make([]interface{}, len(columns))
valuePtrs := make([]interface{}, len(columns))
for i := range columns {
valuePtrs[i] = &values[i]
}
if err := rows.Scan(valuePtrs...); err != nil {
return nil, fmt.Errorf("扫描行数据失败: %v", err)
}
row := make(map[string]interface{})
for i, col := range columns {
val := values[i]
// 处理 []byte 类型的数据
if b, ok := val.([]byte); ok {
row[col] = string(b)
} else {
row[col] = val
}
}
results = append(results, row)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("遍历结果失败: %v", err)
}
return results, nil
}
**作用：**
- 核心查询执行引擎
- 安全执行SQL查询并返回结构化结果
- 处理不同数据类型的转换
- 将查询结果转换为统一的JSON格式
**处理流程：**
1. 安全检查 → 2. 执行查询 → 3. 获取列信息 → 4. 遍历结果 → 5. 数据类型转换 → 6. 返回结构化数据
**数据处理特点：**
- 自动处理
`[]byte`类型转换为字符串
- 支持动态列名和数据类型
- 返回标准的
`map[string]interface{}`格式
---
### 5. 表管理模块
#### 📍 `listTables()` 函数
// 获取数据库表列表
func listTables(ctx context.Context, db *sql.DB) ([]string, error) {
query := "SHOW TABLES"
rows, err := db.QueryContext(ctx, query)
if err != nil {
return nil, fmt.Errorf("获取表列表失败: %v", err)
}
defer rows.Close()
var tables []string
for rows.Next() {
var tableName string
if err := rows.Scan(&tableName); err != nil {
return nil, fmt.Errorf("扫描表名失败: %v", err)
}
tables = append(tables, tableName)
}
return tables, nil
}
**作用：**
- 获取数据库中所有表的列表
- 执行
`SHOW TABLES`命令
- 返回表名字符串数组
- 为数据库结构探索提供基础
#### 📍 `describeTable()` 函数
// 获取表结构信息
func describeTable(ctx context.Context, db *sql.DB, tableName string, config *Config) ([]map[string]interface{}, error) {
// 验证表名以防止 SQL 注入
if !isValidTableName(tableName) {
return nil, fmt.Errorf("无效的表名: %s", tableName)
}
query := fmt.Sprintf("DESCRIBE `%s`", tableName)
return executeQuery(ctx, db, query, config)
}
**作用：**
- 获取指定表的详细结构信息
- 安全地构建
`DESCRIBE`查询
- 复用
`executeQuery()`进行结果处理
- 提供表字段、类型、约束等信息
---
### 6. MCP服务模块
#### 📍 MCP服务器初始化
```go
mcpServer
:=
server.
NewMCPServer(
"MySQL数据库查询服务器",
"1.0.0")
```
**作用：**
- 创建MCP协议服务器实例
- 定义服务器名称和版本信息
- 为后续工具注册提供基础
#### 📍 工具定义和处理器
##### `mysql_query` 工具
// 定义 mysql_query 工具
queryTool := mcp.NewTool("mysql_query",
mcp.WithDescription("执行MySQL查询语句并返回结果（仅支持SELECT等查询语句）"),
mcp.WithString("sql",
mcp.Required(),
mcp.Description("要执行的SQL查询语句"),
),
)
// 添加 mysql_query 工具处理器
mcpServer.AddTool(queryTool, func(arguments map[string]interface{}) (*mcp.CallToolResult, error) {
sqlQuery, ok := arguments["sql"].(string)
if !ok {
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": "无效的SQL查询参数",
},
},
IsError: true,
}, nil
}
log.Printf("执行查询: %s", sqlQuery)
results, err := executeQuery(context.Background(), db, sqlQuery, config)
if err != nil {
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": fmt.Sprintf("查询执行失败: %v", err),
},
},
IsError: true,
}, nil
}
// 格式化输出结果
output := fmt.Sprintf("查询成功，返回 %d 行数据:\n", len(results))
if len(results) == 0 {
output += "没有找到匹配的数据。"
} else {
// 转换为易读的格式
jsonData, _ := json.MarshalIndent(results, "", " ")
output += string(jsonData)
}
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": output,
},
},
}, nil
})
**功能：**
- 执行SQL查询的核心工具
- 参数验证和查询执行
- 结果格式化和错误处理
- 支持JSON格式的结果输出
**处理流程：**
```
参数验证 → 查询执行 → 结果格式化 → 返回响应
```
##### `list_tables` 工具
// 定义 list_tables 工具
listTablesTool := mcp.NewTool("list_tables",
mcp.WithDescription("获取数据库中所有表的列表"),
)
// 添加 list_tables 工具处理器
mcpServer.AddTool(listTablesTool, func(arguments map[string]interface{}) (*mcp.CallToolResult, error) {
log.Printf("获取表列表")
tables, err := listTables(context.Background(), db)
if err != nil {
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": fmt.Sprintf("获取表列表失败: %v", err),
},
},
IsError: true,
}, nil
}
output := fmt.Sprintf("数据库共有 %d 个表:\n", len(tables))
for i, table := range tables {
output += fmt.Sprintf("%d. %s\n", i+1, table)
}
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": output,
},
},
}, nil
})
**功能：**
- 获取数据库表列表
- 无需参数的简单工具
- 格式化表列表输出
- 提供表数量统计
##### `describe_table` 工具
// 添加 describe_table 工具处理器
mcpServer.AddTool(describeTableTool, func(arguments map[string]interface{}) (*mcp.CallToolResult, error) {
tableName, ok := arguments["table_name"].(string)
if !ok {
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": "无效的表名参数",
},
},
IsError: true,
}, nil
}
log.Printf("获取表结构: %s", tableName)
columns, err := describeTable(context.Background(), db, tableName, config)
if err != nil {
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": fmt.Sprintf("获取表结构失败: %v", err),
},
},
IsError: true,
}, nil
}
output := fmt.Sprintf("表 '%s' 的结构信息:\n", tableName)
jsonData, _ := json.MarshalIndent(columns, "", " ")
output += string(jsonData)
return &mcp.CallToolResult{
Content: []interface{}{
map[string]interface{}{
"type": "text",
"text": output,
},
},
}, nil
})
**功能：**
- 查看表结构详情
- 表名参数验证
- JSON格式的结构信息输出
- 完整的字段信息展示
---
### 7. 主程序模块
#### 📍
`main()`
函数
**作用：**
- 程序入口点和初始化控制器
- 协调各个模块的初始化和运行
**启动流程：**
1.
**配置加载** - 从环境变量加载配置
2.
**数据库连接** - 建立MySQL连接
3.
**服务器创建** - 初始化MCP服务器
4.
**工具注册** - 注册所有MCP工具
5.
**配置输出** - 显示运行配置信息
6.
**服务启动** - 启动STDIO服务
**关键代码段：**
```go
// 配置和连接初始化
config
:=
loadConfig()
db,
err
:=
createMySQLConnection(
config)
// 工具注册
mcpServer.
AddTool(
queryTool,
queryHandler)
mcpServer.
AddTool(
listTablesTool,
listTablesHandler)
mcpServer.
AddTool(
describeTableTool,
describeTableHandler)
// 服务启动
server.
ServeStdio(
mcpServer)
```
---
## 🔒 安全设计理念
### 多层安全防护
1.
**SQL注入防护** - 正则表达式验证和参数化查询
2.
**操作权限控制** - 可配置的写操作开关
3.
**查询类型限制** - 默认只允许查询类语句
4.
**表名验证** - 严格的表名格式检查
### 配置灵活性
- 环境变量驱动的配置管理
- 细粒度的操作权限控制
- 默认安全的配置策略
---
## 🚀 扩展性设计
### 模块化架构
- 清晰的功能模块分离
- 可复用的工具函数
- 统一的错误处理机制
### MCP协议支持
- 标准的MCP工具定义
- 统一的响应格式
- 完整的错误信息传递
---
## 📊 性能考虑
### 连接管理
- 单一数据库连接复用
- 连接测试和错误处理
- 资源自动清理
### 查询优化
- 结果集高效处理
- 内存友好的数据转换
- 最小化数据复制
---
## 🛠️ 技术栈
-
**Go 1.23.2+** - 主要编程语言
-
**github.com/go-sql-driver/mysql** - MySQL驱动
-
**github.com/mark3labs/mcp-go** - MCP协议支持
-
**标准库** - JSON处理、正则表达式、上下文管理
项目地址：https://github.com/tianyuso/mcp_mysql_server
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/26366838/viewspace-3095527/ ，如需转载，请注明出处，否则将追究法律责任。
下一篇：
没有了~
请登录后发表评论
登录
全部评论
