# MySQL MCP Server Pro Complete Call Flow
## 📋 Table of Contents
1. [Server Startup Flow](#server-startup-flow)
2. [AI Client Connection Flow](#ai-client-connection-flow)
3. [Complete Call Flow Example](#complete-call-flow-example)
4. [Code Execution Path](#code-execution-path)
---
## Server Startup Flow
### 1. Server Initialization Phase
```
Server startup (server.py)
↓
Load environment variable configuration (.env)
↓
Create database connection pool (ExecuteSqlUtil.create_mysql_pool)
↓
Import handles module (handles/__init__.py)
↓
Automatically register all tool classes
```
### 2. Automatic Tool Registration Mechanism
**Code Location: `src/mysql_mcp_server_pro/handles/base.py`**
```python
# When importing GetTableName class
from .get_table_name import GetTableName # handles/__init__.py
# BaseHandler.__init_subclass__ automatically executed
class BaseHandler:
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
if cls.name: # If class defines name attribute
ToolRegistry.register(cls) # Automatically register to tool registry
↓
# ToolRegistry.register() execution:
tool = tool_class() # Create tool instance
cls._tools[tool.name] = tool # Store in dictionary
# _tools = {
# "get_table_name": GetTableName instance,
# "execute_sql": ExecuteSQL instance,
# "get_table_desc": GetTableDesc instance,
# ...
# }
```
**Execution Timing:**
- ✅ Automatically executed when importing modules
- ✅ No manual invocation required
- ✅ All classes inheriting `BaseHandler` are automatically registered
---
## AI Client Connection Flow
### Step 1: Client Connects to Server
```
AI Client (Cursor/Claude Desktop, etc.)
↓
Connect to server via MCP protocol
↓
Establish communication channel (STDIO/SSE/StreamableHttp)
```
### Step 2: Client Requests Tool List
```
AI client sends request: list_tools()
↓
Server receives request: server.py @app.list_tools()
↓
Call: ToolRegistry.get_all_tools()
↓
Iterate through all registered tool instances
↓
Call tool.get_tool_description() for each tool
```
**Detailed Execution Path:**
```python
# server.py lines 67-72
@app.list_tools()
async def list_tools() -> list[Tool]:
return ToolRegistry.get_all_tools()
↓
# base.py lines 41-48
@classmethod
def get_all_tools(cls) -> list[Tool]:
return [tool.get_tool_description() for tool in cls._tools.values()]
↓
# For get_table_name tool:
# get_table_name.py lines 20-34
def get_tool_description(self) -> Tool:
return Tool(
name="get_table_name",
description="Search for table names in the database based on Chinese table names or table descriptions",
inputSchema={
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "Chinese table name or table description to search, single query only"
}
},
"required": ["text"]
}
)
```
### Step 3: Client Receives Tool List
```
Server returns tool list (JSON format)
↓
AI client parses tool information
↓
AI client now knows:
- What tools are available
- Each tool's name and description
- What parameters each tool requires
```
**Returned Data Example:**
```json
[
{
"name": "get_table_name",
"description": "Search for table names in the database based on Chinese table names or table descriptions",
"inputSchema": {
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "Chinese table name or table description to search, single query only"
}
},
"required": ["text"]
}
},
{
"name": "execute_sql",
"description": "Execute SQL on MySQL database",
...
},
...
]
```
---
## Complete Call Flow Example
### Scenario: User queries "data of Zhang San in user info table"
#### Phase 1: User Input
```
User inputs natural language: "Query data of Zhang San in user info table"
↓
AI client (e.g., Cursor) receives user input
```
#### Phase 2: AI Analysis and Decision
```
AI analyzes user intent
↓
Determines need for:
1. First find the actual table name corresponding to "user info table"
2. Get table structure
3. Generate SQL query
4. Execute query
↓
AI decides to call tool: get_table_name
```
#### Phase 3: Call get_table_name Tool
```
AI client sends request via MCP protocol:
call_tool(
name="get_table_name",
arguments={"text": "user info table"}
)
↓
Server receives request: server.py @app.call_tool()
↓
Code execution path:
```
**Detailed Code Execution:**
```python
# ========== Server-side Execution ==========
# 1. server.py lines 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 lines 24-39
@classmethod
def get_tool(cls, name: str) -> 'BaseHandler':
if name not in cls._tools: # Check if tool exists
raise ValueError(f"Unknown tool: {name}")
return cls._tools[name] # Return GetTableName instance
↓
# 3. Call tool's run_tool method
return await tool.run_tool(arguments) # arguments = {"text": "user info table"}
↓
# 4. get_table_name.py lines 36-58
async def run_tool(self, arguments: Dict[str, Any]):
# 4.1 Validate parameters
if "text" not in arguments:
raise ValueError("Missing query statement")
text = arguments["text"] # text = "user info table"
# 4.2 Get database configuration
config = get_db_config() # Get database connection information
# 4.3 Create ExecuteSQL instance
execute_sql = ExecuteSQL()
# 4.4 Build SQL statement (hardcoded metadata query 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}%';"
# Final SQL:
# SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COMMENT
# FROM information_schema.TABLES
# WHERE TABLE_SCHEMA = 'your_database'
# AND TABLE_COMMENT LIKE '%user info table%';
# 4.5 Call execute_sql tool to execute SQL
return await execute_sql.run_tool({"query": sql})
↓
# 5. execute_sql.py lines 37-64
async def run_tool(self, arguments: Dict[str, Any]):
query = arguments["query"] # Get SQL statement
# 5.1 Create ExecuteSqlUtil instance
exe = ExecuteSqlUtil()
# 5.2 Execute SQL (may contain multiple statements, separated by semicolons)
sql_results = exe.execute_multiple_statements(query)
↓
# 5.3 execute_sql_util.py execution
# - Check permissions
# - Get database connection (from connection pool)
# - Execute SQL
# - Return results
# 5.4 Format results
results = []
for result in sql_results:
formatted_result = exe.format_result(result)
results.append(formatted_result)
# 5.5 Return formatted results
return [TextContent(type="text", text="\n---\n".join(results))]
```
#### Phase 4: Return Results to AI
```
Server returns query results:
TextContent(
type="text",
text="TABLE_SCHEMA,TABLE_NAME,TABLE_COMMENT\n
your_database,user_info,user info table"
)
↓
AI client receives results
↓
AI parses results, learns:
- Table name is "user_info"
- Database is "your_database"
```
#### Phase 5: AI Continues Calling Other Tools
```
AI decides to continue calling: get_table_desc
↓
call_tool(
name="get_table_desc",
arguments={"text": "user_info"}
)
↓
Get table structure:
- id (int)
- name (varchar)
- age (int)
- email (varchar)
↓
AI generates business SQL:
SELECT * FROM user_info WHERE name = 'Zhang San';
↓
AI calls: execute_sql
↓
call_tool(
name="execute_sql",
arguments={"query": "SELECT * FROM user_info WHERE name = 'Zhang San';"}
)
↓
Execute SQL, return query results
↓
AI returns results to user
```
---
## Code Execution Path
### Complete Call Chain Diagram
```
User input
↓
AI Client
↓
MCP protocol communication
↓
┌─────────────────────────────────────┐
│ Server-side (MySQL MCP Server Pro) │
├─────────────────────────────────────┤
│ │
│ 1. app.call_tool │
│ ↓ │
│ 2. Returns tool instance │
│ ↓ │
│ 3. tool.run_tool(arguments) │
│ (Specific tool class method) │
│ ↓ │
│ 4. Tool internal logic: │
│ - Build SQL (e.g., get_table_name) │
│ - Call other tools (e.g., execute_sql) │
│ ↓ │
│ 5. ExecuteSqlUtil │
│ - Permission check │
│ - Get connection from pool │
│ - Execute SQL │
│ - Return results │
│ │
└─────────────────────────────────────┘
↓
Return results to AI client
↓
AI processes results, may continue calling other tools
↓
Final results returned to user
```
### Key Code Location Reference Table
| Step | File Path | Line Number | Function |
|------|-----------|-------------|----------|
| Server startup entry | `server.py` | 202-244 | `main()` function |
| Tool registration | `handles/base.py` | 56-60 | `BaseHandler.__init_subclass__()` |
| Tool registry | `handles/base.py` | 6-48 | `ToolRegistry` class |
| List all tools | `server.py` | 67-72 | `@app.list_tools()` |
| Call tool entry | `server.py` | 74-90 | `@app.call_tool()` |
| Get tool instance | `handles/base.py` | 24-39 | `ToolRegistry.get_tool()` |
| get_table_name implementation | `handles/get_table_name.py` | 36-58 | `run_tool()` method |
| execute_sql implementation | `handles/execute_sql.py` | 37-64 | `run_tool()` method |
| SQL execution utility class | `utils/execute_sql_util.py` | 129-319 | `ExecuteSqlUtil` class |
---
## Sequence Diagram
```
User AI Client Server Tool Registry Tool Instance Database
│ │ │ │ │ │
│ Input query │ │ │ │ │
├─────────────>│ │ │ │ │
│ │ │ │ │ │
│ │ list_tools() │ │ │ │
│ ├───────────────>│ │ │ │
│ │ │ get_all_tools() │ │ │
│ │ ├────────────────>│ │ │
│ │ │ │ get_description() │
│ │ │ ├───────────────>│ │
│ │ │ │ │ │
│ │ │ │<───────────────┤ │
│ │ │<────────────────┤ │ │
│ │<───────────────┤ │ │ │
│ │ │ │ │ │
│ │ call_tool() │ │ │ │
│ │ ("get_table_ │ │ │ │
│ │ name", {...}) │ │ │ │
│ ├───────────────>│ │ │ │
│ │ │ get_tool() │ │ │
│ │ ├────────────────>│ │ │
│ │ │ │ │ │
│ │ │ │ return instance│ │
│ │ │<────────────────┤ │ │
│ │ │ │ │ │
│ │ │ run_tool() │ │ │
│ │ ├─────────────────────────────────>│ │
│ │ │ │ │ │
│ │ │ │ │ Build SQL │
│ │ │ │ │ │
│ │ │ │ │ call execute │
│ │ │ │ │_sql │
│ │ │ │ ├───────────────>│
│ │ │ │ │ │
│ │ │ │ │ Execute SQL │
│ │ │ │ │ │
│ │ │ │ │<───────────────┤
│ │ │ │ │ │
│ │ │ │ │ return result │
│ │ │<─────────────────────────────────┤ │
│ │<───────────────┤ │ │ │
│ │ │ │ │ │
│ │ (Continue calling other tools...)│ │ │
│ │ │ │ │ │
│ │ Return final result │ │ │
│<──────────────┤ │ │ │ │
│ │ │ │ │ │
```
---
## Key Points Summary
### 1. Automatic Registration Mechanism
- ✅ All classes inheriting `BaseHandler` are automatically registered when imported
- ✅ No manual tool registration required
- ✅ Just import in `handles/__init__.py`
### 2. Role of Tool Description
- ✅ AI client learns tool functionality through `get_tool_description()`
- ✅ AI decides when to call which tool based on descriptions
- ✅ `inputSchema` in description tells AI what parameters are needed
### 3. Tool Call Flow
1. AI client sends `call_tool(name, arguments)`
2. Server gets tool instance via `ToolRegistry.get_tool(name)`
3. Call tool's `run_tool(arguments)` method
4. Tool executes internal logic (may call other tools)
5. Return results to AI client
### 4. Inter-Tool Calls
- ✅ Tools can call other tools (e.g., `get_table_name` calls `execute_sql`)
- ✅ By creating tool instances and calling `run_tool()` method
- ✅ Tool chaining forms complex workflows
---
Hope this flow diagram helps you understand the entire call mechanism! 🚀
