execute_sql
Execute SQL queries on MySQL5.6 databases using a supported MCP server. Supports multiple SQL executions, table and field queries, execution plan analysis, and Chinese-to-pinyin conversion.
Instructions
在MySQL5.6s数据库上执行SQL
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | 要执行的SQL语句 |
Implementation Reference
- The `run_tool` method of ExecuteSQL class: core handler logic for 'execute_sql' tool. Validates input, executes SQL using ExecuteSqlUtil, formats and returns results as TextContent.async def run_tool(self, arguments: Dict[str, Any]) -> Sequence[TextContent]: """执行SQL工具 Args: arguments: 包含SQL查询的参数字典 Returns: 执行结果文本序列 """ if "query" not in arguments: return [TextContent(type="text", text="错误: 缺少查询语句")] query = arguments["query"] try: exe = ExecuteSqlUtil() sql_results = exe.execute_multiple_statements(query) # 格式化结果 results = [] for result in sql_results: formatted_result = exe.format_result(result) results.append(formatted_result) return [TextContent(type="text", text="\n---\n".join(results))] except SQLExecutionError as e:
- The `get_tool_description` method defines the input schema for 'execute_sql' tool: object with required 'query' string property.def get_tool_description(self) -> Tool: """获取工具描述""" return Tool( name=self.name, description=self.description, inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "要执行的SQL语句" } }, "required": ["query"] } )
- src/mysql_mcp_server_pro/handles/base.py:56-61 (registration)BaseHandler.__init_subclass__ automatically registers subclasses like ExecuteSQL to ToolRegistry when they have a defined 'name'.def __init_subclass__(cls, **kwargs): """子类初始化时自动注册到工具注册表""" super().__init_subclass__(**kwargs) if cls.name: # 只注册有名称的工具 ToolRegistry.register(cls)
- ExecuteSqlUtil.execute_multiple_statements: splits query by ';', executes each statement via execute_single_statement (with permissions, pooling), collects results. Called by handler.def execute_multiple_statements(self, query: str) -> List[SQLResult]: """执行多条SQL语句 Args: query: 包含多条SQL语句的查询字符串，以分号分隔 Returns: SQL执行结果列表 """ statements = [stmt.strip() for stmt in query.split(';') if stmt.strip()] results = [] for statement in statements: try: result = self.execute_single_statement(statement) results.append(result) except Exception as e: logger.warning(f"SQL执行警告: {e}, SQL: {statement}") results.append(SQLResult( success=False, message=f"执行失败: {str(e)}" )) return results
- src/mysql_mcp_server_pro/handles/base.py:10-22 (registration)ToolRegistry.register: instantiates the handler class and stores it in the registry by name, called automatically during subclass init.@classmethod def register(cls, tool_class: Type['BaseHandler']) -> Type['BaseHandler']: """注册工具类 Args: tool_class: 要注册的工具类 Returns: 返回注册的工具类，方便作为装饰器使用 """ tool = tool_class() cls._tools[tool.name] = tool return tool_class