Как я могу открыть и закрыть поток с помощью Python Socket Server?

Ах. Я сначала пропустил, что происходит...

Проблема в том, что когда вы назначаете _bottle_thread, вы не назначаете его глобальной версии (или даже его версии класса).

Будет поучительно запустить этот короткий пример:

_bottle_thread = "Global"

class C:
    _bottle_thread = "Class"

    def meth(self):
        _bottle_thread = "Method"
        print(f"Method {_bottle_thread}")    

c = C()
c.meth()

print(f"Global: {_bottle_thread}")
print(f"Class: {c._bottle_thread}")

Результат:

Method Method
Global: Global
Class: Class

В результате вы получаете три разные версии _bottle_thread из-за правил области видимости Python: глобальную, которую вы получили, назначив ее вне всех других областей, классовую, которую вы получили, назначив ее в определении класса, и локальный, который вы получили, присвоив ему внутри метода. (Обратите внимание, что невозможно напечатать локальный _bottle_thread, назначенный внутри meth - то есть снаружи метода - поскольку его счетчик ссылок становится равным 0, как только метод завершается, и затем он будет уничтожен.)

Вам нужно будет использовать ключевое слово global (https://docs.python.org/3/reference/simple_stmts.html#the-global-statement) в каждой области, в которой вы хотите, чтобы глобальный объект был виден:

def meth(self):
    global _bottle_thread
    _bottle_thread = "Method"

Лучшее решение — просто сохранить переменную _bottle_thread как переменную экземпляра:

class CalibrationServer(socketserver.BaseRequestHandler):
    ...
    def __init__(self):
        # (Not strictly necessary but considered best practice to 
        # initialize instance variables in the constructor)
        self._bottle_thread = None 
        

    def handle_start(self):
        ...
        self._bottle_thread = threading.Thread(target=start, args=(bottle,))
        ...
    def handle_stop(self):
        if self._bottle_thread and self._bottle_thread.is_alive():
             ...

Сохраняйте поток бутылки как состояние сервера и управляйте потоком с помощью флага в состоянии потока.

Также обратите внимание, что TCP не является протоколом на основе сообщений. Это просто поток байтов, поэтому вам необходимо предоставить протокол, чтобы знать, где начинаются и заканчиваются сообщения. socketserver.StreamRequestHandler предоставляет файловые оболочки rfile и wfile для потока, которые буферизуют данные и позволяют .readline() работать. Я использовал их ниже, чтобы рассматривать сообщения как завершающиеся новой строкой:

сервер.py

import threading
import socketserver
import time

class Bottle(threading.Thread):

    def __init__(self, bottle):
        super().__init__(daemon=True)  # So thread dies on main thread halt (ctrl-C)
        # Thread state variables
        self.bottle = bottle
        self.running = True

    def run(self):
        while self.running:
            print(self.bottle)
            time.sleep(.5)

class Server(socketserver.TCPServer):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Server state variable
        self.bottle_thread = None

class Handler(socketserver.StreamRequestHandler):

    def handle(self):
        line = self.rfile.readline()
        if not line: return  # client closed socket
        self.request = line.strip().decode().split(' ')
        method = self.request[0]
        if method == 'start':
            self.handle_start()
        elif method == 'stop':
            self.handle_stop()
        else:
            self.response = 'ERROR: Unknown request'

        self.wfile.write(self.response.encode() + b'\n')
        self.wfile.flush()  # makefile wrappers are buffered, flush ensures all is sent

    def handle_start(self):
        if self.server.bottle_thread is None:
            if len(self.request) > 1:
                bottle = self.request[1]
                self.server.bottle_thread = Bottle(bottle)
                self.server.bottle_thread.start()
                self.response = 'Thread running'
            else:
                self.response = 'ERROR: missing parameter'
        else:
            self.response = 'Thread already running'

    def handle_stop(self):
        if self.server.bottle_thread is not None:
            self.server.bottle_thread.running = False
            self.server.bottle_thread.join()  # Wait for the thread to finish
            self.server.bottle_thread = None
            self.response = 'Thread stopped'
        else:
            self.response = 'Thread not running'

if __name__ == '__main__':
    HOST, PORT = '', 5000
    with Server((HOST, PORT), Handler) as server:
        server.serve_forever()

client.py — здесь я вручную обернул сокет в файловые объекты:

import socket
import sys

HOST, PORT = 'localhost', 5000
data = ' '.join(sys.argv[1:])

with socket.socket() as sock:
    sock.connect((HOST, PORT))
    with (sock.makefile('r', encoding='utf8') as rfile,
          sock.makefile('w', encoding='utf8') as wfile
    ):
        wfile.write(data + '\n')
        wfile.flush()
        response = rfile.readline()

print(f'Sent:     {data}')
print(f'Received: {response}')

Выход:

C:\>test stop
Sent:     stop
Received: Thread not running


C:\>test blah
Sent:     blah
Received: ERROR: Unknown request


C:\>test start
Sent:     start
Received: ERROR: missing parameter


C:\>test start abc
Sent:     start abc
Received: Thread running


C:\>test start
Sent:     start
Received: Thread already running


C:\>test stop
Sent:     stop
Received: Thread stopped