List Managed Threads
list_threadsRead-only
Retrieve saved conversation threads with IDs, names, and timestamps to continue discussions using the ask_assistant_in_thread tool.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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}") - 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() - 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()