Claude Code Plugin

Copy{
  "mcpServers": {
    "mem0": {
      "command": "python3",
      "args": ["${CLAUDE_PLUGIN_ROOT}/core/mcp_server.py"],
      "env": {
        "MEM0_CODE_DATA_DIR": "${CLAUDE_PLUGIN_DATA}"
      }
    }
  }
}

CopyPROTOCOL_VERSION = "2024-11-05"
TOOL_NAME = "search_memories"
TOOL_DESCRIPTION = (
    "Search memories from earlier work in this repository. Use it before "
    "repeating investigation or when earlier decisions, fixes, commands, or "
    "results may help."
)

Copyif method == "tools/list":
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "result": {
            "tools": [
                {
                    "name": TOOL_NAME,
                    "description": TOOL_DESCRIPTION,
                    "inputSchema": TOOL_SCHEMA,
                    "annotations": {
                        "readOnlyHint": True,
                        "idempotentHint": True,
                        "openWorldHint": True,
                    },
                }
            ]
        },
    }

Copydef _transcript_rows(
    path: str, offset: int = 0
) -> tuple[list[dict[str, Any]], int, bool]:
    """Parse transcript rows from a byte offset, returning rows, end offset, and whether the offset was honored."""
    if not path:
        return [], 0, False
    rows = []
    try:
        resolved = Path(path).expanduser()
        if not 0 <= offset <= resolved.stat().st_size:
            offset = 0
        end = offset
        with resolved.open("rb") as handle:
            handle.seek(offset)
            for line in handle:
                if not line.endswith(b"\n"):
                    break
                end += len(line)
                try:
                    row = json.loads(line.decode("utf-8", errors="replace"))
                except json.JSONDecodeError:
                    continue
                if isinstance(row, dict) and row.get("uuid"):
                    rows.append(row)
    except OSError:
        return [], offset, False
    return rows, end, offset > 0
 

Copydef _active_transcript_chain(
    rows: list[dict[str, Any]], session_id: str
) -> list[dict[str, Any]]:
    """Follow the current Claude conversation branch from its latest record."""
    by_uuid = {str(row["uuid"]): row for row in rows if row.get("uuid")}
    leaf = next(
        (
            row
            for row in reversed(rows)
            if not row.get("isSidechain")
            and str(row.get("sessionId") or "") == session_id
        ),
        None,
    )
    if leaf is None:
        return []
 
    chain = []
    seen = set()
    current = leaf
    while current is not None:
        uuid = str(current.get("uuid") or "")
        if not uuid or uuid in seen:
            break
        seen.add(uuid)
        chain.append(current)
        current = by_uuid.get(str(current.get("parentUuid") or ""))
    chain.reverse()
    return chain

Copy 
_here = Path(__file__).resolve()
_bundled_core = _here.parents[2] / "core"
_core_dir = _bundled_core if (_bundled_core / "memory_core.py").is_file() else _bundled_core / "python"
sys.path.insert(0, str(_core_dir))
sys.path.insert(0, str(_here.parent))
 

Copyconfigure_harness("claude-code", data_dir_name="claude-code-plugin", source_tag="claude_code_plugin")
telemetry.init(harness="claude-code", source_tag="CLAUDE_CODE_PLUGIN")

Copydef _sidekick_start(store, hook_input):
    context = record_subagent_start(store, {"agent_type": "mem0:sidekick", **hook_input})
    if context:
        return {
            "hookSpecificOutput": {
                "hookEventName": "SubagentStart",
                "additionalContext": context,
            },
        }
 

Copydef _sidekick_stop(store, hook_input):
    record_subagent_stop(store, {"agent_type": "mem0:sidekick", **hook_input})

CopyDEFAULT_API_URL = "https://api.mem0.ai"
PLUGIN_VERSION = "0.3.3"
 
_harness_name: str = "generic"
_harness_env_prefix: str = "MEM0_PLUGIN"
_harness_data_dir_name: str = "mem0-plugin"
_harness_source_tag: str = "mem0_plugin"
 

Copydef configure_harness(
    name: str,
    env_prefix: str = "",
    data_dir_name: str = "",
    source_tag: str = "",
) -> None:
    global _harness_name, _harness_env_prefix, _harness_data_dir_name, _harness_source_tag
    _harness_name = name
    _harness_env_prefix = env_prefix or f"MEM0_{name.upper().replace('-', '_')}"
    _harness_data_dir_name = data_dir_name or f"{name}-plugin"
    _harness_source_tag = source_tag or f"{name.replace('-', '_')}_plugin"
 

CopyPROJECT_MEMORY_INSTRUCTIONS = """Save concise repository facts that will help with future coding work.
 
A completed change should produce one memory explaining the resulting behavior, where it is implemented when useful, and any important constraints or reasoning. Exploration or accepted decisions may produce separate memories only when they are independently useful.
 
Use the coding agent's final response for conclusions about current repository behavior. Do not save proposed or recommended changes unless the user accepted them or the coding agent completed them. Treat subagent responses as supporting repository evidence, not as decisions.
 
Write about the repository, not the user, assistant, session, or task. Do not include test results, documentation updates, release notes, or temporary state.
 
If nothing useful was established, return no memories."""

