MCP Simple OpenAI Assistant

Instructions

Implementation Reference

• mcp_simple_openai_assistant/app.py:65-91 (handler)

  MCP tool handler and registration for 'list_threads'. Calls manager.list_threads() and formats the thread list into a readable string output.

  @app.tool(annotations={"title": "List Managed Threads", "readOnlyHint": True})
  async def list_threads() -> str:
      """
      Lists all locally saved conversation threads from the database.
      Returns a list of threads with their ID, name, description, and last used time.
      The thread ID can be used in the ask_assistant_in_thread tool to specify this thread to be continued.
      """
      if not manager:
          raise ToolError("AssistantManager not initialized.")
      try:
          threads = manager.list_threads()
          if not threads:
              return "No managed threads found."
  
          thread_list = [
              dedent(f"""
              Thread ID: {t['thread_id']}
              Name: {t['name']}
              Description: {t['description']}
              Last Used: {t['last_used_at']}
              """)
              for t in threads
          ]
          return "Managed Threads:\\n\\n" + "\\n---\\n".join(thread_list)
      except Exception as e:
          raise ToolError(f"Failed to list threads: {e}")

• mcp_simple_openai_assistant/assistant_manager.py:59-62 (helper)

  Helper method in AssistantManager that delegates the list_threads call to the ThreadStore instance.

  def list_threads(self) -> list[sqlite3.Row]:
      """List all managed threads from the local database."""
      return self.thread_store.list_threads()

• mcp_simple_openai_assistant/thread_store.py:73-84 (helper)

  Core data access helper in ThreadStore that executes SQL query to fetch all threads ordered by last_used_at DESC.

  def list_threads(self) -> list[sqlite3.Row]:
      """Retrieves all thread records from the database.
  
      Returns:
          A list of rows, where each row is a dictionary-like object
          representing a thread.
      """
      conn = self._get_connection()
      cursor = conn.cursor()
      cursor.execute("SELECT * FROM threads ORDER BY last_used_at DESC")
      return cursor.fetchall()