Persistent Memory for Claude Agents: SQLite GUI Build (2026)

When you first add Claude to a desktop tool, the conversation disappears the moment the app closes. The next time you launch it, the agent starts from a blank slate, forcing you to copy‑paste old logs or re‑prompt the model. That friction kills productivity and makes long‑running tasks—like file‑system clean‑up or multi‑step code generation—practically impossible.

The real problem isn’t Claude itself; it’s the missing bridge between the API’s stateless calls and your local UI. A lightweight SQLite store, tightly coupled to the event loop, can give the agent a durable “brain” without pulling in heavyweight cloud databases or reinventing a custom file format.

—

Persistent memory for Claude agents is implemented by integrating a local SQLite database with the agent’s architecture. In a 2026 desktop application using Python and a GUI framework like PyQt, a dedicated memory module handles storing and querying full chat history, tool calls, and session state, enabling the agent to maintain context across restarts and conversations.

System architecture & workflow

At a high level the app consists of three independent layers:

1. **GUI layer** – PyQt6 widgets that emit user events (send, cancel, edit). The UI runs on the main thread.
2. **Agent layer** – An async coroutine that talks to the Anthropic Claude API, receives streaming tokens, and queries the memory module for context.
3. **Memory layer** – Synchronous SQLite CRUD helper wrapped in an `asyncio.to_thread` call to keep DB I/O off the UI thread.

The data flow is simple:

User → GUI (send button) → Agent coroutine → Memory.read_context() → Claude API → stream → Memory.append_message() → GUI.update()

flowchart LR
    A[User Input] --> B[Qt GUI Event]
    B --> C[Agent Async Loop]
    C --> D[Memory Module (SQLite)]
    C --> E[Anthropic Claude API]
    E --> F[Streaming Tokens]
    F --> G[GUI Update]
    D --> C