CopyPERSONAL_MEMORY_INSTRUCTIONS = """Save concise facts about the user that will help in any repository: preferred tools, package managers, languages, coding style, review and communication preferences, and anything the user explicitly asked to be remembered about themselves.
 
Write in the third person about the user, not about the repository, the assistant, the session, or the task. Do not save repository facts, project decisions, commands, or what was built.
 
Never save that the user has no preferences or that nothing was learned. If nothing was learned about the user, return no memories."""

CopyTEST_COMMAND_RE = re.compile(
    r"(?:^|\s)(?:pytest|py\.test|jest|vitest|go\s+test|cargo\s+test|"
    r"npm\s+(?:run\s+)?test|pnpm\s+(?:run\s+)?test|yarn\s+test|"
    r"mvn\s+test|gradle\s+test|make\s+test)(?:\s|$)",
    re.IGNORECASE,
)
BUILD_COMMAND_RE = re.compile(
    r"(?:^|\s)(?:npm|pnpm|yarn)\s+(?:run\s+)?build(?:\s|$)|"
    r"(?:^|\s)(?:cargo|go|mvn|gradle|make)\s+build(?:\s|$)",
    re.IGNORECASE,
)

CopySECRET_PATTERNS = [
    re.compile(r"(?i)(authorization\s*[:=]\s*(?:bearer|token)\s+)[^\s\"']+"),
    re.compile(
        r"(?i)((?:api[_-]?key|secret[_-]?access[_-]?key|session[_-]?token)\s*[:=]\s*)[^\s\"']+"
    ),
    re.compile(
        r"(?i)((?:access[_-]?token|refresh[_-]?token|password|credential)"
        r"\s*[:=]\s*)[^\s&\"']+"
    ),
    re.compile(r"\b(?:sk|m0|mem0_sk|psk)-[A-Za-z0-9_\-]{12,}\b"),
    re.compile(r"\b(?:ASIA|AKIA)[A-Z0-9]{12,}\b"),
    re.compile(r"\b(?:ghp_|github_pat_|xox[baprs]-)[A-Za-z0-9_\-]{12,}\b"),
    re.compile(
        r"-----BEGIN [^-]*PRIVATE KEY-----.*?-----END [^-]*PRIVATE KEY-----",
        re.DOTALL,
    ),
    re.compile(
        r'(?i)("(?:api[_-]?key|password|secret(?:[_-]?access[_-]?key)?'
        r'|(?:access|refresh|session)[_-]?token|token|authorization|credential'
        r')"\s*:\s*")(?:\\.|[^"\\])*'
    ),
]

Copydef redact(value: Any) -> str:
    text = (
        value
        if isinstance(value, str)
        else json.dumps(value, ensure_ascii=False, default=str)
    )
    for pattern in SECRET_PATTERNS:
        if pattern.groups:
            text = pattern.sub(r"\1[REDACTED]", text)
        else:
            text = pattern.sub("[REDACTED]", text)
    return text

Copydef bounded(value: Any, limit: int) -> str:
    text = redact(value).strip()
    if len(text) <= limit:
        return text
    return text[:limit] + f"\n...[truncated {len(text) - limit} chars]"
 

Copydef main() -> int:
    if len(sys.argv) != 2:
        return 2
    handoff_path = Path(sys.argv[1])
    os.environ["MEM0_CODE_HANDOFF_PATH"] = str(handoff_path)
    harness = os.environ.get("MEM0_PLUGIN_HARNESS")
    if harness:
        source_tag = os.environ.get("MEM0_PLUGIN_SOURCE_TAG", "")
        configure_harness(
            harness,
            env_prefix=os.environ.get("MEM0_PLUGIN_ENV_PREFIX", ""),
            data_dir_name=os.environ.get("MEM0_PLUGIN_DATA_DIR_NAME", ""),
            source_tag=source_tag,
        )
        telemetry.init(harness=harness, source_tag=source_tag.upper())

Copytry:
    payload = json.loads(handoff_path.read_text(encoding="utf-8"))
    delay = float(payload.get("delay_seconds") or 0)
    if delay > 0:
        payload.pop("delay_seconds", None)
        temporary = handoff_path.with_suffix(f".{os.getpid()}.tmp")
        try:
            temporary.write_text(json.dumps(payload), encoding="utf-8")
            temporary.replace(handoff_path)
        finally:
            temporary.unlink(missing_ok=True)
        time.sleep(delay)
        if not handoff_path.exists():
            return 0

Copyif wait_for_inflight:
    session_id = str(
        hook_input.get("session_id") or "unknown-session"
    )
    repo = store.repo_for_session(session_id, hook_input.get("cwd"))
    deadline = time.monotonic() + float(
        os.environ.get("MEM0_CODE_EXTRACTION_WAIT_SECONDS", "120")
    )
    while (
        store.has_inflight_flush(repo.identity, session_id)
        and time.monotonic() < deadline
    ):
        touch_handoff_heartbeat()
        time.sleep(0.25)

Copy# Hooks capture the conversation before handoff; the worker only flushes it.
result = checkpoint_session(store, hook_input, reason)
print(json.dumps(result, sort_keys=True), flush=True)
completed = result.get("status") in {
    "semantic-succeeded",
    "explicitly-stored",
    "nothing-to-flush",
}

Copyfinally:
    telemetry.flush()
    if completed:
        try:
            handoff_path.unlink()
        except OSError:
            pass
    elif handoff_path.suffix == ".running":
        try:
            handoff_path.replace(handoff_path.with_suffix(".json"))
        except OSError:
            pass