# MySQL MCP Server Pro 完整调用流程
## 📋 目录
1. [服务器启动流程](#服务器启动流程)
2. [AI客户端连接流程](#ai客户端连接流程)
3. [完整调用流程示例](#完整调用流程示例)
4. [代码执行路径](#代码执行路径)
---
## 服务器启动流程
### 1. 服务器初始化阶段
```
服务器启动 (server.py)
↓
加载环境变量配置 (.env)
↓
创建数据库连接池 (ExecuteSqlUtil.create_mysql_pool)
↓
导入 handles 模块 (handles/__init__.py)
↓
自动注册所有工具类
```
### 2. 工具自动注册机制
**代码位置：`src/mysql_mcp_server_pro/handles/base.py`**
```python
# 当导入 GetTableName 类时
from .get_table_name import GetTableName # handles/__init__.py
# BaseHandler.__init_subclass__ 自动执行
class BaseHandler:
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
if cls.name: # 如果类定义了 name 属性
ToolRegistry.register(cls) # 自动注册到工具注册表
↓
# ToolRegistry.register() 执行：
tool = tool_class() # 创建工具实例
cls._tools[tool.name] = tool # 存储到字典中
# _tools = {
# "get_table_name": GetTableName实例,
# "execute_sql": ExecuteSQL实例,
# "get_table_desc": GetTableDesc实例,
# ...
# }
```
**执行时机：**
- ✅ 在导入模块时自动执行
- ✅ 无需手动调用
- ✅ 所有继承 `BaseHandler` 的类都会被自动注册
---
## AI客户端连接流程
### 步骤 1：客户端连接到服务器
```
AI客户端 (Cursor/Claude Desktop等)
↓
通过 MCP 协议连接到服务器
↓
建立通信通道 (STDIO/SSE/StreamableHttp)
```
### 步骤 2：客户端请求工具列表
```
AI客户端发送请求: list_tools()
↓
服务器收到请求: server.py @app.list_tools()
↓
调用: ToolRegistry.get_all_tools()
↓
遍历所有已注册的工具实例
↓
对每个工具调用: tool.get_tool_description()
```
**详细执行路径：**
```python
# server.py 第 67-72 行
@app.list_tools()
async def list_tools() -> list[Tool]:
return ToolRegistry.get_all_tools()
↓
# base.py 第 41-48 行
@classmethod
def get_all_tools(cls) -> list[Tool]:
return [tool.get_tool_description() for tool in cls._tools.values()]
↓
# 对 get_table_name 工具：
# get_table_name.py 第 20-34 行
def get_tool_description(self) -> Tool:
return Tool(
name="get_table_name",
description="根据表中文名或表描述搜索数据库中对应的表名",
inputSchema={
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "要搜索的表中文名、表描述，仅支持单个查询"
}
},
"required": ["text"]
}
)
```
### 步骤 3：客户端接收工具列表
```
服务器返回工具列表 (JSON格式)
↓
AI客户端解析工具信息
↓
AI客户端现在知道：
- 有哪些工具可用
- 每个工具的名称、描述
- 每个工具需要什么参数
```
**返回的数据示例：**
```json
[
{
"name": "get_table_name",
"description": "根据表中文名或表描述搜索数据库中对应的表名",
"inputSchema": {
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "要搜索的表中文名、表描述，仅支持单个查询"
}
},
"required": ["text"]
}
},
{
"name": "execute_sql",
"description": "在MySQL数据库上执行SQL",
...
},
...
]
```
---
## 完整调用流程示例
### 场景：用户查询"用户信息表中张三的数据"
#### 阶段 1：用户输入
```
用户输入自然语言: "查询用户信息表中张三的数据"
↓
AI客户端 (如 Cursor) 接收到用户输入
```
#### 阶段 2：AI 分析和决策
```
AI 分析用户意图
↓
判断需要：
1. 先找到"用户信息表"对应的实际表名
2. 获取表结构
3. 生成 SQL 查询
4. 执行查询
↓
AI 决定调用工具: get_table_name
```
#### 阶段 3：调用 get_table_name 工具
```
AI客户端通过 MCP 协议发送请求:
call_tool(
name="get_table_name",
arguments={"text": "用户信息表"}
)
↓
服务器接收请求: server.py @app.call_tool()
↓
代码执行路径：
```
**详细代码执行：**
```python
# ========== 服务器端执行 ==========
# 1. server.py 第 74-90 行
@app.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]):
tool = ToolRegistry.get_tool(name) # name = "get_table_name"
↓
# 2. base.py 第 24-39 行
@classmethod
def get_tool(cls, name: str) -> 'BaseHandler':
if name not in cls._tools: # 检查工具是否存在
raise ValueError(f"未知的工具: {name}")
return cls._tools[name] # 返回 GetTableName 实例
↓
# 3. 调用工具的 run_tool 方法
return await tool.run_tool(arguments) # arguments = {"text": "用户信息表"}
↓
# 4. get_table_name.py 第 36-58 行
async def run_tool(self, arguments: Dict[str, Any]):
# 4.1 验证参数
if "text" not in arguments:
raise ValueError("缺少查询语句")
text = arguments["text"] # text = "用户信息表"
# 4.2 获取数据库配置
config = get_db_config() # 获取数据库连接信息
# 4.3 创建 ExecuteSQL 实例
execute_sql = ExecuteSQL()
# 4.4 构建 SQL 语句（硬编码的元数据查询 SQL）
sql = "SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COMMENT "
sql += f"FROM information_schema.TABLES "
sql += f"WHERE TABLE_SCHEMA = '{config['database']}' "
sql += f"AND TABLE_COMMENT LIKE '%{text}%';"
# 最终 SQL:
# SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COMMENT
# FROM information_schema.TABLES
# WHERE TABLE_SCHEMA = 'your_database'
# AND TABLE_COMMENT LIKE '%用户信息表%';
# 4.5 调用 execute_sql 工具执行 SQL
return await execute_sql.run_tool({"query": sql})
↓
# 5. execute_sql.py 第 37-64 行
async def run_tool(self, arguments: Dict[str, Any]):
query = arguments["query"] # 获取 SQL 语句
# 5.1 创建 ExecuteSqlUtil 实例
exe = ExecuteSqlUtil()
# 5.2 执行 SQL（可能包含多条，以分号分隔）
sql_results = exe.execute_multiple_statements(query)
↓
# 5.3 execute_sql_util.py 执行
# - 检查权限
# - 获取数据库连接（从连接池）
# - 执行 SQL
# - 返回结果
# 5.4 格式化结果
results = []
for result in sql_results:
formatted_result = exe.format_result(result)
results.append(formatted_result)
# 5.5 返回格式化的结果
return [TextContent(type="text", text="\n---\n".join(results))]
```
#### 阶段 4：返回结果给 AI
```
服务器返回查询结果:
TextContent(
type="text",
text="TABLE_SCHEMA,TABLE_NAME,TABLE_COMMENT\n
your_database,user_info,用户信息表"
)
↓
AI客户端接收结果
↓
AI解析结果，得知：
- 表名是 "user_info"
- 数据库是 "your_database"
```
#### 阶段 5：AI 继续调用其他工具
```
AI 决定继续调用: get_table_desc
↓
call_tool(
name="get_table_desc",
arguments={"text": "user_info"}
)
↓
获取表结构:
- id (int)
- name (varchar)
- age (int)
- email (varchar)
↓
AI 生成业务 SQL:
SELECT * FROM user_info WHERE name = '张三';
↓
AI 调用: execute_sql
↓
call_tool(
name="execute_sql",
arguments={"query": "SELECT * FROM user_info WHERE name = '张三';"}
)
↓
执行 SQL，返回查询结果
↓
AI 将结果返回给用户
```
---
## 代码执行路径
### 完整的调用链路图
```
用户输入
↓
AI客户端 (Cursor/Claude Desktop)
↓
MCP 协议通信
↓
┌─────────────────────────────────────┐
│ 服务器端 (MySQL MCP Server Pro) │
├─────────────────────────────────────┤
│ │
│ 1. server.py │
│ @app.call_tool() │
│ ↓ │
│ 2. ToolRegistry.get_tool(name) │
│ 返回工具实例 │
│ ↓ │
│ 3. tool.run_tool(arguments) │
│ (具体工具类的方法) │
│ ↓ │
│ 4. 工具内部逻辑: │
│ - 构建 SQL (如 get_table_name) │
│ - 调用其他工具 (如 execute_sql) │
│ ↓ │
│ 5. ExecuteSqlUtil │
│ - 权限检查 │
│ - 连接池获取连接 │
│ - 执行 SQL │
│ - 返回结果 │
│ │
└─────────────────────────────────────┘
↓
返回结果给 AI客户端
↓
AI处理结果，可能继续调用其他工具
↓
最终结果返回给用户
```
### 关键代码位置对照表
| 步骤 | 文件路径 | 行号 | 功能 |
|------|---------|------|------|
| 服务器启动入口 | `server.py` | 202-244 | `main()` 函数 |
| 工具注册 | `handles/base.py` | 56-60 | `BaseHandler.__init_subclass__()` |
| 工具注册表 | `handles/base.py` | 6-48 | `ToolRegistry` 类 |
| 列出所有工具 | `server.py` | 67-72 | `@app.list_tools()` |
| 调用工具入口 | `server.py` | 74-90 | `@app.call_tool()` |
| 获取工具实例 | `handles/base.py` | 24-39 | `ToolRegistry.get_tool()` |
| get_table_name 实现 | `handles/get_table_name.py` | 36-58 | `run_tool()` 方法 |
| execute_sql 实现 | `handles/execute_sql.py` | 37-64 | `run_tool()` 方法 |
| SQL 执行工具类 | `utils/execute_sql_util.py` | 129-319 | `ExecuteSqlUtil` 类 |
---
## 时序图
```
用户 AI客户端 服务器 工具注册表 工具实例 数据库
│ │ │ │ │ │
│ 输入查询 │ │ │ │ │
├─────────────>│ │ │ │ │
│ │ │ │ │ │
│ │ list_tools() │ │ │ │
│ ├──────────────>│ │ │ │
│ │ │ get_all_tools()│ │ │
│ │ ├───────────────>│ │ │
│ │ │ │ get_description() │
│ │ │ ├──────────────>│ │
│ │ │ │ │ │
│ │ │ │<──────────────┤ │
│ │ │<───────────────┤ │ │
│ │<──────────────┤ │ │ │
│ │ │ │ │ │
│ │ call_tool() │ │ │ │
│ │ ("get_table_ │ │ │ │
│ │ name", {...})│ │ │ │
│ ├──────────────>│ │ │ │
│ │ │ get_tool() │ │ │
│ │ ├───────────────>│ │ │
│ │ │ │ │ │
│ │ │ │ return instance │
│ │ │<───────────────┤ │ │
│ │ │ │ │ │
│ │ │ run_tool() │ │ │
│ │ ├───────────────────────────────>│ │
│ │ │ │ │ │
│ │ │ │ │ 构建 SQL │
│ │ │ │ │ │
│ │ │ │ │ call execute │
│ │ │ │ │_sql │
│ │ │ │ ├─────────────>│
│ │ │ │ │ │
│ │ │ │ │ 执行 SQL │
│ │ │ │ │ │
│ │ │ │ │<─────────────┤
│ │ │ │ │ │
│ │ │ │ │ return result│
│ │ │<───────────────────────────────┤ │
│ │<──────────────┤ │ │ │
│ │ │ │ │ │
│ │ (继续调用其他工具...) │ │ │
│ │ │ │ │ │
│ │ 返回最终结果 │ │ │ │
│<─────────────┤ │ │ │ │
│ │ │ │ │ │
```
---
## 关键点总结
### 1. 自动注册机制
- ✅ 所有继承 `BaseHandler` 的类在导入时自动注册
- ✅ 无需手动注册工具
- ✅ 只需在 `handles/__init__.py` 中导入即可
### 2. 工具描述的作用
- ✅ AI 客户端通过 `get_tool_description()` 了解工具功能
- ✅ AI 根据描述决定何时调用哪个工具
- ✅ 描述中的 `inputSchema` 告诉 AI 需要什么参数
### 3. 工具调用流程
1. AI 客户端发送 `call_tool(name, arguments)`
2. 服务器通过 `ToolRegistry.get_tool(name)` 获取工具实例
3. 调用工具的 `run_tool(arguments)` 方法
4. 工具内部执行逻辑（可能调用其他工具）
5. 返回结果给 AI 客户端
### 4. 工具之间的调用
- ✅ 工具可以调用其他工具（如 `get_table_name` 调用 `execute_sql`）
- ✅ 通过创建工具实例并调用 `run_tool()` 方法
- ✅ 工具链式调用形成复杂的工作流
---
希望这份流程图能帮助你理解整个调用机制！🚀
