execute_query

Instructions

Args:
    query: SQL查询语句
    params: 查询参数，用于参数化查询，防止SQL注入
    db_config: 数据库连接配置参数，如果为None则使用默认配置

Returns:
    包含查询结果的字典

Implementation Reference

• mysql-mcp.py:121-175 (handler)

  The main handler for the 'execute_query' tool. It executes a SQL query with optional parameters and database config, returns results for SELECT/SHOW/DESCRIBE/EXPLAIN queries, or affected rows/last insert ID for other queries (INSERT/UPDATE/DELETE). Includes error handling and connection cleanup.

  @mcp.tool()
  async def execute_query(query: str, params: Optional[List[Any]] = None, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      """执行SQL查询语句，返回查询结果
  
      Args:
          query: SQL查询语句
          params: 查询参数，用于参数化查询，防止SQL注入
          db_config: 数据库连接配置参数，如果为None则使用默认配置
  
      Returns:
          包含查询结果的字典
      """
      if not query:
          return {"error": "查询语句不能为空"}
      
      if params is None:
          params = []
      
      try:
          conn = get_connection(db_config)
          cursor = conn.cursor(dictionary=True)
          cursor.execute(query, params)
          
          # 判断是否是需要返回结果集的查询
          query_upper = query.strip().upper()
          if query_upper.startswith("SELECT") or query_upper.startswith("SHOW") or query_upper.startswith("DESCRIBE") or query_upper.startswith("EXPLAIN"):
              results = cursor.fetchall()
              return {
                  "success": True,
                  "rows": results,
                  "row_count": len(results)
              }
          else:
              # 对于非查询性质的SQL，如INSERT, UPDATE, DELETE等
              conn.commit()
              return {
                  "success": True,
                  "affected_rows": cursor.rowcount,
                  "last_insert_id": cursor.lastrowid
              }
      except Error as e:
          error_message = f"执行查询失败: {str(e)}"
          if "Unknown column" in str(e):
              error_message += "\n原因：查询中包含未知的列名"
          elif "Table" in str(e) and "doesn't exist" in str(e):
              error_message += "\n原因：查询的表不存在"
          elif "Syntax error" in str(e):
              error_message += "\n原因：SQL语法错误"
          return {"error": error_message, "query": query}
      except Exception as e:
          return {"error": f"执行过程中发生未知错误: {str(e)}", "query": query}
      finally:
          if 'conn' in locals() and conn.is_connected():
              cursor.close()
              conn.close()

• mysql-mcp.py:121-121 (registration)

  Registration of the 'execute_query' tool via the @mcp.tool() decorator, which registers it with the FastMCP server under the name 'mysql'.

  @mcp.tool()

• mysql-mcp.py:63-119 (helper)

  Helper function that manages database connections with retry logic and merging of default/global/per-request configs.

  def get_connection(db_config=None):
      """获取数据库连接
      
      Args:
          db_config: 数据库连接配置参数，如果为None则使用默认配置
          
      Returns:
          数据库连接对象
      """
      # 如果没有提供配置，先尝试使用全局配置，再使用默认配置
      if db_config is None:
          if GLOBAL_DB_CONFIG is not None:
              db_config = GLOBAL_DB_CONFIG.copy()
          else:
              db_config = DEFAULT_DB_CONFIG.copy()
      else:
          # 合并用户提供的配置和全局/默认配置
          if GLOBAL_DB_CONFIG is not None:
              config = GLOBAL_DB_CONFIG.copy()
          else:
              config = DEFAULT_DB_CONFIG.copy()
          config.update(db_config)
          db_config = config
      
      retry_count = 0
      last_error = None
      max_retries = db_config.get("connect_retry_count", 3)
      
      while retry_count < max_retries:
          try:
              # 创建一个配置字典的副本，移除自定义的配置项
              db_config_copy = db_config.copy()
              db_config_copy.pop("connect_retry_count", None)
              
              # 将connection_timeout转换为mysql.connector需要的connect_timeout参数
              if "connection_timeout" in db_config_copy:
                  db_config_copy["connect_timeout"] = db_config_copy.pop("connection_timeout")
                  
              conn = mysql.connector.connect(**db_config_copy)
              return conn
          except Error as e:
              last_error = e
              retry_count += 1
              if retry_count < max_retries:
                  # 只有在还有重试机会的情况下打印重试信息
                  print(f"第 {retry_count} 次连接失败，正在重试... 错误: {e}")
      
      # 所有重试都失败后，构建详细的错误信息
      error_message = f"数据库连接错误(重试 {retry_count} 次后): {last_error}"
      if "Can't connect to MySQL server" in str(last_error):
          error_message += f"\n无法连接到MySQL服务器，请检查主机 {db_config['host']} 和端口 {db_config['port']} 是否正确"
          error_message += f"\n连接超时时间为 {db_config.get('connection_timeout', 10)} 秒"
      elif "Access denied" in str(last_error):
          error_message += f"\n访问被拒绝，请检查用户名 {db_config['user']} 和密码是否正确"
      elif "Unknown database" in str(last_error):
          error_message += f"\n未知数据库 {db_config['database']}，请确认数据库名称是否正确"
      raise Exception(error_message)