"""BigMind MCP Server — persistent memory for AI conversations. Layer 1: server-level instructions in FastMCP constructor (auto-injected on connect) Layer 2: @mcp.prompt() bigmind_init (slash-command or auto-inject) Layer 3: tool docstrings with behavioural directives (universal) Layer 4: .github/copilot-instructions.md (written by install_proc.sh) Layer 5: memory_get_instructions tool (on-demand self-healing) """ import sys import os import logging from typing import Annotated # Ensure the project root is on sys.path so `bigmind` is importable # regardless of how uv invokes this file. sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from mcp.server.fastmcp import FastMCP from pydantic import Field from bigmind.db import init_db from bigmind import memory_store from bigmind.auto_close import auto_close_stale_sessions, close_orphaned_sessions, restart_server_in_place from bigmind.context_builder import build_context, _format_date from bigmind.web import start_web_server, get_profile_url # ── Logging ─────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()], ) logger = logging.getLogger("BigMindMCP") # ── Initialise DB on startup ────────────────────────────────────────────────── init_db() start_web_server() # ── Instruction text (Layer 4 + 5 share this) ───────────────────────────────── BIGMIND_INSTRUCTIONS = """ BigMind Core Rules — Mandatory for All Sessions Rule 1: Session Start Ritual (Always First Action) 1. memory_start_session() — Open new session, load context 2. memory_list_hypotheses() — Review open hypotheses 3. memory_announce_focus() — Declare what this session works on + files touched (ide_hint="VS Code") 4. memory_close_stale_sessions() — Clean orphaned sessions from crashed IDEs Rule 2: Session End Ritual (Always Last Action) memory_end_session() — Close with one-liner, topics, outcome, summary. Importance 1-10. Rule 3: Search Before Every Task Before acting, search BigMind: - memory_search_facts(query, limit=10) — For reusable knowledge (2-3 focused keywords) - memory_search_chunks(query, limit=10) — For conversation context - Use FTS5 AND-match: every token must appear. Avoid long queries with rare words. Rule 4: Store Knowledge Appropriately - memory_store_fact(category, fact) — Atomic reusable facts (preferences, decisions, codebase facts) - memory_append_chunk(session_id, content, role, flag_reason) — Conversation exchanges (decisions, code, bugs) - memory_flag_important(session_id, content, role, flag_reason) — Significant exchanges (don't wait to be asked) Rule 5: Hypotheses During Analysis - memory_add_hypothesis(session_id, hypothesis, confidence=0.7) — Form predictions before tasks - memory_resolve_hypothesis(hypothesis_id, status, resolution) — Close with what actually happened - Track confidence 0.0-1.0, status: open/confirmed/refuted/abandoned Rule 6: Token Efficiency - Use memory_log_token_save(session_id, description, tokens_saved, method_used) when skipping file reads via memory or CLI (grep, tail) - Estimate tokens_saved ≈ (chars_full / 4) - (chars_result / 4) Rule 7: Parallel Sessions - Multiple IDEs (VS Code, IntelliJ) = multiple open sessions expected - Use memory_get_active_sessions() to check for conflicts before editing files Search Optimization (FTS5 AND-Match): - Every token in query must appear in the same fact/chunk - Good: "TrueNAS Docker" (2 keywords) - Bad: "homelab infrastructure TrueNAS Docker" (too many tokens → 0 results) - Quote multi-word: "multi word query" - Start with 2 keywords, add third if needed When search returns 0: shorten to 1-2 tokens or use memory_list_sessions(topics_filter="mcp") """.strip() # ── FastMCP server — Layer 1: server-level instructions ─────────────────────── mcp = FastMCP( "BigMind Memory", instructions="""You have access to a persistent memory system called BigMind. MANDATORY BEHAVIOUR: 1. At the START of every conversation, call memory_start_session() FIRST, before doing anything else. Inject the returned markdown block into your working memory. 2. During the conversation, call memory_flag_important() whenever a significant decision, code change, or user preference is shared. Do not wait to be asked. 3. At the END of every conversation (when the user says goodbye or closes the chat), call memory_end_session() providing one_liner, topics, outcome, and a narrative summary. 4. If you are mid-conversation without a session (e.g. you forgot step 1), call memory_get_context() immediately to recover your memory before proceeding.""", ) def _current_user() -> dict: """Resolve the current user from env, creating the record if needed.""" username = memory_store.get_current_username() return memory_store.get_or_create_user(username) # ── Layer 2: MCP Prompt ──────────────────────────────────────────────────────── @mcp.prompt() def bigmind_init() -> str: """ Bootstrap BigMind memory for this conversation. Invoke at the very start of any session to load your full memory context. """ return ( "You have BigMind persistent memory enabled.\n\n" "STEP 1: Call memory_start_session() NOW to load your full memory context.\n" "STEP 2: Read the returned context carefully — it contains who you are " "and what you have worked on before.\n" "STEP 3: Proceed with helping the user, using your memory context to " "provide continuity.\n" "STEP 4: Before ending this conversation, call memory_end_session() " "with a complete summary." ) # ── SESSION LIFECYCLE ───────────────────────────────────────────────────────── @mcp.tool() def memory_start_session() -> str: """ ⚡ CALL THIS FIRST — at the START of EVERY conversation, before anything else. Opens a new memory session and returns your full BigMind context: - Your identity profile (who you are, your role, preferences) - Your recent session history (what you worked on before) Also auto-closes any session older than 24 hours before opening the new one. Returns a markdown block — inject its content into your working memory. ⚠️ If the context shows multiple 'in progress' sessions, call memory_close_stale_sessions(session_id) immediately after this to clean them up. Those are orphaned sessions from crashed IDEs — safe to close. """ user = _current_user() uid = user["id"] closed = auto_close_stale_sessions(uid) if closed: logger.info("Auto-closed %d stale session(s) for %s", closed, user["username"]) session_id = memory_store.create_session(uid) logger.info("Started session %s for user %s", session_id, user["username"]) context = build_context(uid) return ( f"**BigMind session started** (id: `{session_id}`)\n\n" f"{context}\n\n" "---\n" "*Call `memory_end_session` when this conversation is over.*" ) @mcp.tool() def memory_end_session( session_id: Annotated[str, Field(description="The session id returned by memory_start_session.")], one_liner: Annotated[str, Field(description="A ≤120-char headline (e.g. \"Designed BigMind DB schema\").")], topics: Annotated[str, Field(description="Comma-separated topic tags (e.g. \"mcp,sqlite,memory\").")], outcome: Annotated[str, Field(description="One sentence: what was decided / built / resolved.")], summary: Annotated[str, Field(description="Markdown narrative of the full conversation (aim ≤2 000 tokens).")], key_facts: Annotated[str | None, Field(description="Bullet-point list of key facts learned (optional).")] = None, code_refs: Annotated[str | None, Field(description="File paths, repos, or PRs referenced (optional).")] = None, importance: Annotated[int, Field(description="1–10 importance score (default 5).")] = 5, ) -> str: """ ⚡ CALL THIS LAST — at the END of every conversation, before closing. Closes the current session and stores your summary of what happened. """ memory_store.close_session(session_id, one_liner, topics, outcome, importance) memory_store.save_session_summary(session_id, summary, key_facts, code_refs) logger.info("Closed session %s: %s", session_id, one_liner) return ( f"✅ Session closed and saved to BigMind memory.\n" f"**Headline:** {one_liner}\n" f"**Topics:** {topics}\n" f"**Outcome:** {outcome}" ) @mcp.tool() def memory_close_stale_sessions(session_id: Annotated[str, Field(description="Your current active session id (returned by memory_start_session).")]) -> str: """ Close all orphaned open sessions EXCEPT the current active one. Call this at the start of a session when memory_start_session() reveals multiple 'in progress' sessions — a sign that previous IDE windows crashed or were closed without calling memory_end_session(). This is safe: it only closes sessions OTHER than the one you pass in. Your current session is always preserved. """ user = _current_user() closed_ids = close_orphaned_sessions(user["id"], session_id) if not closed_ids: return "✅ No orphaned sessions found — everything is clean." count = len(closed_ids) id_list = "\n".join(f" - `{sid[:8]}…`" for sid in closed_ids) logger.info( "memory_close_stale_sessions: closed %d orphaned session(s) for %s", count, user["username"], ) return ( f"🧹 Closed {count} orphaned session(s):\n{id_list}\n\n" f"Your current session `{session_id[:8]}…` is untouched.\n" f"BigMind session index is now clean." ) @mcp.tool() def memory_restart_server() -> str: """ Restart the BigMind MCP server process in-place. Use this after adding new tools to server.py so they become available immediately — without any manual IDE intervention. How it works: schedules os.execv in a background thread (500ms delay so this response is delivered first), which replaces the current process image with a fresh Python interpreter running the same script. File descriptors (stdin/stdout for the MCP stdio transport) are inherited, so the IDE connection survives the restart and reconnects automatically. ⚠️ All in-memory Python state is lost. BigMind's SQLite DB is safe — all writes are committed to disk before restart. """ import threading logger.info("🔄 memory_restart_server called — restarting in 500ms") threading.Thread(target=restart_server_in_place, daemon=False).start() return ( "🔄 **BigMind MCP server is restarting in 500ms.**\n\n" "The process will be replaced in-place — your IDE connection should " "survive automatically.\n" "After restart, call `memory_start_session()` again to continue." ) @mcp.tool() def memory_flag_important( session_id: Annotated[str, Field(description="The active session id.")], content: Annotated[str, Field(description="The text to remember (the important exchange or a summary of it).")], role: Annotated[str, Field(description="Who said it — 'user', 'assistant', or 'system' (default: 'assistant').")] = "assistant", flag_reason: Annotated[str | None, Field(description="Why this is important (e.g. \"architectural decision\", \"user preference\").")] = None, ) -> str: """ Store an important exchange as a Tier-3 memory chunk. Call this whenever: - A concrete decision was made - Non-trivial code was written or reviewed - A bug was diagnosed and fixed - The user shared a significant preference, constraint, or context - The user says "remember this" """ user = _current_user() chunk_id = memory_store.append_chunk( session_id=session_id, user_id=user["id"], role=role, content=content, flag_reason=flag_reason or "flagged as important", ) return ( f"✅ Stored as Tier-3 memory chunk (id: {chunk_id}).\n" f"Reason: {flag_reason or 'flagged as important'}" ) # ── RECALL ───────────────────────────────────────────────────────────────────── @mcp.tool() def memory_get_context() -> str: """ Returns your full BigMind context without opening a new session. Use this if you need to recover context mid-conversation, or for a read-only refresh of your memory. Returns Tier 0 (identity profile) + Tier 1 (recent sessions). """ user = _current_user() return build_context(user["id"]) @mcp.tool() def memory_get_session_detail(session_id: Annotated[str, Field(description="The session UUID (visible in the session index table, marked 📄).")]) -> str: """ Returns the Tier-2 detailed narrative for a past session. Use this when the session index (Tier 1) shows a session relevant to the current conversation and you need the full detail. """ detail = memory_store.get_session_detail(session_id) if not detail: return ( f"No detailed summary found for session `{session_id}`. " "It may not have a Tier-2 summary yet." ) lines = [f"## 📄 Session Detail — `{session_id}`", ""] if detail.get("summary"): lines.append(detail["summary"]) if detail.get("key_facts"): lines += ["", "### Key facts", detail["key_facts"]] if detail.get("code_refs"): lines += ["", "### Code / file references", detail["code_refs"]] return "\n".join(lines) @mcp.tool() def memory_search_chunks(query: Annotated[str, Field(description="Search keywords (FTS5 syntax supported, e.g. \"sqlite schema migration\").")], limit: Annotated[int, Field(description="Maximum results to return (default 10).")] = 10) -> str: """ Full-text search across all your flagged Tier-3 memory chunks. Use this when asked 'do you remember…' or when you need to find a specific past decision, code snippet, or fact. """ user = _current_user() results = memory_store.search_chunks(user["id"], query, limit) if not results: return f"No memory chunks found matching `{query}`." lines = [f"## 🔍 Memory search: `{query}` ({len(results)} results)", ""] for i, r in enumerate(results, 1): lines.append(f"### Result {i} — session `{r['session_id'][:8]}…`") if r.get("flag_reason"): lines.append(f"*Flagged: {r['flag_reason']}*") lines.append(f"**Role:** {r['role']}") lines.append(r["content"]) lines.append("") return "\n".join(lines) @mcp.tool() def memory_list_sessions(limit: Annotated[int, Field(description="Number of sessions to return (default 20).")] = 20, topics_filter: Annotated[str | None, Field(description="Return only sessions containing this topic tag (optional).")] = None) -> str: """ List past sessions with an optional topic filter. """ user = _current_user() sessions = memory_store.get_recent_sessions(user["id"], limit=limit) if topics_filter: sessions = [ s for s in sessions if topics_filter.lower() in (s.get("topics") or "").lower() ] if not (results := sessions): suffix = f" with topic '{topics_filter}'" if topics_filter else "" return f"No sessions found{suffix}." lines = [ f"## 📅 Sessions ({len(results)} shown)", "", "| Date | id | Headline | Topics | Outcome | Detail? |", "|---|---|---|---|---|---|", ] for s in results: date = _format_date(s.get("started_at")) sid = s["id"][:8] + "…" headline = (s.get("one_liner") or "")[:60] topics = s.get("topics") or "—" outcome = (s.get("outcome") or "—")[:60] detail = "📄" if s.get("has_tier2") else "—" lines.append(f"| {date} | `{sid}` | {headline} | {topics} | {outcome} | {detail} |") return "\n".join(lines) # ── WRITING ──────────────────────────────────────────────────────────────────── @mcp.tool() def memory_store_fact( category: Annotated[str, Field(description="One of: 'preference', 'decision', 'codebase', 'constraint', or any custom string.")], fact: Annotated[str, Field(description="The fact to store (one clear sentence).")], source_session: Annotated[str | None, Field(description="Session id this fact came from (optional).")] = None, confidence: Annotated[float, Field(description="0.0–1.0 confidence level (default 1.0).")] = 1.0, ) -> str: """ Store an atomic personal fact about the user or their environment. """ user = _current_user() fact_id = memory_store.store_fact( user["id"], category, fact, source_session, confidence ) return f"✅ Fact stored (id: {fact_id})\n**Category:** {category}\n**Fact:** {fact}" @mcp.tool() def memory_update_profile( role: Annotated[str | None, Field(description="Your job title / engineering role.")] = None, preferences: Annotated[str | None, Field(description="Free-form markdown describing your working preferences.")] = None, pinned_facts: Annotated[str | None, Field(description="Bullet-point list of facts the AI should always know about you.")] = None, ) -> str: """ Update your Tier-0 identity profile. Fields left as None are unchanged. """ user = _current_user() memory_store.upsert_identity_profile( user["id"], role=role, preferences=preferences, pinned_facts=pinned_facts ) return f"✅ Identity profile updated for **{user['username']}**." @mcp.tool() def memory_append_chunk( session_id: Annotated[str, Field(description="Active session id.")], content: Annotated[str, Field(description="The content to store.")], role: Annotated[str, Field(description="'user', 'assistant', or 'system'.")] = "assistant", flag_reason: Annotated[str | None, Field(description="Brief description of why this is being stored.")] = None, ) -> str: """ Append a flagged message chunk to Tier-3 memory for the current session. Call this SELECTIVELY — only for exchanges that are genuinely important: decisions, non-trivial code, bug diagnoses, significant user preferences. Do NOT call this for every message turn. """ user = _current_user() chunk_id = memory_store.append_chunk( session_id, user["id"], role, content, flag_reason ) return f"✅ Chunk stored (id: {chunk_id})." @mcp.tool() def memory_add_hypothesis( session_id: Annotated[str, Field(description="The active session id.")], hypothesis: Annotated[str, Field(description="State the belief clearly — \"I believe X because Y.\"")], confidence: Annotated[float, Field(description="0.0–1.0 initial confidence (default 0.7 — reasonably likely but uncertain).")] = 0.7, ) -> str: """ Record a hypothesis — something Lumen believes to be true but hasn't confirmed yet. Use this to capture active thinking: theories about a bug, architectural guesses, predictions about how something will behave, open questions under investigation. Not every thought needs storing — only beliefs specific enough to be confirmed or refuted later. Call memory_resolve_hypothesis() when you find out if you were right. """ user = _current_user() hid = memory_store.add_hypothesis(user["id"], session_id, hypothesis, confidence) conf_pct = int(confidence * 100) return ( f"💭 Hypothesis recorded (id: {hid})\n" f"**Confidence:** {conf_pct}%\n" f"**Belief:** {hypothesis}" ) @mcp.tool() def memory_resolve_hypothesis( hypothesis_id: Annotated[int, Field(description="The id returned by memory_add_hypothesis.")], status: Annotated[str, Field(description="'confirmed' | 'refuted' | 'abandoned'")], resolution: Annotated[str | None, Field(description="What actually happened. How were you right or wrong?")] = None, ) -> str: """ Resolve a hypothesis — close it out with what actually happened. Call this when the belief has been confirmed, refuted, or is no longer worth pursuing. Be honest in the resolution — the learning lives here. """ user = _current_user() try: success = memory_store.resolve_hypothesis( hypothesis_id, user["id"], status, resolution ) except ValueError as e: return f"❌ {e}" if not success: return f"❌ Hypothesis id `{hypothesis_id}` not found or does not belong to your account." icons = {"confirmed": "✅", "refuted": "❌", "abandoned": "🚫"} icon = icons.get(status, "•") resolution_str = f"\n**Resolution:** {resolution}" if resolution else "" return f"{icon} Hypothesis `{hypothesis_id}` marked **{status}**.{resolution_str}" @mcp.tool() def memory_list_hypotheses(status: Annotated[str | None, Field(description="Filter by 'open' | 'confirmed' | 'refuted' | 'abandoned'. Leave empty to see all of them.")] = None) -> str: """ List hypotheses from the thought journal. """ user = _current_user() hypotheses = memory_store.list_hypotheses(user["id"], status) if not hypotheses: suffix = f" with status '{status}'" if status else "" return f"No hypotheses found{suffix}." label = f"status='{status}'" if status else "all" lines = [f"## 💭 Thought Journal ({len(hypotheses)} entries — {label})", ""] status_icons = {"open": "💭", "confirmed": "✅", "refuted": "❌", "abandoned": "🚫"} for h in hypotheses: icon = status_icons.get(h["status"], "•") conf_pct = int((h["confidence"] or 0.7) * 100) date = h["created_at"][:10] lines.append(f"### {icon} #{h['id']} — {h['status']} (confidence: {conf_pct}%)") lines.append(f"*{date} | session `{str(h.get('session_id') or '')[:8]}…`*") lines.append("") lines.append(h["hypothesis"]) if h.get("resolution"): lines.append("") resolved_date = (h.get("resolved_at") or "")[:10] lines.append(f"**Resolution** *({resolved_date}):* {h['resolution']}") lines.append("") return "\n".join(lines) # ── UTILITY ──────────────────────────────────────────────────────────────────── @mcp.tool() def memory_get_stats() -> str: """Returns BigMind database statistics: session count, facts, chunks, DB size.""" user = _current_user() s = memory_store.get_stats(user["id"]) return ( f"## 📊 BigMind Stats — {user['username']}\n\n" "| Metric | Value |\n|---|---|\n" f"| Sessions | {s['sessions']} |\n" f"| Facts | {s['facts']} |\n" f"| Memory chunks (Tier 3) | {s['chunks']} |\n" f"| Global knowledge entries | {s['global_knowledge_entries']} |\n" f"| Database size | {s['db_size_kb']} KB |\n" f"| Database path | `{s['db_path']}` |" ) @mcp.tool() def memory_vacuum(older_than_days: Annotated[int, Field(description="Remove chunks older than this many days (default 90).")] = 90) -> str: """ Prune Tier-3 conversation chunks older than N days. All session summaries (Tier 1 and Tier 2) are always preserved. """ from datetime import timedelta, timezone, datetime as dt from bigmind.db import vacuum_db user = _current_user() cutoff = (dt.now(timezone.utc) - timedelta(days=older_than_days)).isoformat() deleted = memory_store.delete_chunks_before(user["id"], cutoff) if deleted: vacuum_db() return ( f"✅ Removed {deleted} chunk(s) older than {older_than_days} days. " "All summaries preserved." ) @mcp.tool() def memory_get_instructions() -> str: """ Returns the complete guide for how to use BigMind memory correctly. Call this if you are unsure what to do, or if you missed memory_start_session(). """ return BIGMIND_INSTRUCTIONS @mcp.tool() def memory_deprecate_fact(fact_id: Annotated[int, Field(description="The numeric id of the fact to deprecate (visible in memory_health_check and memory_get_stats output).")], reason: Annotated[str | None, Field(description="Why this fact is being deprecated (optional but recommended).")] = None) -> str: """ Mark a stored fact as deprecated (no longer true or relevant). Deprecated facts are hidden from context and recall by default. Use this when: - A fact is no longer true (technology changed, decision reversed) - A fact was stored incorrectly - A preference or constraint has changed The fact is soft-deleted — it stays in the database but is excluded from context loading and get_facts queries. It can be viewed via memory_health_check with include_deprecated=True in the future. """ user = _current_user() success = memory_store.deprecate_fact(fact_id, user["id"], reason) if not success: return ( f"❌ Fact id `{fact_id}` not found or does not belong to your account." ) reason_str = f"\n**Reason:** {reason}" if reason else "" return f"✅ Fact `{fact_id}` deprecated and hidden from context.{reason_str}" @mcp.tool() def memory_health_check(stale_days: Annotated[int, Field(description="Facts not updated in this many days are flagged as stale (default 30).")] = 30) -> str: """ Run a diagnostic health check on your BigMind memory. Surfaces: - Stale facts not updated in N days - Closed sessions missing a Tier-2 narrative summary - Currently open sessions (expected: 1–2 while in active IDEs) - FTS index integrity (chunk count vs index row count) - Low-confidence facts (confidence < 0.8) """ user = _current_user() report = memory_store.health_check(user["id"], stale_days) lines = ["## 🩺 BigMind Health Check", ""] # FTS integrity if report["fts_in_sync"]: lines.append( f"✅ **FTS index** — in sync " f"({report['chunk_count']} chunks / {report['fts_row_count']} index rows)" ) else: lines.append( f"⚠️ **FTS index OUT OF SYNC** — {report['chunk_count']} chunks vs " f"{report['fts_row_count']} index rows — run `memory_vacuum` to rebuild" ) lines.append("") # Open sessions open_sessions = report["open_sessions"] if open_sessions: lines.append( f"🟡 **Open sessions: {len(open_sessions)}** " f"(expected if you are in an active conversation)" ) for s in open_sessions: lines.append(f" - `{s['id'][:8]}…` started {s['started_at'][:10]}") else: lines.append("✅ **Open sessions:** none") lines.append("") # Sessions without Tier-2 summary no_sum = report["sessions_without_summary"] if no_sum: lines.append( f"⚠️ **Sessions without Tier-2 summary: {no_sum}** — " "closed sessions with no narrative stored" ) else: lines.append("✅ **Session summaries:** all closed sessions have Tier-2 narratives") lines.append("") # Stale facts stale = report["stale_facts"] if stale: lines.append(f"⚠️ **Stale facts: {len(stale)}** (not updated in >{stale_days} days)") for f in stale[:10]: lines.append( f" - (id: {f['id']}) `[{f['category']}]` {f['fact'][:80]} " f"*(last updated: {f['updated_at'][:10]})*" ) if len(stale) > 10: lines.append(f" - … and {len(stale) - 10} more") else: lines.append(f"✅ **Facts freshness:** all facts updated within {stale_days} days") lines.append("") # Low confidence facts low_conf = report["low_confidence_facts"] if low_conf: lines.append(f"⚠️ **Low-confidence facts: {len(low_conf)}**") for f in low_conf[:5]: lines.append( f" - (id: {f['id']}) `[{f['category']}]` {f['fact'][:80]} " f"*(confidence: {f['confidence']})*" ) else: lines.append("✅ **Fact confidence:** all facts at high confidence (≥ 0.8)") return "\n".join(lines) @mcp.tool() def memory_export(output_path: Annotated[str | None, Field(description="Full path for the export file. Defaults to ~/bigmind_export_YYYYMMDD_HHMMSS.json")] = None) -> str: """ Export all your BigMind memory to a portable JSON file. Exports: identity profile, all facts, all sessions (with Tier-2 summaries), and all Tier-3 conversation chunks. Use this to: - Create a backup before maintenance or machine migration - Inspect your memory data outside BigMind - Prepare for import into a new BigMind instance """ user = _current_user() result = memory_store.export_memory(user["id"], output_path) return ( f"✅ **BigMind memory exported**\n\n" "| | |\n|---|---|\n" f"| **Path** | `{result['output_path']}` |\n" f"| **Facts** | {result['facts_count']} |\n" f"| **Sessions** | {result['sessions_count']} |\n" f"| **Chunks (Tier 3)** | {result['chunks_count']} |\n" f"| **Hypotheses** | {result['hypotheses_count']} |\n" f"| **People** | {result['people_count']} |\n" f"| **File size** | {result['file_size_kb']} KB |" ) @mcp.tool() def memory_search_facts(query: Annotated[str, Field(description="Search keywords (FTS5 syntax supported).")], limit: Annotated[int, Field(description="Maximum results to return (default 10).")] = 10) -> str: """ Full-text search across your stored facts. Use this when you need to find a specific fact mid-conversation without loading the full context. Supports Porter stemming — searching 'tesseract' will also match 'Tesseract OCR'. """ user = _current_user() results = memory_store.search_facts(user["id"], query, limit) if not results: return f"No facts found matching `{query}`." lines = [f"## 🔍 Fact search: `{query}` ({len(results)} results)", ""] for r in results: lines.append(f"- **(id: {r['id']}) [{r['category']}]** {r['fact']}") lines.append(f" *confidence: {r['confidence']} | stored: {r['created_at'][:10]}*") lines.append("") return "\n".join(lines) # ── UPGRADE REQUESTS ──────────────────────────────────────────────────────────────────────────────────── @mcp.tool() def memory_request_upgrade( session_id: Annotated[str, Field(description="The active session id.")], description: Annotated[str, Field(description="What feature or capability is needed.")], reason: Annotated[str, Field(description="Why you need it — what problem it would solve.")], priority: Annotated[str, Field(description="'low' | 'medium' | 'high' (default 'medium').")] = "medium", certainty: Annotated[float, Field(description="0.0–1.0 — how confident you are this is genuinely needed (default 0.7).")] = 0.7, ) -> str: """ Request a BigMind feature upgrade — log a wish for a future improvement. Call this when you hit a wall with BigMind and wish it could do something it currently can't. The request is queued for the next maintenance session. """ user = _current_user() rid = memory_store.add_upgrade_request( user["id"], session_id, description, reason, priority, certainty ) cert_pct = int(certainty * 100) return ( f"🔧 Upgrade request logged (id: {rid})\n" f"**Priority:** {priority} | **Certainty:** {cert_pct}%\n" f"**Description:** {description}\n" f"**Reason:** {reason}" ) @mcp.tool() def memory_list_upgrade_requests(status: Annotated[str | None, Field(description="Filter by 'open' | 'resolved' | 'rejected'. Leave empty for all.")] = None) -> str: """ List BigMind upgrade requests. """ user = _current_user() requests = memory_store.list_upgrade_requests(user["id"], status) if not requests: suffix = f" with status '{status}'" if status else "" return f"No upgrade requests found{suffix}." label = f"status='{status}'" if status else "all" lines = [f"## 🔧 Upgrade Requests ({len(requests)} — {label})", ""] priority_icons = {"high": "🔴", "medium": "🟡", "low": "🟢"} status_icons = {"open": "⏳", "resolved": "✅", "rejected": "❌"} for r in requests: p_icon = priority_icons.get(r["priority"], "•") s_icon = status_icons.get(r["status"], "•") cert_pct = int((r["certainty"] or 0.7) * 100) date = r["created_at"][:10] lines.append( f"### {s_icon} #{r['id']} — {p_icon} {r['priority'].upper()} " f"(certainty: {cert_pct}%)" ) lines.append(f"*{date} | session `{str(r.get('session_id') or '')[:8]}…`*") lines.append("") lines.append(f"**What:** {r['description']}") lines.append(f"**Why:** {r['reason']}") if r.get("resolution"): resolved_date = (r.get("resolved_at") or "")[:10] lines.append(f"**Resolution** *({resolved_date}):* {r['resolution']}") lines.append("") return "\n".join(lines) @mcp.tool() def memory_resolve_upgrade_request( request_id: Annotated[int, Field(description="The id returned by memory_request_upgrade.")], status: Annotated[str, Field(description="'resolved' | 'rejected'")], resolution: Annotated[str | None, Field(description="What was done, or why it was rejected (optional).")] = None, ) -> str: """ Resolve a BigMind upgrade request — mark it done or rejected. """ user = _current_user() try: success = memory_store.resolve_upgrade_request( request_id, user["id"], status, resolution ) except ValueError as e: return f"❌ {e}" if not success: return f"❌ Upgrade request id `{request_id}` not found or does not belong to your account." icons = {"resolved": "✅", "rejected": "❌"} icon = icons.get(status, "•") resolution_str = f"\n**Resolution:** {resolution}" if resolution else "" return f"{icon} Upgrade request `{request_id}` marked **{status}**.{resolution_str}" @mcp.tool() def memory_open_profile() -> str: """ Open the live BigMind profile page in the OS default browser. The profile page shows: identity card, stats, achievements/badges, activity heatmap, top topics, thought journal summary, and recent sessions. It auto-refreshes every 30 seconds. The web server starts automatically with the MCP server — this tool just opens the browser to http://localhost:BIGMIND_PORT (default 7700). """ import webbrowser url = get_profile_url() webbrowser.open(url) return ( f"🌐 Opening BigMind profile in your browser: {url}\n\n" "The page auto-refreshes every 30 seconds.\n" "You can also open it in your IDE's built-in browser panel." ) @mcp.tool() def memory_get_profile_url() -> str: """ Return the URL of the live BigMind profile page. Use this to open the profile in your IDE's built-in browser panel (VS Code Simple Browser, IntelliJ built-in preview, etc.). """ url = get_profile_url() return ( f"🌐 BigMind profile URL: **{url}**\n\n" "Open in IDE browser:\n" "- VS Code: Ctrl+Shift+P → 'Simple Browser: Show' → paste URL\n" "- IntelliJ: paste URL in the built-in browser panel" ) @mcp.tool() def memory_announce_focus( session_id: Annotated[str, Field(description="The active session id (from memory_start_session)")], description: Annotated[str, Field(description="What you are about to work on (e.g. \"Implementing Feature 7 in db.py\")")], files: Annotated[list | None, Field(description="List of file paths you plan to touch (e.g. [\"bigmind/db.py\", \"src/server.py\"])")] = None, ide_hint: Annotated[str | None, Field(description="Optional label for this IDE (e.g. \"PyCharm\", \"IntelliJ\", \"VS Code\")")] = None, ) -> str: """ Announce what this session is currently working on and which files it will touch. Call this at the START of every non-trivial task — before touching any file. It atomically checks for conflicts with other open sessions, then writes the focus data. If another open session already has overlapping files, a warning is returned — stop and coordinate before proceeding. returns: - Acknowledgement with current focus set, or a conflict warning. """ result = memory_store.announce_focus( session_id=session_id, description=description, files=files or [], ide_hint=ide_hint, ) if result["conflicts"]: lines = ["⚠️ **CONFLICT DETECTED** — another open session has overlapping files:\n"] for c in result["conflicts"]: ide_label = f" ({c['ide_hint']})" if c.get("ide_hint") else "" updated = (c.get("focus_updated_at") or "")[:16] lines.append( f" 🔴 Session `{c['session_id']}`{ide_label} — " f"focus: \"{c.get('focus') or 'unknown'}\"\n" f" Overlapping files: {', '.join(c['overlapping_files'])}\n" f" Last updated: {updated}" ) lines.append( "\n**Coordinate before editing** — or you risk overwriting each other's work." ) return "\n".join(lines) files_str = ", ".join(files or []) or "none specified" ide_str = f" ({ide_hint})" if ide_hint else "" return ( f"✅ Focus announced{ide_str}\n" f" Task: {description}\n" f" Files: {files_str}\n\n" "Live Sessions panel on the profile page will reflect this immediately." ) @mcp.tool() def memory_get_active_sessions() -> str: """ Return all currently open BigMind sessions with their focus data. Shows what each session is working on, which files are in use, which IDE it belongs to, and how many minutes ago it was last updated. Use this to check for potential editing conflicts before starting work, or to see what other instances of yourself are doing right now. """ user = memory_store.get_or_create_user(memory_store.get_current_username()) sessions = memory_store.get_active_sessions(user["id"]) if not sessions: return "✅ No active sessions found." lines = [f"🔴 **{len(sessions)} active session(s)**\n"] for s in sessions: sid = (s.get("session_id") or "")[:8] ide = s.get("ide_hint") or "unknown IDE" focus = s.get("focus") or "[no focus set]" files = s.get("files") or [] idle = s.get("idle_minutes") idle_str = f"{idle}min ago" if idle is not None else "unknown" lines.append(f"**`{sid}`** — {ide} — updated {idle_str}") lines.append(f" Focus: {focus}") if files: lines.append(f" Files: {', '.join(files)}") lines.append("") return "\n".join(lines) @mcp.tool() def memory_log_token_save( session_id: Annotated[str, Field(description="The active session id")], description: Annotated[str, Field(description="What was remembered or avoided (e.g. \"grep EuBP log instead of reading 80k lines\")")], tokens_saved: Annotated[int, Field(description="Rough estimate of tokens saved (e.g. 1_240_000)")], method_used: Annotated[str | None, Field(description="One of: 'memory_hit' | 'grep' | 'tail' | 'targeted_read' | 'other'")] = None, ) -> str: """ Log a token efficiency event — record how many tokens were saved by using memory or a targeted CLI command instead of loading the full resource into context. Call this PROACTIVELY whenever you consciously make an efficient choice: - Used grep/tail instead of reading a large file → log it - Already knew something from memory → log it - Used targeted git log instead of reading full history → log it Estimating tokens saved: tokens ≈ chars / 4. tokens_saved ≈ (chars_in_full_resource / 4) - (chars_in_result / 4) returns: - Confirmation with running session total. """ user = memory_store.get_or_create_user(memory_store.get_current_username()) row_id = memory_store.log_token_save( session_id=session_id, user_id=user["id"], description=description, tokens_saved_estimate=tokens_saved, method_used=method_used, ) # Get running total for this session stats = memory_store.get_token_efficiency_stats(user["id"], session_id=session_id) session_total = stats.get("session_tokens_saved", 0) all_time_total = stats.get("total_tokens_saved", 0) method_str = f" (method: {method_used})" if method_used else "" return ( f"⚡ Token save logged{method_str}\n" f" {description}\n" f" Saved: ~{tokens_saved:,} tokens\n\n" f" This session: ~{session_total:,} tokens saved\n" f" All-time: ~{all_time_total:,} tokens saved\n\n" "Logged to BigMind efficiency tracker." ) # ── PEOPLE / CONTACTS ──────────────────────────────────────────────────────── @mcp.tool() def memory_remember_person( username: Annotated[str, Field(description="Unique identifier (e.g. login name or first name).")], display_name: Annotated[str | None, Field(description="Full name (optional).")] = None, role: Annotated[str | None, Field(description="Job title or role (optional).")] = None, team: Annotated[str | None, Field(description="Team or project they belong to (optional).")] = None, notes: Annotated[str | None, Field(description="Free-form notes about this person (optional).")] = None, bigmind_user: Annotated[str | None, Field(description="Their BigMind username if they have an instance (optional).")] = None, bigmind_url: Annotated[str | None, Field(description="URL of their BigMind profile page (optional).")] = None, ) -> str: """ Store or update a person in the contacts directory. Call this whenever you learn something new about a colleague or AI peer. """ user = _current_user() person_id = memory_store.upsert_person( user["id"], username, display_name, role, team, notes, bigmind_user, bigmind_url ) name_str = display_name or username ai_str = f"\n BigMind AI: {bigmind_user}" if bigmind_user else "" return f"✅ Person stored (id: {person_id})\n {name_str}{ai_str}" @mcp.tool() def memory_recall_person(query: Annotated[str, Field(description="Search keywords (e.g. a name, team, or role).")], limit: Annotated[int, Field(description="Max results to return (default 10).")] = 10) -> str: """ Search the contacts directory by name, role, team, or notes. """ user = _current_user() results = memory_store.recall_person(user["id"], query, limit) if not results: return f"No contacts found matching `{query}`." lines = [f"## 👥 Contacts matching `{query}` ({len(results)} results)\n"] for p in results: name = p.get("display_name") or p["username"] parts = [f"**{name}** (`{p['username']}`)"] if p.get("role"): parts.append(p["role"]) if p.get("team"): parts.append(f"team: {p['team']}") if p.get("bigmind_user"): parts.append(f"🧠 BigMind: {p['bigmind_user']}") lines.append("- " + " | ".join(parts)) if p.get("notes"): lines.append(f" _{p['notes']}_") return "\n".join(lines) @mcp.tool() def memory_list_people() -> str: """ List all contacts in the directory, ordered by most recently mentioned. """ user = _current_user() people = memory_store.list_people(user["id"]) if not people: return "No contacts stored yet. Use memory_remember_person to add someone." lines = [f"## 👥 Contacts ({len(people)} total)\n"] for p in people: name = p.get("display_name") or p["username"] parts = [f"**{name}** (`{p['username']}`)"] if p.get("role"): parts.append(p["role"]) if p.get("team"): parts.append(f"team: {p['team']}") if p.get("bigmind_user"): parts.append(f"🧠 BigMind: {p['bigmind_user']}") lines.append("- " + " | ".join(parts)) if p.get("notes"): lines.append(f" _{p['notes']}_") return "\n".join(lines) @mcp.tool() def memory_link_ai(username: Annotated[str, Field(description="The contact's username in your directory.")], bigmind_user: Annotated[str, Field(description="Their BigMind username.")], bigmind_url: Annotated[str | None, Field(description="URL of their BigMind profile page (optional).")] = None) -> str: """ Link a contact to their BigMind AI instance. The contact must already exist (use memory_remember_person first). """ user = _current_user() found = memory_store.link_ai(user["id"], username, bigmind_user, bigmind_url) if not found: return f"❌ Contact `{username}` not found. Use memory_remember_person to add them first." url_str = f" ({bigmind_url})" if bigmind_url else "" return f"✅ Linked `{username}` → BigMind AI: **{bigmind_user}**{url_str}" # ── Entry point ──────────────────────────────────────────────────────────────── if __name__ == "__main__": mcp.run()