designcomputer
表名不准确
我发现这里面的执行逻辑可能有点问题，服务里提供了查询表结构的能力，但是在调用的时候并非强制执行，这会导致一个问题，大模型在生成sql语句的时候，会去猜测表名，导致无法查到具体有效的表； 2025-05-07 19:51:54,447 - mcp.server.lowlevel.server - INFO - Processing request of type ListToolsRequest 2025-05-07 19:51:54,448 - mysql_mcp_server - INFO - Listing tools... 2025-05-07 19:51:54,451 - mcp.server.lowlevel.server - INFO - Processing request of type ListToolsRequest 2025-05-07 19:51:54,451 - mysql_mcp_server - INFO - Listing tools... 2025-05-07 19:52:01,516 - mcp.server.lowlevel.server - INFO - Processing request of type CallToolRequest 2025-05-07 19:52:01,517 - mysql_mcp_server - INFO - Calling tool: execute_sql with arguments: {'query': "SELECT MONTH(date) AS month, SUM(revenue) AS total_revenue FROM tech_service_revenue WHERE YEAR(date) = 2023 AND city = 'A市' GROUP BY MONTH(date) ORDER BY total_revenue DESC LIMIT 3;"} 2025-05-07 19:52:01,575 - mysql.connector - INFO - package: mysql.connector.plugins 2025-05-07 19:52:01,575 - mysql.connector - INFO - plugin_name: caching_sha2_password 2025-05-07 19:52:01,575 - mysql.connector - INFO - AUTHENTICATION_PLUGIN_CLASS: MySQLCachingSHA2PasswordAuthPlugin 2025-05-07 19:52:01,578 - mysql_mcp_server - ERROR - Error executing SQL 'SELECT MONTH(date) AS month, SUM(revenue) AS total_revenue FROM tech_service_revenue WHERE YEAR(date) = 2023 AND city = 'A市' GROUP BY MONTH(date) ORDER BY total_revenue DESC LIMIT 3;': 1146 (42S02): Table 'caiwu.tech_service_revenue' doesn't exist
Result: 看起来数据库中不存在名为 tech_service_revenue 的表，或者表名可能有误。我需要确认正确的表名或查询其他可能包含科技服务收入数据的表。
请提供以下信息：
- 数据库中是否有其他表可能包含科技服务收入数据？
- 表名或字段名是否有其他拼写方式？
或者，您可以提供相关的表结构信息，以便我调整查询。
应该在执行sql语句前，让大模型了解表结构信息以及示例数据，以便保证输出可执行的sql语句；
@app.list_tools() async def list_tools() -> list[Tool]: """List available MySQL tools.""" logger.info("Listing tools...") return [ Tool( name="show_tables", description="在用户没有明确具体的表名的情况下，应该优先执行这个方法，以便了解当前数据库中有哪些表", inputSchema={ "type": "object", "properties": {}, "required": [] } ), Tool( name="execute_sql", description="执行只读SQL查询（仅支持SELECT语句）", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "要执行的SQL查询（仅允许SELECT语句）" } }, "required": ["query"] } ), Tool( name="describe_table", description="获取指定表的结构信息", inputSchema={ "type": "object", "properties": { "table_name": { "type": "string", "description": "要描述的表名" } }, "required": ["table_name"] } ), Tool( name="show_sample_data", description="获取指定表的示例数据（前5行）", inputSchema={ "type": "object", "properties": { "table_name": { "type": "string", "description": "要获取示例数据的表名" } }, "required": ["table_name"] } ) ]
@app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Execute SQL commands with safety checks.""" config = get_db_config() logger.info(f"Calling tool: {name} with arguments: {arguments}")
try:
with connect(**config) as conn:
with conn.cursor() as cursor:
if name == "show_tables":
cursor.execute("SHOW TABLES")
tables = cursor.fetchall()
result = ["数据库中的表"] # 标题
result.extend([table[0] for table in tables])
return [TextContent(type="text", text="\n".join(result))]
if name == "execute_sql":
query = arguments.get("query", "").strip()
if not query:
raise ValueError("Query is required")
# 安全检查：只允许 SELECT 查询
if not query.startswith("SELECT") and not query.startswith("SHOW") and not query.startswith("DESCRIBE"):
return [TextContent(type="text", text="错误：只允许执行 SELECT、SHOW 和 DESCRIBE 查询")]
# 禁止危险关键字
dangerous_keywords = ["DELETE", "DROP", "UPDATE", "INSERT", "TRUNCATE", "ALTER", "CREATE", "RENAME"]
if any(keyword in query for keyword in dangerous_keywords):
return [TextContent(type="text", text="错误：查询包含不允许的操作关键字")]
try:
# 创建新的连接和游标来执行查询
with connect(**config) as new_conn:
with new_conn.cursor() as new_cursor:
new_cursor.execute(query)
# 获取列名
columns = [desc[0] for desc in new_cursor.description]
# 获取所有数据
rows = new_cursor.fetchall()
# 格式化结果
result = [",".join(map(str, row)) for row in rows]
return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]
except Error as e:
logger.error(f"Error executing SQL '{query}': {e}")
return [TextContent(type="text", text=f"执行SQL时发生错误: {str(e)}")]
elif name == "describe_table":
table_name = arguments.get("table_name")
if not table_name:
raise ValueError("Table name is required")
cursor.execute(f"DESCRIBE {table_name}")
columns = ["Field", "Type", "Null", "Key", "Default", "Extra"]
rows = cursor.fetchall()
result = [",".join(columns)]
result.extend([",".join(map(str, row)) for row in rows])
return [TextContent(type="text", text="\n".join(result))]
elif name == "show_sample_data":
table_name = arguments.get("table_name")
if not table_name:
raise ValueError("Table name is required")
cursor.execute(f"SELECT * FROM {table_name} LIMIT 5")
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
result = [",".join(columns)]
result.extend([",".join(map(str, row)) for row in rows])
return [TextContent(type="text", text="\n".join(result))]
elif name == "execute_sql":
if name != "execute_sql":
raise ValueError(f"Unknown tool: {name}")
query = arguments.get("query")
if not query:
raise ValueError("Query is required")
try:
with connect(**config) as conn:
with conn.cursor() as cursor:
cursor.execute(query)
# Special handling for SHOW TABLES
if query.strip().upper().startswith("SHOW TABLES"):
tables = cursor.fetchall()
result = ["Tables_in_" + config["database"]] # Header
result.extend([table[0] for table in tables])
return [TextContent(type="text", text="\n".join(result))]
# Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.)
elif cursor.description is not None:
columns = [desc[0] for desc in cursor.description]
try:
rows = cursor.fetchall()
result = [",".join(map(str, row)) for row in rows]
return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]
except Error as e:
logger.warning(f"Error fetching results: {str(e)}")
return [TextContent(type="text", text=f"Query executed but error fetching results: {str(e)}")]
# Non-SELECT queries
else:
conn.commit()
return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")]
except Error as e:
logger.error(f"Error executing SQL '{query}': {e}")
return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
else:
raise ValueError(f"Unknown tool: {name}")
except Error as e:
logger.error(f"Error executing tool {name}: {e}")
return [TextContent(type="text", text=f"Error executing tool: {str(e)}")]
声明两个新的方法，查看表结构及查看示例数据，提高准确率
Thanks, I'll consider adding more tools. In my experience, I've found that success can vary greatly depending on the model and the prompt.