*Why this split?* Keeping DB ops off the main thread avoids the dreaded “application not responding” state, while separating concerns makes it trivial to swap SQLite for PostgreSQL later (see my post on [Multi‑tenant Database Schema for AI Agents: 4 Ways](https://nileshblog.tech/?p=7007)).

—

Step‑by‑step implementation

1. Project scaffolding

# Create isolated environment (optional but recommended)
uv venv .venv
source .venv/bin/activate

# Install deps
pip install "anthropic>=0.20" "pydantic>=2.5" "PyQt6>=6.5" "aiosqlite>=0.20"

Create the following file layout:

claude_gui/
│─ main.py
│─ memory.py
│─ agent.py
│─ ui.py
│─ schema.sql
│─ config.py

2. SQLite schema (`schema.sql`)

-- version: 1.0
BEGIN TRANSACTION;

CREATE TABLE IF NOT EXISTS conversations (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    convo_id INTEGER NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    role TEXT CHECK(role IN ('user','assistant','system')) NOT NULL,
    content TEXT NOT NULL,
    tool_calls JSON,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    token_count INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_msg_convo_ts ON messages(convo_id, created_at);
COMMIT;

**Why these columns?** `tool_calls` stores JSON blobs produced when Claude invokes the Computer Use API, preserving the exact request that led to an action. `token_count` helps you enforce the model’s token window without re‑tokenizing every time.

3. Memory module (`memory.py`)

# memory.py
# Requires: Python 3.12+, aiosqlite 0.20+

import aiosqlite
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Any

DB_PATH = Path("claude_memory.db")
SCHEMA_PATH = Path("schema.sql")

class SQLiteMemory:
    def __init__(self, db_path: Path = DB_PATH):
        self.db_path = db_path

    async def init(self) -> None:
        """Create DB and tables if they don't exist."""
        async with aiosqlite.connect(self.db_path) as db:
            async with db.executescript(SCHEMA_PATH.read_text()) as cur:
                await cur.fetchall()  # force execution
            await db.commit()

    async def new_conversation(self, name: str) -> int:
        async with aiosqlite.connect(self.db_path) as db:
            cur = await db.execute(
                "INSERT INTO conversations (name) VALUES (?) RETURNING id;",
                (name,),
            )
            row = await cur.fetchone()
            await db.commit()
            return row[0]

    async def add_message(
        self,
        convo_id: int,
        role: str,
        content: str,
        token_count: int,
        tool_calls: List[Dict[str, Any]] | None = None,
    ) -> int:
        async with aiosqlite.connect(self.db_path) as db:
            cur = await db.execute(
                """
                INSERT INTO messages (convo_id, role, content, token_count, tool_calls)
                VALUES (?, ?, ?, ?, ?)
                RETURNING id;
                """,
                (
                    convo_id,
                    role,
                    content,
                    token_count,
                    json.dumps(tool_calls) if tool_calls else None,
                ),
            )
            row = await cur.fetchone()
            await db.commit()
            return row[0]

    async def get_recent_context(
        self,
        convo_id: int,
        max_tokens: int = 4000,
        lookback: timedelta = timedelta(hours=12),
    ) -> List[Dict[str, Any]]:
        """
        Pull the most recent messages that fit within max_tokens.
        Older messages are dropped in favor of recency and relevance.
        """
        async with aiosqlite.connect(self.db_path) as db:
            query = """
                SELECT role, content, tool_calls, token_count, created_at
                FROM messages
                WHERE convo_id = ?
                  AND created_at >= datetime('now', ?)
                ORDER BY created_at DESC;
            """
            cur = await db.execute(query, (convo_id, f"-{lookback.total_seconds()} seconds"))
            rows = await cur.fetchall()

        # Accumulate until token budget is exhausted (descending order -> reverse later)
        context = []
        used = 0
        for role, content, tool_json, token_cnt, created_at in rows:
            if used + token_cnt > max_tokens:
                break
            entry = {"role": role, "content": content}
            if tool_json:
                entry["tool_calls"] = json.loads(tool_json)
            context.append(entry)
            used += token_cnt

        # Reverse to chronological order for Claude
        return list(reversed(context))

**My take:** Wrapping `aiosqlite` in an `async` class avoids the need for `run_in_executor` hacks; the library already does the thread‑offload for you. If you ever move to PostgreSQL, the public methods stay identical.

4. Agent layer (`agent.py`)

# agent.py
# Requires: anthropic 0.20+, pydantic 2.5

import asyncio
import os
import json
from anthropic import Anthropic, AsyncClient
from anthropic.types import Message, MessageStreamEvent
from pydantic import BaseModel, Field
from typing import List, Dict, Any

from memory import SQLiteMemory

# Load API key from env (never hard‑code!)
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
if not ANTHROPIC_API_KEY:
    raise RuntimeError("Set ANTHROPIC_API_KEY environment variable")

class ClaudeAgent:
    def __init__(self, memory: SQLiteMemory, model: str = "claude-3-5-sonnet-20240620"):
        self.memory = memory
        self.client = AsyncClient(api_key=ANTHROPIC_API_KEY)
        self.model = model
        self.convo_id: int | None = None

    async def start_conversation(self, name: str = "default"):
        self.convo_id = await self.memory.new_conversation(name)

    async def _build_messages(self) -> List[Dict[str, Any]]:
        """Fetch recent context and prepend system prompt."""
        context = await self.memory.get_recent_context(self.convo_id)
        system_prompt = {
            "role": "system",
            "content": "You are a helpful assistant that can also use the Computer Use API. Keep your responses concise.",
        }
        return [system_prompt] + context

    async def send_user_message(self, text: str) -> None:
        # Token estimation (very rough, 4 chars per token)
        token_cnt = max(1, len(text) // 4)
        await self.memory.add_message(
            convo_id=self.convo_id,
            role="user",
            content=text,
            token_count=token_cnt,
        )
        asyncio.create_task(self._query_claude_stream(text))

    async def _query_claude_stream(self, user_text: str) -> None:
        try:
            messages = await self._build_messages()
            async with self.client.messages.stream(
                model=self.model,
                max_tokens=1024,
                temperature=0.7,
                system=None,
                messages=messages,
            ) as stream:
                # Buffer to count tokens
                assistant_content = ""
                tool_calls: List[Dict[str, Any]] = []
                async for event in stream:
                    if isinstance(event, MessageStreamEvent):
                        delta = event.delta
                        if delta.type == "content_block_delta":
                            assistant_content += delta.text
                            # Forward to UI (hooked later)
                        elif delta.type == "tool_use":
                            tool_calls.append(
                                {
                                    "id": delta.id,
                                    "name": delta.name,
                                    "input": delta.input,
                                }
                            )
                # Store assistant reply
                token_cnt = max(1, len(assistant_content) // 4)
                await self.memory.add_message(
                    convo_id=self.convo_id,
                    role="assistant",
                    content=assistant_content,
                    token_count=token_cnt,
                    tool_calls=tool_calls if tool_calls else None,
                )
                # Signal UI to render final assistant message
                # (will be wired in ui.py)
        except Exception as exc:
            # Central error handling – UI will show a toast
            print(f"[Agent] error: {exc}")
            # Re‑raise so UI can catch if needed
            raise

**Why async streaming?** Claude can return partial tokens for up to a minute. Feeding each token directly to the UI gives the user immediate feedback, and because the coroutine runs in the background the Qt event loop stays snappy.

5. UI layer (`ui.py`)

# ui.py
# Requires: PyQt6>=6.5

import sys
import asyncio
from PyQt6.QtWidgets import (
    QApplication,
    QWidget,
    QVBoxLayout,
    QTextEdit,
    QPushButton,
    QLabel,
    QHBoxLayout,
)
from PyQt6.QtCore import Qt, QTimer, QObject, pyqtSignal

from agent import ClaudeAgent
from memory import SQLiteMemory

class AsyncSignal(QObject):
    """Bridge async coroutine results to Qt slots."""
    new_chunk = pyqtSignal(str)
    finished = pyqtSignal()

class ChatWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Claude Desktop – Persistent Memory")
        self.resize(720, 540)

        self.layout = QVBoxLayout(self)

        self.log = QTextEdit(readOnly=True)
        self.log.setMinimumHeight(400)
        self.layout.addWidget(self.log)

        self.input = QTextEdit()
        self.input.setFixedHeight(80)
        self.layout.addWidget(self.input)

        btn_box = QHBoxLayout()
        self.send_btn = QPushButton("Send")
        self.clear_btn = QPushButton("Clear")
        btn_box.addWidget(self.send_btn)
        btn_box.addWidget(self.clear_btn)
        self.layout.addLayout(btn_box)

        self.status = QLabel("")
        self.layout.addWidget(self.status)

        # Async bridge
        self.signals = AsyncSignal()
        self.signals.new_chunk.connect(self.append_chunk)
        self.signals.finished.connect(self.response_done)

        # Setup agent & memory
        self.memory = SQLiteMemory()
        self.agent = ClaudeAgent(self.memory)

        # Qt → asyncio integration
        self.loop = asyncio.get_event_loop()
        self.loop.create_task(self.agent.start_conversation("Desktop Session"))

        # Connect UI events
        self.send_btn.clicked.connect(self.handle_send)
        self.clear_btn.clicked.connect(self.handle_clear)

        # Periodic heartbeat to keep Qt responsive
        self.timer = QTimer()
        self.timer.timeout.connect(lambda: None)
        self.timer.start(100)

    def append_chunk(self, text: str):
        self.log.moveCursor(Qt.TextCursor.MoveOperation.End)
        self.log.insertPlainText(text)
        self.log.moveCursor(Qt.TextCursor.MoveOperation.End)

    def response_done(self):
        self.status.setText("✅ Response complete")
        self.send_btn.setEnabled(True)

    def handle_send(self):
        user_msg = self.input.toPlainText().strip()
        if not user_msg:
            return
        self.log.append(f"<User>: {user_msg}")
        self.input.clear()
        self.send_btn.setEnabled(False)
        self.status.setText("⏳ Waiting for Claude…")
        asyncio.create_task(self.agent.send_user_message(user_msg))

    def handle_clear(self):
        self.log.clear()
        self.status.setText("")

def main():
    app = QApplication(sys.argv)
    win = ChatWindow()
    win.show()
    sys.exit(app.exec())

if __name__ == "__main__":
    main()

Key parts:

• `AsyncSignal` translates async updates (`new_chunk`) into Qt slots, so the UI never blocks.
• The tiny `QTimer` heartbeat keeps the Qt event loop alive while `asyncio` runs in the background (a pattern described in detail in my post about [Event Sourcing vs CRUD for AI Agents: 5 Benefits](https://nileshblog.tech/?p=6970)).

6. Wiring streaming output into the UI

To actually push each token chunk from `ClaudeAgent` to `ChatWindow`, we add a tiny hook in `agent.py`:

# At the top of agent.py
from typing import Callable

# In ClaudeAgent __init__
    def __init__(self, memory: SQLiteMemory, model: str = "...", on_chunk: Callable[[str], None] | None = None):
        ...
        self.on_chunk = on_chunk

# Inside _query_claude_stream, after assembling each delta:
                        if delta.type == "content_block_delta":
                            assistant_content += delta.text
                            if self.on_chunk:
                                # Schedule Qt signal on the main thread
                                asyncio.get_event_loop().call_soon_threadsafe(self.on_chunk, delta.text)

And in `ui.py` when constructing the agent:

self.agent = ClaudeAgent(
    self.memory,
    on_chunk=self.signals.new_chunk.emit,
)

Now every token appears instantly in the chat log.

—

Pro tips & warnings