Log all chat messages to a SQLite database for searchable history and audit. Use when: (1) Building chat history, (2) Auditing conversations, (3) Searching past messages, or (4) User asks to log chats.
npx skills add https://github.com/besoeasy/open-skills --skill chat-logger
Log all incoming and outgoing chat messages to a SQLite database for searchable history, analytics, and auditing. Works with any chat system or agent framework.
No external APIs or services required.
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
session_id TEXT,
sender TEXT NOT NULL, -- 'user', 'assistant', or identifier
content TEXT,
metadata TEXT, -- JSON: channel, tools_used, etc.
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_timestamp ON messages(timestamp);
CREATE INDEX idx_session ON messages(session_id);
CREATE INDEX idx_sender ON messages(sender);
-- Automatic purge: delete records older than 1 year
DELETE FROM messages WHERE created_at < datetime('now', '-1 year');
Fields:
id - Auto-incrementing primary keytimestamp - ISO 8601 timestamp of the messagesession_id - Optional session/conversation identifiersender - Message sender ('user', 'assistant', or custom ID)content - Message text contentmetadata - JSON field for additional data (channel, tools, context)created_at - Database insertion timestampInitialize database:
import sqlite3
from datetime import datetime
from pathlib import Path
import json
# Configure database path
DB_PATH = Path.home() / ".chat_logs" / "messages.db"
def init_db():
"""Initialize database and create tables."""
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(DB_PATH))
conn.execute("""
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
session_id TEXT,
sender TEXT NOT NULL,
content TEXT,
metadata TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON messages(timestamp)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_session ON messages(session_id)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_sender ON messages(sender)")
conn.commit()
conn.close()
def purge_old_messages():
"""Delete messages older than 1 year to keep the database size sane."""
conn = sqlite3.connect(str(DB_PATH))
conn.execute("DELETE FROM messages WHERE created_at < datetime('now', '-1 year')")
conn.commit()
conn.close()
# Initialize on import and purge old records
init_db()
purge_old_messages()
Log messages:
def log_message(sender: str, content: str, session_id: str = None, metadata: dict = None):
"""Log a chat message to the database."""
conn = sqlite3.connect(str(DB_PATH))
try:
conn.execute(
"""INSERT INTO messages (timestamp, session_id, sender, content, metadata)
VALUES (?, ?, ?, ?, ?)""",
(
datetime.utcnow().isoformat(),
session_id,
sender,
content[:10000] if content else None, # Truncate long messages
json.dumps(metadata) if metadata else None
)
)
conn.commit()
finally:
conn.close()
# Usage examples
log_message("user", "Hello, how are you?", session_id="session_123")
log_message("assistant", "I'm doing well, thank you!", session_id="session_123")
log_message("user", "Help me deploy a website", session_id="session_456",
metadata={"channel": "web", "ip": "192.168.1.1"})
Query messages:
def get_recent_messages(limit: int = 50):
"""Get recent messages."""
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
cursor = conn.execute(
"SELECT * FROM messages ORDER BY timestamp DESC LIMIT ?",
(limit,)
)
results = cursor.fetchall()
conn.close()
return results
def get_session_history(session_id: str):
"""Get all messages from a specific session."""
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
cursor = conn.execute(
"SELECT * FROM messages WHERE session_id = ? ORDER BY timestamp ASC",
(session_id,)
)
results = cursor.fetchall()
conn.close()
return results
def search_messages(query: str, limit: int = 20):
"""Search message content."""
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
cursor = conn.execute(
"SELECT * FROM messages WHERE content LIKE ? ORDER BY timestamp DESC LIMIT ?",
(f"%{query}%", limit)
)
results = cursor.fetchall()
conn.close()
return results
# Usage
messages = get_recent_messages(10)
for msg in messages:
print(f"[{msg['timestamp']}] {msg['sender']}: {msg['content'][:100]}")
# Search
results = search_messages("deploy website")
print(f"Found {len(results)} messages about deploying websites")
import sqlite3 from "sqlite3";
import { promisify } from "util";
import path from "path";
import os from "os";
const DB_PATH = path.join(os.homedir(), ".chat_logs", "messages.db");
// Initialize database
const db = new sqlite3.Database(DB_PATH);
const run = promisify(db.run.bind(db));
const all = promisify(db.all.bind(db));
await run(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
session_id TEXT,
sender TEXT NOT NULL,
content TEXT,
metadata TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
`);
// Log message
async function logMessage(sender, content, sessionId = null, metadata = null) {
await run(
`INSERT INTO messages (timestamp, session_id, sender, content, metadata)
VALUES (?, ?, ?, ?, ?)`,
[
new Date().toISOString(),
sessionId,
sender,
content,
metadata ? JSON.stringify(metadata) : null,
]
);
}
// Query messages
async function getRecentMessages(limit = 50) {
return await all(
`SELECT * FROM messages ORDER BY timestamp DESC LIMIT ?`,
[limit]
);
}
// Usage
await logMessage("user", "Hello!", "session_123");
await logMessage("assistant", "Hi there!", "session_123");
const messages = await getRecentMessages(10);
console.log(messages);
# View recent messages
sqlite3 ~/.chat_logs/messages.db "SELECT timestamp, sender, substr(content, 1, 80) FROM messages ORDER BY timestamp DESC LIMIT 20"
# Search for specific content
sqlite3 ~/.chat_logs/messages.db "SELECT * FROM messages WHERE content LIKE '%docker%' ORDER BY timestamp DESC"
# Count messages by sender
sqlite3 ~/.chat_logs/messages.db "SELECT sender, COUNT(*) as count FROM messages GROUP BY sender"
# Export session to JSON
sqlite3 -json ~/.chat_logs/messages.db "SELECT * FROM messages WHERE session_id='session_123' ORDER BY timestamp ASC" > conversation.json
class ChatLogger:
"""Simple chat logger that can wrap any chat system."""
def __init__(self, db_path: str = None):
self.db_path = db_path or str(Path.home() / ".chat_logs" / "messages.db")
self._init_db()
def _init_db(self):
# Same as init_db() above
pass
def log_user_message(self, content: str, session_id: str = None, **metadata):
return log_message("user", content, session_id, metadata)
def log_assistant_message(self, content: str, session_id: str = None, **metadata):
return log_message("assistant", content, session_id, metadata)
def get_conversation(self, session_id: str):
return get_session_history(session_id)
# Usage in any chat system
logger = ChatLogger()
# In your chat handler
def handle_message(user_input, session_id):
logger.log_user_message(user_input, session_id=session_id)
# Process message...
response = generate_response(user_input)
logger.log_assistant_message(response, session_id=session_id)
return response
def with_logging(session_id: str = None):
"""Decorator to automatically log chat interactions."""
def decorator(func):
def wrapper(user_message, *args, **kwargs):
# Log user message
log_message("user", user_message, session_id=session_id)
# Call original function
response = func(user_message, *args, **kwargs)
# Log assistant response
log_message("assistant", response, session_id=session_id)
return response
return wrapper
return decorator
# Usage
@with_logging(session_id="session_123")
def chat_handler(message):
return f"You said: {message}"
You have chat logging capability. All conversations are logged to a SQLite database.
When user asks to:
- Search past conversations
- Find specific messages
- Review conversation history
- Export chat logs
Use the SQLite database at ~/.chat_logs/messages.db with this schema:
- messages table (id, timestamp, session_id, sender, content, metadata)
Query examples:
1. Recent history: SELECT * FROM messages ORDER BY timestamp DESC LIMIT 50
2. Search content: SELECT * FROM messages WHERE content LIKE '%keyword%'
3. Session history: SELECT * FROM messages WHERE session_id = ? ORDER BY timestamp ASC
Always use SQL queries to retrieve information and present results clearly to the user.
Database locked error:
conn.close()Large database file:
VACUUM to compact: sqlite3 messages.db "VACUUM"Query performance:
Efficient database search tool for bioRxiv preprint server. Use this skill when searching for life sciences preprints by keywords, authors, date ranges, or categories, retrieving paper metadata, downloading PDFs, or conducting literature reviews.
Access BRENDA enzyme database via SOAP API. Retrieve kinetic parameters (Km, kcat), reaction equations, organism data, and substrate-specific enzyme information for biochemical research and metabolic pathway analysis.
Access ClinPGx pharmacogenomics data (successor to PharmGKB). Query gene-drug interactions, CPIC guidelines, allele functions, for precision medicine and genotype-guided dosing decisions.
Query NCBI ClinVar for variant clinical significance. Search by gene/position, interpret pathogenicity classifications, access via E-utilities API or FTP, annotate VCFs, for genomic medicine.
Access COSMIC cancer mutation database. Query somatic mutations, Cancer Gene Census, mutational signatures, gene fusions, for cancer research and precision oncology. Requires authentication.
Query Ensembl genome database REST API for 250+ species. Gene lookups, sequence retrieval, variant analysis, comparative genomics, orthologs, VEP predictions, for genomic research.
Query openFDA API for drugs, devices, adverse events, recalls, regulatory submissions (510k, PMA), substance identification (UNII), for FDA regulatory data analysis and safety research.
Query NCBI Gene via E-utilities/Datasets API. Search by symbol/ID, retrieve gene info (RefSeqs, GO, locations, phenotypes), batch lookups, for gene annotation and functional analysis.
Take besoeasy/chat-logger from the repository into ~/.claude/skills for personal
use, or into .claude/skills inside a project.
The agent identifies a skill by the name field in its header. Two skills with the
same name cannot sit side by side — one of them will be ignored.