Assistants "Threads" deleter, local Python UI utility - paste any text with thread_id for auto-deletion

Issue:
If you’ve been experimenting with Assistants over the past two years, especially in the “playground”, you may have lots of server-side threads that are orphaned without IDs in a database.

  • there is no UI button to delete them
  • there is no API method to list them for deletion

Solution:

A little Python tool.

The app captures any paste while it is in focus, besides input into the Thread ID box, extracts thread IDs, and immediately deletes. You can copy the platform site’s URL when viewing a thread, copy selection from the page, or use whatever thread listing you have to copy-paste.

What’s seen above in its log ‘console’:

  • pasting some text with multiple threads already deleted;
  • pasting some text without thread IDs (identified by their format);
  • pasting text from selecting within the body of the assistants’ listing of threads in the platform site.

The final console action was from pasting this arbitrary text selection into the app, obtained from within the assistants->thread page itself:

* 4 months ago, Jun 8

* [
Heads or tails? Flip!
thread_cujMP6LmZoea38Tki1nsbW8p
5:32 PM
](https://platform.openai.com/assistants/thread_cujMP6LmZoea38Tki1nsbW8p)
* 5 months ago, May 26

* [
The tool recipient names a

Multiple thread ids are also discovered and dispatched asynchronously for deletion.

  • have an environment OPENAI_API_KEY configured, or input one if the app prompts you that it didn’t find the environment variable
  • no OpenAI SDK module nor even ‘requests’ or ‘httpx’ module is required, just standard library with tkinter as part of the install (typical).
  • start pasting thread IDs for remorseless deleting via API calls.

Coded for Python 3.11+ (walrus operators, builtin generics as type annotation, etc), with gpt-5 doing most of the work.

Save as auto_thread_deleter.pyw or similar, and run.

#!/usr/bin/env python3
"""
OpenAI Assistants thread deleter - tkinter UI using Python standard library.

Operation:
- Paste a thread ID (Ctrl+V / Command+V / Shift+Insert) anywhere in the app window or use right-click -> Paste to trigger deletion.
- Typing a thread ID and pressing Enter in the input box also triggers deletion.
- Input may be a bare ID, a full OpenAI URL containing the ID, or arbitrary text with one or more IDs.
- The entry is cleared immediately and briefly flashes to acknowledge receipt.
- Logs show immediate "deleting {thread_id}" lines followed by success or error responses,
  or a message when no valid thread IDs are found.
- Requests run concurrently on background threads without blocking the UI.
"""

import os
import re
import ssl
import queue
import urllib.request
import urllib.error
import tkinter as tk
from tkinter import scrolledtext, simpledialog
from concurrent.futures import ThreadPoolExecutor


API_BASE = "https://api.openai.com/v1"
TIMEOUT = 60.0

# A thread ID has the form 'thread_' followed by exactly 24 alphanumeric characters.
THREAD_ID_RE = re.compile(r"thread_[A-Za-z0-9]{24}")


def _build_headers() -> dict[str, str]:
    """Build HTTP headers using environment variables."""
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("Set the OPENAI_API_KEY environment variable.")
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "OpenAI-Beta": "assistants=v2",
        "Accept": "application/json",
        "Accept-Encoding": "identity",
    }
    if org := os.getenv("OPENAI_ORG_ID"):
        headers["OpenAI-Organization"] = org
    if proj := os.getenv("OPENAI_PROJECT_ID"):
        headers["OpenAI-Project"] = proj
    return headers


def extract_thread_ids(text: str) -> list[str]:
    """Extract and deduplicate thread IDs found anywhere in the provided text."""
    if not text:
        return []
    candidates = THREAD_ID_RE.findall(text)
    unique = list(dict.fromkeys(candidates))
    return unique


class DeleterWorker:
    """Background worker that deletes threads using urllib."""

    def __init__(self, log_callback, headers: dict[str, str], max_workers: int = 4):
        """
        Args:
            log_callback: Callable that accepts a string; must be thread-safe.
            headers: Fully prepared HTTP headers (includes Authorization).
            max_workers: Maximum concurrent deletion threads.
        """
        self._log_cb = log_callback
        self._headers = headers
        self._executor = ThreadPoolExecutor(
            max_workers=max_workers, thread_name_prefix="deleter"
        )
        self._ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
        try:
            self._ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
        except Exception:
            pass

    def _emit(self, message: str) -> None:
        """Emit a log message through the provided callback."""
        try:
            self._log_cb(message)
        except Exception:
            pass

    def stop(self) -> None:
        """Shut down background threads."""
        try:
            self._executor.shutdown(wait=False, cancel_futures=False)
        except Exception:
            pass

    def schedule_delete(self, thread_id: str) -> None:
        """Schedule a deletion request."""
        if not thread_id:
            return
        self._executor.submit(self._delete_thread, thread_id)

    def _delete_thread(self, thread_id: str) -> None:
        """Perform the DELETE request for the given thread_id."""
        url = f"{API_BASE}/threads/{thread_id}"
        req = urllib.request.Request(url=url, headers=self._headers, method="DELETE")
        try:
            with urllib.request.urlopen(req, timeout=TIMEOUT, context=self._ssl_context) as resp:
                _ = resp.read(0)
                self._emit(f"Deleted Thread {thread_id}.")
        except urllib.error.HTTPError as e:
            body = None
            try:
                body_bytes = e.read()
                body = body_bytes.decode("utf-8", errors="replace")
            except Exception:
                body = None
            self._emit(
                f"Request failed: HTTP {e.code} - {e.reason} for url '{e.geturl()}'"
            )
            if body:
                self._emit("Error response body:\n" + body)
        except urllib.error.URLError as e:
            self._emit(f"Request error: {e.reason}")
        except Exception as e:
            self._emit(f"Unhandled error: {e}")


class AutoThreadDeleterApp:
    """Tkinter app for deleting OpenAI Assistants threads by ID."""

    def __init__(self, root: tk.Tk):
        self.root = root
        self.root.title("OpenAI Assistants - Thread Deleter")
        self.root.geometry("540x380")

        # Top frame for input
        top = tk.Frame(self.root)
        top.pack(side=tk.TOP, fill=tk.X, padx=10, pady=8)

        lbl = tk.Label(top, text="Thread ID:")
        lbl.pack(side=tk.LEFT)

        self.entry = tk.Entry(top, width=36)
        self.entry.pack(side=tk.LEFT, padx=8)
        self.entry.focus_set()

        # Keep original background to allow a flash acknowledgment
        self._entry_bg = self.entry.cget("background")

        # Log area
        self.log_text = scrolledtext.ScrolledText(
            self.root, height=12, wrap=tk.WORD, state=tk.DISABLED
        )
        self.log_text.pack(side=tk.TOP, fill=tk.BOTH, expand=True, padx=10, pady=(0, 10))
        try:
            self.log_text.configure(takefocus=0)
        except Exception:
            pass

        # Right-click paste menu (only Paste)
        self._popup = tk.Menu(self.entry, tearoff=False)
        self._popup.add_command(label="Paste", command=self._on_paste_menu)

        # Bindings for Entry interactions
        self.entry.bind("<Return>", self._on_enter)
        self.entry.bind("<<Paste>>", self._on_paste)
        self.entry.bind("<Control-v>", self._on_paste)
        self.entry.bind("<Control-V>", self._on_paste)
        self.entry.bind("<Shift-Insert>", self._on_paste)
        self.entry.bind("<Command-v>", self._on_paste)
        self.entry.bind("<Command-V>", self._on_paste)
        self.entry.bind("<Button-3>", self._show_popup)
        self.entry.bind("<Button-2>", self._show_popup)

        # Explicit widget-level paste bindings for the console so they fire before Text class bindings
        self.log_text.bind("<<Paste>>", self._on_paste)
        self.log_text.bind("<Control-v>", self._on_paste)
        self.log_text.bind("<Control-V>", self._on_paste)
        self.log_text.bind("<Shift-Insert>", self._on_paste)
        self.log_text.bind("<Command-v>", self._on_paste)
        self.log_text.bind("<Command-V>", self._on_paste)

        # Application-wide paste bindings so Ctrl+V/Command+V anywhere in the window works
        self.root.bind_all("<<Paste>>", self._on_paste, add="+")
        self.root.bind_all("<Control-v>", self._on_paste, add="+")
        self.root.bind_all("<Control-V>", self._on_paste, add="+")
        self.root.bind_all("<Shift-Insert>", self._on_paste, add="+")
        self.root.bind_all("<Command-v>", self._on_paste, add="+")
        self.root.bind_all("<Command-V>", self._on_paste, add="+")
        self.root.bind("<FocusIn>", self._on_toplevel_focus_in, add="+")

        # Thread-safe log queue and pump
        self._log_queue: "queue.Queue[str]" = queue.Queue()
        self.root.after(80, self._drain_log_queue)

        # Headers and worker bootstrap with API key prompt if needed
        headers = self._ensure_headers()
        if headers:
            self.worker = DeleterWorker(self._enqueue_log, headers=headers)
        else:
            self.worker = None
            self._append_log("API key is required to send delete requests. Use Paste after setting the key.")

        # Window close
        self.root.protocol("WM_DELETE_WINDOW", self._on_close)

    def _ensure_headers(self) -> dict[str, str] | None:
        """Ensure headers are available by reading env or prompting for API key if missing."""
        try:
            return _build_headers()
        except ValueError:
            key = simpledialog.askstring(
                "API key required",
                "OPENAI_API_KEY environment variable not found; enter an API key for deleting Assistant threads.",
                show="*",
                parent=self.root,
            )
            if not key:
                return None
            os.environ["OPENAI_API_KEY"] = key.strip()
            try:
                return _build_headers()
            except Exception as e:
                self._append_log(str(e))
                return None

    def _enqueue_log(self, message: str) -> None:
        """Put a message into the UI log queue."""
        self._log_queue.put(message)

    def _drain_log_queue(self) -> None:
        """Flush queued log messages into the UI log area."""
        try:
            while True:
                msg = self._log_queue.get_nowait()
                self._append_log(msg)
        except queue.Empty:
            pass
        finally:
            self.root.after(80, self._drain_log_queue)

    def _append_log(self, text: str) -> None:
        """Append a line to the log and scroll."""
        self.log_text.configure(state=tk.NORMAL)
        self.log_text.insert(tk.END, text + "\n")
        self.log_text.see(tk.END)
        self.log_text.configure(state=tk.DISABLED)

    def _flash_entry(self, color="#e6f7ff", duration_ms=180) -> None:
        """Briefly flash the entry to acknowledge receipt of input."""
        try:
            self.entry.configure(background=color)
            self.root.after(
                duration_ms, lambda: self.entry.configure(background=self._entry_bg)
            )
        except Exception:
            pass

    def _process_input_text(self, text: str) -> None:
        """Parse input text for thread IDs, acknowledge, clear entry, and schedule deletions."""
        self.entry.delete(0, tk.END)
        trimmed = text.strip()
        if not trimmed:
            self.entry.focus_set()
            return
        ids = extract_thread_ids(trimmed)
        if not ids:
            self._append_log("Format 'thread_xxx' not found in paste")
            self.entry.focus_set()
            return
        self._flash_entry()
        if not self.worker:
            self._append_log("API key not configured; cannot delete threads.")
            self.entry.focus_set()
            return
        for tid in ids:
            self._append_log(f"deleting {tid}")
            self.worker.schedule_delete(tid)
        self.entry.focus_set()

    def _on_enter(self, event=None) -> str:
        """Handle Enter key: process the current entry contents."""
        self._process_input_text(self.entry.get())
        return "break"

    def _on_paste(self, event=None) -> str:
        """Handle paste events to trigger immediate deletion."""
        try:
            text = self.root.clipboard_get()
        except tk.TclError:
            text = ""
        self._process_input_text(text)
        return "break"

    def _on_paste_menu(self) -> None:
        """Handle context menu Paste."""
        self._on_paste()

    def _show_popup(self, event) -> str:
        """Show right-click paste-only menu."""
        try:
            self._popup.tk_popup(event.x_root, event.y_root)
        finally:
            self._popup.grab_release()
        return "break"

    def _on_toplevel_focus_in(self, event) -> None:
        """Ensure a child widget owns focus when the window itself gains focus."""
        try:
            if event.widget is self.root and self.root.focus_get() is None:
                self.entry.focus_set()
        except Exception:
            pass

    def _on_close(self) -> None:
        """Stop background worker and close the window."""
        try:
            if self.worker:
                self.worker.stop()
        finally:
            self.root.destroy()


def main() -> None:
    """Entry point."""
    root = tk.Tk()
    app = AutoThreadDeleterApp(root)
    root.mainloop()


if __name__ == "__main__":
    main()

Warning: this will do exactly what it was designed to do: unrecoverable data loss.

1 Like