mysql_mcp_server

Instructions

Implementation Reference

• src/mysql_mcp_server_pro/handles/execute_sql.py:37-63 (handler)

  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:

• src/mysql_mcp_server_pro/handles/execute_sql.py:20-36 (schema)

  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)

• src/mysql_mcp_server_pro/utils/execute_sql_util.py:207-230 (helper)

  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