Google Workspace MCP Server

""" Comprehensive Personal Memory System Stores everything about the user - who they are, what they do, how they do it """ import sqlite3 import json import os from datetime import datetime from pathlib import Path # Database file location - centralized in pbdb folder MEMORY_DB_PATH = os.path.join(os.path.expanduser('~'), 'Documents', 'pbdb', 'mcp', 'personal_memory.db') def get_connection(): """Get database connection with row factory""" conn = sqlite3.connect(MEMORY_DB_PATH) conn.row_factory = sqlite3.Row return conn def init_memory_database(): """Initialize the comprehensive memory database""" conn = get_connection() cursor = conn.cursor() # Core identity - who you are cursor.execute(''' CREATE TABLE IF NOT EXISTS identity ( id INTEGER PRIMARY KEY, category TEXT NOT NULL, key TEXT NOT NULL, value TEXT, confidence TEXT DEFAULT 'confirmed', source TEXT, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(category, key) ) ''') # Behaviors - how you do things cursor.execute(''' CREATE TABLE IF NOT EXISTS behaviors ( id INTEGER PRIMARY KEY AUTOINCREMENT, behavior_type TEXT NOT NULL, description TEXT NOT NULL, context TEXT, frequency TEXT, importance TEXT DEFAULT 'medium', examples TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Preferences - what you like/dislike cursor.execute(''' CREATE TABLE IF NOT EXISTS preferences ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT NOT NULL, preference TEXT NOT NULL, value TEXT, reason TEXT, priority INTEGER DEFAULT 5, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Patterns - recurring things you do cursor.execute(''' CREATE TABLE IF NOT EXISTS patterns ( id INTEGER PRIMARY KEY AUTOINCREMENT, pattern_type TEXT NOT NULL, description TEXT NOT NULL, trigger TEXT, frequency TEXT, time_context TEXT, notes TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Decisions - how you make decisions cursor.execute(''' CREATE TABLE IF NOT EXISTS decisions ( id INTEGER PRIMARY KEY AUTOINCREMENT, decision_area TEXT NOT NULL, approach TEXT NOT NULL, factors TEXT, examples TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Goals - what you want to achieve cursor.execute(''' CREATE TABLE IF NOT EXISTS goals ( id INTEGER PRIMARY KEY AUTOINCREMENT, goal_type TEXT NOT NULL, description TEXT NOT NULL, timeframe TEXT, priority TEXT DEFAULT 'medium', status TEXT DEFAULT 'active', milestones TEXT, progress TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Interactions - things you've done/discussed cursor.execute(''' CREATE TABLE IF NOT EXISTS interactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, interaction_type TEXT NOT NULL, summary TEXT NOT NULL, details TEXT, outcome TEXT, learnings TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Facts - factual information about you cursor.execute(''' CREATE TABLE IF NOT EXISTS facts ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT NOT NULL, fact TEXT NOT NULL, context TEXT, verified BOOLEAN DEFAULT 1, source TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Relationships - people in your life/network cursor.execute(''' CREATE TABLE IF NOT EXISTS relationships ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, relationship_type TEXT, context TEXT, importance TEXT DEFAULT 'medium', notes TEXT, last_interaction TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Work style - how you approach work cursor.execute(''' CREATE TABLE IF NOT EXISTS work_style ( id INTEGER PRIMARY KEY AUTOINCREMENT, aspect TEXT NOT NULL, description TEXT NOT NULL, examples TEXT, tips_for_ai TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Communication style cursor.execute(''' CREATE TABLE IF NOT EXISTS communication_style ( id INTEGER PRIMARY KEY AUTOINCREMENT, aspect TEXT NOT NULL, preference TEXT NOT NULL, examples TEXT, avoid TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Learning log - things learned about user over time cursor.execute(''' CREATE TABLE IF NOT EXISTS learning_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT NOT NULL, learned TEXT NOT NULL, context TEXT, confidence TEXT DEFAULT 'medium', timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Quick memory - for any ad-hoc information cursor.execute(''' CREATE TABLE IF NOT EXISTS quick_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT NOT NULL, content TEXT NOT NULL, tags TEXT, importance TEXT DEFAULT 'normal', expires_at TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() return {'success': True, 'message': 'Memory database initialized'} # Initialize on import init_memory_database() # ==================== IDENTITY FUNCTIONS ==================== def set_identity(category, key, value, confidence='confirmed', source=None): """Set identity information (who you are) Categories: personal, professional, contact, social, preferences """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT OR REPLACE INTO identity (category, key, value, confidence, source, updated_at) VALUES (?, ?, ?, ?, ?, ?) ''', (category, key, value, confidence, source, datetime.now())) conn.commit() conn.close() return {'success': True, 'category': category, 'key': key, 'value': value} except Exception as e: return {'success': False, 'error': str(e)} def get_identity(category=None): """Get identity information""" try: conn = get_connection() cursor = conn.cursor() if category: cursor.execute('SELECT * FROM identity WHERE category = ?', (category,)) else: cursor.execute('SELECT * FROM identity ORDER BY category, key') rows = cursor.fetchall() conn.close() return {'success': True, 'identity': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== BEHAVIOR FUNCTIONS ==================== def add_behavior(behavior_type, description, context=None, frequency=None, importance='medium', examples=None): """Add a behavior pattern (how you do things) Types: work, communication, decision_making, problem_solving, learning, social """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO behaviors (behavior_type, description, context, frequency, importance, examples) VALUES (?, ?, ?, ?, ?, ?) ''', (behavior_type, description, context, frequency, importance, examples)) conn.commit() conn.close() return {'success': True, 'message': f'Behavior added: {description[:50]}...'} except Exception as e: return {'success': False, 'error': str(e)} def get_behaviors(behavior_type=None): """Get behavior patterns""" try: conn = get_connection() cursor = conn.cursor() if behavior_type: cursor.execute('SELECT * FROM behaviors WHERE behavior_type = ?', (behavior_type,)) else: cursor.execute('SELECT * FROM behaviors ORDER BY importance DESC, behavior_type') rows = cursor.fetchall() conn.close() return {'success': True, 'behaviors': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== PREFERENCE FUNCTIONS ==================== def add_preference(category, preference, value=None, reason=None, priority=5): """Add a preference Categories: tools, communication, work, lifestyle, technology, food, etc. """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO preferences (category, preference, value, reason, priority) VALUES (?, ?, ?, ?, ?) ''', (category, preference, value, reason, priority)) conn.commit() conn.close() return {'success': True, 'message': f'Preference added: {preference}'} except Exception as e: return {'success': False, 'error': str(e)} def get_preferences(category=None): """Get preferences""" try: conn = get_connection() cursor = conn.cursor() if category: cursor.execute('SELECT * FROM preferences WHERE category = ? ORDER BY priority DESC', (category,)) else: cursor.execute('SELECT * FROM preferences ORDER BY category, priority DESC') rows = cursor.fetchall() conn.close() return {'success': True, 'preferences': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== PATTERN FUNCTIONS ==================== def add_pattern(pattern_type, description, trigger=None, frequency=None, time_context=None, notes=None): """Add a recurring pattern Types: daily, weekly, work, personal, seasonal, situational """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO patterns (pattern_type, description, trigger, frequency, time_context, notes) VALUES (?, ?, ?, ?, ?, ?) ''', (pattern_type, description, trigger, frequency, time_context, notes)) conn.commit() conn.close() return {'success': True, 'message': f'Pattern added: {description[:50]}...'} except Exception as e: return {'success': False, 'error': str(e)} def get_patterns(pattern_type=None): """Get patterns""" try: conn = get_connection() cursor = conn.cursor() if pattern_type: cursor.execute('SELECT * FROM patterns WHERE pattern_type = ?', (pattern_type,)) else: cursor.execute('SELECT * FROM patterns ORDER BY pattern_type') rows = cursor.fetchall() conn.close() return {'success': True, 'patterns': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== GOAL FUNCTIONS ==================== def add_goal(goal_type, description, timeframe=None, priority='medium', milestones=None): """Add a goal Types: career, learning, health, financial, personal, project """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO goals (goal_type, description, timeframe, priority, milestones) VALUES (?, ?, ?, ?, ?) ''', (goal_type, description, timeframe, priority, milestones)) conn.commit() conn.close() return {'success': True, 'message': f'Goal added: {description[:50]}...'} except Exception as e: return {'success': False, 'error': str(e)} def get_goals(goal_type=None, status='active'): """Get goals""" try: conn = get_connection() cursor = conn.cursor() query = 'SELECT * FROM goals WHERE 1=1' params = [] if goal_type: query += ' AND goal_type = ?' params.append(goal_type) if status: query += ' AND status = ?' params.append(status) query += ' ORDER BY priority DESC' cursor.execute(query, params) rows = cursor.fetchall() conn.close() return {'success': True, 'goals': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} def update_goal(goal_id, **kwargs): """Update a goal""" try: conn = get_connection() cursor = conn.cursor() updates = [] values = [] for field, value in kwargs.items(): updates.append(f'{field} = ?') values.append(value) updates.append('updated_at = ?') values.append(datetime.now()) values.append(goal_id) cursor.execute(f''' UPDATE goals SET {', '.join(updates)} WHERE id = ? ''', values) conn.commit() conn.close() return {'success': True, 'goal_id': goal_id} except Exception as e: return {'success': False, 'error': str(e)} # ==================== FACT FUNCTIONS ==================== def add_fact(category, fact, context=None, source=None): """Add a factual piece of information Categories: personal, professional, education, achievement, certification, etc. """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO facts (category, fact, context, source) VALUES (?, ?, ?, ?) ''', (category, fact, context, source)) conn.commit() conn.close() return {'success': True, 'message': f'Fact added: {fact[:50]}...'} except Exception as e: return {'success': False, 'error': str(e)} def get_facts(category=None, search=None): """Get facts""" try: conn = get_connection() cursor = conn.cursor() query = 'SELECT * FROM facts WHERE 1=1' params = [] if category: query += ' AND category = ?' params.append(category) if search: query += ' AND (fact LIKE ? OR context LIKE ?)' params.extend([f'%{search}%', f'%{search}%']) cursor.execute(query, params) rows = cursor.fetchall() conn.close() return {'success': True, 'facts': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== WORK STYLE FUNCTIONS ==================== def add_work_style(aspect, description, examples=None, tips_for_ai=None): """Add work style information Aspects: time_management, focus, collaboration, communication, tools, environment """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO work_style (aspect, description, examples, tips_for_ai) VALUES (?, ?, ?, ?) ''', (aspect, description, examples, tips_for_ai)) conn.commit() conn.close() return {'success': True, 'message': f'Work style added: {aspect}'} except Exception as e: return {'success': False, 'error': str(e)} def get_work_style(aspect=None): """Get work style information""" try: conn = get_connection() cursor = conn.cursor() if aspect: cursor.execute('SELECT * FROM work_style WHERE aspect = ?', (aspect,)) else: cursor.execute('SELECT * FROM work_style') rows = cursor.fetchall() conn.close() return {'success': True, 'work_style': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== COMMUNICATION STYLE FUNCTIONS ==================== def add_communication_style(aspect, preference, examples=None, avoid=None): """Add communication style preference Aspects: tone, formality, length, format, frequency, channels """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO communication_style (aspect, preference, examples, avoid) VALUES (?, ?, ?, ?) ''', (aspect, preference, examples, avoid)) conn.commit() conn.close() return {'success': True, 'message': f'Communication style added: {aspect}'} except Exception as e: return {'success': False, 'error': str(e)} def get_communication_style(aspect=None): """Get communication style""" try: conn = get_connection() cursor = conn.cursor() if aspect: cursor.execute('SELECT * FROM communication_style WHERE aspect = ?', (aspect,)) else: cursor.execute('SELECT * FROM communication_style') rows = cursor.fetchall() conn.close() return {'success': True, 'communication_style': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== QUICK MEMORY FUNCTIONS ==================== def remember(topic, content, tags=None, importance='normal', expires_at=None): """Quickly remember something This is for ad-hoc information that doesn't fit other categories """ try: conn = get_connection() cursor = conn.cursor() tags_str = ','.join(tags) if isinstance(tags, list) else tags cursor.execute(''' INSERT INTO quick_memory (topic, content, tags, importance, expires_at) VALUES (?, ?, ?, ?, ?) ''', (topic, content, tags_str, importance, expires_at)) conn.commit() conn.close() return {'success': True, 'message': f'Remembered: {topic}'} except Exception as e: return {'success': False, 'error': str(e)} def recall(topic=None, tags=None, search=None): """Recall stored memories""" try: conn = get_connection() cursor = conn.cursor() query = 'SELECT * FROM quick_memory WHERE (expires_at IS NULL OR expires_at > datetime("now"))' params = [] if topic: query += ' AND topic LIKE ?' params.append(f'%{topic}%') if tags: query += ' AND tags LIKE ?' params.append(f'%{tags}%') if search: query += ' AND (topic LIKE ? OR content LIKE ?)' params.extend([f'%{search}%', f'%{search}%']) query += ' ORDER BY importance DESC, created_at DESC' cursor.execute(query, params) rows = cursor.fetchall() conn.close() return {'success': True, 'memories': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== LEARNING LOG FUNCTIONS ==================== def learn_about_user(category, learned, context=None, confidence='medium'): """Log something learned about the user Categories: preference, behavior, style, fact, pattern """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO learning_log (category, learned, context, confidence) VALUES (?, ?, ?, ?) ''', (category, learned, context, confidence)) conn.commit() conn.close() return {'success': True, 'message': f'Learned: {learned[:50]}...'} except Exception as e: return {'success': False, 'error': str(e)} def get_learnings(category=None, limit=50): """Get learnings about the user""" try: conn = get_connection() cursor = conn.cursor() if category: cursor.execute(''' SELECT * FROM learning_log WHERE category = ? ORDER BY timestamp DESC LIMIT ? ''', (category, limit)) else: cursor.execute(''' SELECT * FROM learning_log ORDER BY timestamp DESC LIMIT ? ''', (limit,)) rows = cursor.fetchall() conn.close() return {'success': True, 'learnings': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== INTERACTION LOG FUNCTIONS ==================== def log_interaction(interaction_type, summary, details=None, outcome=None, learnings=None): """Log an interaction or activity Types: conversation, task, decision, meeting, research, project """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO interactions (interaction_type, summary, details, outcome, learnings) VALUES (?, ?, ?, ?, ?) ''', (interaction_type, summary, details, outcome, learnings)) conn.commit() conn.close() return {'success': True, 'message': f'Interaction logged: {summary[:50]}...'} except Exception as e: return {'success': False, 'error': str(e)} def get_interactions(interaction_type=None, limit=50): """Get past interactions""" try: conn = get_connection() cursor = conn.cursor() if interaction_type: cursor.execute(''' SELECT * FROM interactions WHERE interaction_type = ? ORDER BY timestamp DESC LIMIT ? ''', (interaction_type, limit)) else: cursor.execute(''' SELECT * FROM interactions ORDER BY timestamp DESC LIMIT ? ''', (limit,)) rows = cursor.fetchall() conn.close() return {'success': True, 'interactions': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== RELATIONSHIP FUNCTIONS ==================== def add_relationship(name, relationship_type, context=None, importance='medium', notes=None): """Add a relationship/contact""" try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO relationships (name, relationship_type, context, importance, notes) VALUES (?, ?, ?, ?, ?) ''', (name, relationship_type, context, importance, notes)) conn.commit() conn.close() return {'success': True, 'message': f'Relationship added: {name}'} except Exception as e: return {'success': False, 'error': str(e)} def get_relationships(relationship_type=None, search=None): """Get relationships""" try: conn = get_connection() cursor = conn.cursor() query = 'SELECT * FROM relationships WHERE 1=1' params = [] if relationship_type: query += ' AND relationship_type = ?' params.append(relationship_type) if search: query += ' AND (name LIKE ? OR context LIKE ? OR notes LIKE ?)' params.extend([f'%{search}%'] * 3) query += ' ORDER BY importance DESC, name' cursor.execute(query, params) rows = cursor.fetchall() conn.close() return {'success': True, 'relationships': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== DECISION FUNCTIONS ==================== def add_decision_style(decision_area, approach, factors=None, examples=None): """Add how you make decisions in a certain area Areas: career, financial, technical, personal, work, purchases """ try: conn = get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO decisions (decision_area, approach, factors, examples) VALUES (?, ?, ?, ?) ''', (decision_area, approach, factors, examples)) conn.commit() conn.close() return {'success': True, 'message': f'Decision style added: {decision_area}'} except Exception as e: return {'success': False, 'error': str(e)} def get_decision_styles(decision_area=None): """Get decision-making styles""" try: conn = get_connection() cursor = conn.cursor() if decision_area: cursor.execute('SELECT * FROM decisions WHERE decision_area = ?', (decision_area,)) else: cursor.execute('SELECT * FROM decisions') rows = cursor.fetchall() conn.close() return {'success': True, 'decisions': [dict(row) for row in rows]} except Exception as e: return {'success': False, 'error': str(e)} # ==================== COMPREHENSIVE RETRIEVAL ==================== def get_everything_about_user(): """Get ALL stored information about the user""" try: conn = get_connection() cursor = conn.cursor() result = { 'success': True, 'complete_profile': {} } # Get all tables tables = [ ('identity', 'SELECT * FROM identity'), ('behaviors', 'SELECT * FROM behaviors'), ('preferences', 'SELECT * FROM preferences'), ('patterns', 'SELECT * FROM patterns'), ('goals', 'SELECT * FROM goals WHERE status = "active"'), ('facts', 'SELECT * FROM facts'), ('work_style', 'SELECT * FROM work_style'), ('communication_style', 'SELECT * FROM communication_style'), ('relationships', 'SELECT * FROM relationships'), ('decisions', 'SELECT * FROM decisions'), ('recent_learnings', 'SELECT * FROM learning_log ORDER BY timestamp DESC LIMIT 20'), ('recent_memories', 'SELECT * FROM quick_memory WHERE expires_at IS NULL OR expires_at > datetime("now") ORDER BY created_at DESC LIMIT 20'), ] for table_name, query in tables: cursor.execute(query) rows = cursor.fetchall() result['complete_profile'][table_name] = [dict(row) for row in rows] conn.close() return result except Exception as e: return {'success': False, 'error': str(e)} def search_all_memory(search_term): """Search across all memory tables""" try: conn = get_connection() cursor = conn.cursor() results = [] # Search in multiple tables searches = [ ('identity', 'SELECT "identity" as source, key as title, value as content FROM identity WHERE key LIKE ? OR value LIKE ?'), ('behaviors', 'SELECT "behavior" as source, behavior_type as title, description as content FROM behaviors WHERE description LIKE ?'), ('preferences', 'SELECT "preference" as source, preference as title, value as content FROM preferences WHERE preference LIKE ? OR value LIKE ?'), ('facts', 'SELECT "fact" as source, category as title, fact as content FROM facts WHERE fact LIKE ?'), ('goals', 'SELECT "goal" as source, goal_type as title, description as content FROM goals WHERE description LIKE ?'), ('memories', 'SELECT "memory" as source, topic as title, content FROM quick_memory WHERE topic LIKE ? OR content LIKE ?'), ] term = f'%{search_term}%' for source, query in searches: try: param_count = query.count('?') cursor.execute(query, tuple([term] * param_count)) rows = cursor.fetchall() results.extend([dict(row) for row in rows]) except: pass conn.close() return {'success': True, 'results': results, 'count': len(results)} except Exception as e: return {'success': False, 'error': str(e)}