Hook 1
mem0 Session Start
The core hook. Searches mem0 for recent project memories and injects them as context so Claude knows where you left off — without you typing anything.
Two modes
| Mode | When | Behavior |
|---|
| Normal |
source: "startup" or "resume" |
Lightweight — 5 recent memories, last 7 days, brief format |
| Compact recovery |
source: "compact" |
Heavy — searches for pre-compact handoff state + 10 context memories, formatted with recovery header |
The compact recovery mode is critical. When Claude's context window fills up and compacts, it loses the conversation history. This hook detects compaction and injects a heavier context block with a ## Context restored after compaction header, including any handoff state that was saved before the compact event.
Project detection
The hook detects which project you're in by reading the git root directory name and mapping it to a mem0 agent_id tag. This ensures you only get memories relevant to the current project, not cross-contamination from other work.
# Project mapping — git directory name → mem0 agent_id
PROJECT_MAP = {
"StakTrakr": "staktrakr",
"HexTrackr": "hextrackr",
"MyMelo": "mymelo",
"Forge": "forge",
# ... add your projects
}
# The hook reads the git root:
# git rev-parse --show-toplevel → /path/to/StakTrakr
# basename → StakTrakr
# PROJECT_MAP["StakTrakr"] → "staktrakr"
# Then searches mem0 with agent_id="staktrakr"
Source filtering
Not all mem0 memories are equal. The hook prefers memories from high-signal sources (session hooks, compaction hooks) over bulk-imported or migrated memories. If preferred-source memories exist, lower-quality ones are filtered out.
# High-signal sources — prefer these
PREFERRED_SOURCES = {"stop-hook", "pre-compact-hook", "post-tool-use-hook"}
# If any preferred-source memories exist, only show those
# Otherwise, fall back to showing everything
What Claude sees
On a normal startup, Claude receives something like:
Recent mem0 context for StakTrakr (last 7 days):
1. [2026-04-03] (session-digest) Fixed JM Bullion price scraper —
eCheck column was being read as Card/PayPal. PR #903 merged.
2. [2026-04-02] (retro-learning) Always check column order in
vendor-specific parsers before assuming index positions.
3. [2026-04-01] (session-digest) Deployed poller v3.33.83 to
Portainer Stack 7 and Fly.io. All prices verified.
#!/usr/bin/env python3
"""
mem0 SessionStart hook for Claude Code.
Two modes:
- startup/resume: Lightweight — 5 recent memories
- compact: Heavy — handoff state + 10 context memories
Reads MEM0_API_KEY from ~/.claude/.mcp.json.
Detects project from git root directory name.
"""
import json, os, re, subprocess, sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
MEM0_API_URL = "https://api.mem0.ai/v1/memories/search/"
MCP_CONFIG = Path.home() / ".claude" / ".mcp.json"
MAX_NORMAL = 5
MAX_COMPACT = 10
LOOKBACK_DAYS = 7
EMPTY = json.dumps({})
# Map git directory names to mem0 agent_id tags
PROJECT_MAP = {
"StakTrakr": "staktrakr",
"HexTrackr": "hextrackr",
"MyMelo": "mymelo",
"Forge": "forge",
# Add your projects here
}
PREFERRED_SOURCES = {"stop-hook", "pre-compact-hook", "post-tool-use-hook"}
def get_api_key():
"""Read MEM0_API_KEY from .mcp.json."""
try:
with open(MCP_CONFIG) as f:
text = re.sub(r",\s*([}\]])", r"\1", f.read())
config = json.loads(text)
servers = config.get("mcpServers", config)
return servers.get("mem0", {}).get("env", {}).get("MEM0_API_KEY")
except Exception:
return None
def get_project():
"""Detect project from git root."""
try:
r = subprocess.run(["git", "rev-parse", "--show-toplevel"],
capture_output=True, text=True, timeout=5)
if r.returncode == 0:
name = os.path.basename(r.stdout.strip())
return name, PROJECT_MAP.get(name)
except Exception:
pass
name = os.path.basename(os.getcwd())
return name, PROJECT_MAP.get(name)
def search_mem0(api_key, query, limit, agent_id=None):
"""Search mem0 API, post-filter by project and date."""
since = (datetime.now(timezone.utc)
- timedelta(days=LOOKBACK_DAYS)).strftime("%Y-%m-%d")
payload = {"query": query, "user_id": "lbruton",
"limit": limit * 4 if agent_id else limit * 2}
req = Request(MEM0_API_URL, data=json.dumps(payload).encode(),
headers={"Authorization": f"Token {api_key}",
"Content-Type": "application/json"},
method="POST")
try:
with urlopen(req, timeout=10) as resp:
data = json.loads(resp.read().decode())
except Exception:
return []
if not isinstance(data, list):
data = data.get("results", [])
filtered = []
for mem in data:
if mem.get("created_at", "")[:10] < since:
continue
if agent_id:
proj = mem.get("metadata", {}).get("project", "").lower()
if proj and proj != agent_id:
continue
filtered.append(mem)
return filtered[:limit]
def filter_preferred(memories):
preferred = [m for m in memories
if m.get("metadata", {}).get("source") in PREFERRED_SOURCES]
return preferred if preferred else memories
def format_normal(memories, project_name):
if not memories:
return None
memories = filter_preferred(memories)
lines = [f"Recent mem0 context for {project_name} "
f"(last {LOOKBACK_DAYS} days):"]
for i, mem in enumerate(memories, 1):
text = mem.get("memory", "").strip()
if not text:
continue
if len(text) > 200:
text = text[:197] + "..."
date = mem.get("created_at", "")[:10]
mtype = mem.get("metadata", {}).get("type", "")
tag = f" ({mtype})" if mtype else ""
lines.append(f" {i}. [{date}]{tag} {text}")
return "\n".join(lines) if len(lines) > 1 else None
def format_compact(handoff, context, project_name):
sections = ["## Context restored after compaction\n"]
context = filter_preferred(context)
if handoff:
sections.append("### Session state before compaction")
for mem in handoff[:2]:
text = mem.get("memory", "").strip()
if text:
sections.append(text[:500])
sections.append("")
if context:
sections.append("### Recent project context")
for i, mem in enumerate(context, 1):
text = mem.get("memory", "").strip()
if not text:
continue
if len(text) > 200:
text = text[:197] + "..."
date = mem.get("created_at", "")[:10]
mtype = mem.get("metadata", {}).get("type", "")
tag = f" ({mtype})" if mtype else ""
sections.append(f" {i}. [{date}]{tag} {text}")
return "\n".join(sections) if len(sections) > 2 else None
def main():
source = "startup"
try:
raw = sys.stdin.read()
if raw:
source = json.loads(raw).get("source", "startup")
except Exception:
pass
api_key = get_api_key()
project_name, agent_id = get_project()
if not api_key or not agent_id:
print(EMPTY)
return
if source == "compact":
handoff = search_mem0(api_key,
"pre-compact handoff session state", 3, agent_id)
context = search_mem0(api_key,
"recent decision pattern insight task", MAX_COMPACT, agent_id)
result = format_compact(handoff, context, project_name)
else:
memories = search_mem0(api_key,
"recent session decision in-progress pattern", MAX_NORMAL,
agent_id)
result = format_normal(memories, project_name)
if not result:
print(EMPTY)
return
print(json.dumps({"hookSpecificOutput": {
"hookEventName": "SessionStart",
"additionalContext": result}}))
if __name__ == "__main__":
main()