10
10
from starlette .applications import Starlette
11
11
from starlette .routing import Route , Mount
12
12
13
- from handles import (
14
- ExecuteSQL ,
15
- GetChineseInitials ,
16
- GetTableIndex ,
17
- GetTableLock ,
18
- GetTableName ,
19
- GetTableDesc
20
- )
21
-
22
- # 初始化工具实例
23
- execute_sql = ExecuteSQL ()
24
- get_chinese_initials = GetChineseInitials ()
25
- get_table_index = GetTableIndex ()
26
- get_table_desc = GetTableDesc ()
27
- get_table_name = GetTableName ()
28
- get_table_lock = GetTableLock ()
29
-
13
+ from handles .base import ToolRegistry
30
14
31
15
# 初始化服务器
32
16
app = Server ("operateMysql" )
33
17
34
18
35
19
@app .list_tools ()
36
20
async def list_tools () -> list [Tool ]:
37
- """列出所有可用的MySQL操作工具
38
-
39
- Returns:
40
- list[Tool]: 返回工具列表,包含:
41
- - execute_sql: 执行SQL语句
42
- - get_chinese_initials: 获取中文拼音首字母
43
- - get_table_index: 获取表索引信息
44
- - get_table_desc: 获取表结构描述
45
- - get_table_name: 获取表名
46
- - get_table_lock: 获取表锁信息
47
21
"""
48
- return [
49
- execute_sql .get_tool_description (),
50
- get_chinese_initials .get_tool_description (),
51
- get_table_index .get_tool_description (),
52
- get_table_desc .get_tool_description (),
53
- get_table_name .get_tool_description (),
54
- get_table_lock .get_tool_description ()
55
- ]
22
+ 列出所有可用的MySQL操作工具
23
+ """
24
+ return ToolRegistry .get_all_tools ()
56
25
57
26
@app .call_tool ()
58
27
async def call_tool (name : str , arguments : dict ) -> Sequence [TextContent ]:
@@ -68,20 +37,9 @@ async def call_tool(name: str, arguments: dict) -> Sequence[TextContent]:
68
37
Raises:
69
38
ValueError: 当指定了未知的工具名称时抛出异常
70
39
"""
71
- if name == execute_sql .name :
72
- return await execute_sql .run_tool (arguments )
73
- elif name == get_table_index .name :
74
- return await get_table_index .run_tool (arguments )
75
- elif name == get_table_name .name :
76
- return await get_table_name .run_tool (arguments )
77
- elif name == get_table_lock .name :
78
- return await get_table_lock .run_tool (arguments )
79
- elif name == get_table_desc .name :
80
- return await get_table_desc .run_tool (arguments )
81
- elif name == get_chinese_initials .name :
82
- return await get_chinese_initials .run_tool (arguments )
83
-
84
- raise ValueError (f"未知的工具: { name } " )
40
+ tool = ToolRegistry .get_tool (name )
41
+
42
+ return await tool .run_tool (arguments )
85
43
86
44
87
45
async def run_stdio ():
0 commit comments