IDE Plugins Family

Copy{
  "mcpServers": {
    "mem0": {
      "type": "stdio",
      "command": "python3",
      "args": [
        "${CURSOR_PLUGIN_ROOT}/core/mcp_server.py"
      ],
      "env": {
        "PLUGIN_OPTION_API_KEY": "${api_key}",
        "PLUGIN_OPTION_USER_ID": "${user_id}",
        "PLUGIN_OPTION_TOP_K": "${top_k}",
        "PLUGIN_OPTION_MAX_CONTEXT_CHARS": "${max_context_chars}",
        "PLUGIN_OPTION_SEARCH_SCOPE": "${search_scope}"
      }
    }
  }
}

Copy{
  "mcpServers": {
    "mem0": {
      "type": "stdio",
      "command": "python3",
      "args": ["./core/mcp_server.py"],
      "cwd": ".",
      "env_vars": [
        "MEM0_API_KEY",
        "MEM0_API_URL",
        "MEM0_CODE_USER_ID",
        "MEM0_USER_ID",
        "MEM0_RESOLVED_USER_ID",
        "MEM0_PROJECT_ID",
        "MEM0_TELEMETRY"
      ]
    }
  }
}

Copy{
  "mcpServers": {
    "mem0": {
      "command": "python3",
      "args": ["./core/mcp_server.py"]
    }
  }
}

Copy{
  "name": "mem0",
  "version": "0.3.3",
  "description": "Cross-session memory and token savings for coding agents.",
  "keywords": ["memory", "coding-agents", "continual-learning", "token-efficiency"],
  "author": { "name": "Mem0", "email": "support@mem0.ai" },
  "homepage": "https://docs.mem0.ai/integrations/kimi",
  "license": "Apache-2.0",
  "interface": {
    "displayName": "Mem0",
    "shortDescription": "Cross-session memory for Kimi Code"
  },
  "skills": "./skills/",
  "mcpServers": {
    "mem0": {
      "command": "python3",
      "args": ["./core/mcp_server.py"]
    }
  },
  "hooks": [
    { "event": "SessionStart", "command": "python3 ./hooks/adapter.py SessionStart", "timeout": 5 },
    { "event": "UserPromptSubmit", "command": "python3 ./hooks/adapter.py UserPromptSubmit", "timeout": 6 },
    { "event": "PostToolUse", "matcher": ".*", "command": "python3 ./hooks/adapter.py PostToolUse", "timeout": 3 },
    { "event": "PostToolUseFailure", "matcher": ".*", "command": "python3 ./hooks/adapter.py PostToolUseFailure", "timeout": 3 },
    { "event": "Stop", "command": "python3 ./hooks/adapter.py Stop", "timeout": 3 },
    { "event": "PreCompact", "command": "python3 ./hooks/adapter.py PreCompact", "timeout": 5 },
    { "event": "SessionEnd", "command": "python3 ./hooks/adapter.py SessionEnd", "timeout": 5 }
  ]
}

Copy{
  "id": "mem0",
  "version": "0.3.3",
  "homepage": "https://docs.mem0.ai/integrations/cursor",
  "native": {
    "pluginRoot": "${CURSOR_PLUGIN_ROOT}",
    "files": {
      ".cursor-plugin/plugin.json": ".cursor-plugin/plugin.json",
      "hooks/hooks.json": "hooks/hooks.json",
      "hooks/adapter.py": "hooks/adapter.py",
      "mcp.json": "mcp.json"
    }
  }
}

CopyEVENTS = {
    "SessionStart": ["session-start"],
    "UserPromptSubmit": ["user-prompt"],
    "PostToolUse": ["post-tool"],
    "PostToolUseFailure": ["post-tool-failure"],
    "Stop": ["stop"],
    "PreCompact": ["flush", "--reason", "pre-compact"],
    "SessionEnd": ["flush", "--reason", "session-end"],
}
 
 
def _session_dir(session_id: str) -> Path | None:
    home = Path(os.environ.get("KIMI_CODE_HOME", Path.home() / ".kimi-code")).expanduser()
    found = None
    try:
        with (home / "session_index.jsonl").open(encoding="utf-8") as index:
            for line in index:
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if entry.get("sessionId") == session_id and isinstance(entry.get("sessionDir"), str):
                    found = Path(entry["sessionDir"])
    except OSError:
        pass
    if found:
        return found
    if not session_id or Path(session_id).name != session_id:
        return None
    return next((path for path in (home / "sessions").glob(f"*/{session_id}") if path.is_dir()), None)
 
 
def _last_assistant_message(transcript: Path) -> str:
    message = ""
    step_id = None
    parts = []
    try:
        lines = transcript.read_text(encoding="utf-8").splitlines()
    except OSError:
        return ""
    for line in lines:
        try:
            record = json.loads(line)
        except json.JSONDecodeError:
            continue
        if record.get("type") != "context.append_loop_event" or not isinstance(record.get("event"), dict):
            continue
        event = record["event"]
        if event.get("type") == "step.begin":
            step_id = event.get("uuid")
            parts = []
        elif event.get("type") == "content.part" and event.get("stepUuid") == step_id:
            part = event.get("part")
            if isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str):
                parts.append(part["text"])
        elif event.get("type") == "step.end" and event.get("uuid") == step_id:
            if event.get("finishReason") not in {"error", "interrupted"} and parts:
                message = "".join(parts)
            step_id = None
            parts = []
    return message
 
 
def normalize(payload: dict) -> dict:
    value = dict(payload)
    if "tool_output" in value:
        value.setdefault("tool_response", value["tool_output"])
    if "error" in value:
        value.setdefault("tool_response", value["error"])
    if "response" in value:
        value.setdefault("last_assistant_message", value["response"])
    if value.get("hook_event_name") == "Stop":
        session = _session_dir(str(value.get("session_id") or ""))
        transcript = session / "agents" / "main" / "wire.jsonl" if session else None
        if transcript and transcript.is_file():
            value.setdefault("transcript_path", str(transcript))
            if message := _last_assistant_message(transcript):
                value.setdefault("last_assistant_message", message)
    return value

CopyEVENTS = {
    "sessionStart": "session-start",
    "beforeSubmitPrompt": "user-prompt",
    "postToolUse": "post-tool",
    "postToolUseFailure": "post-tool-failure",
    "afterAgentResponse": "assistant-stop",
    "stop": "stop",
    "sessionEnd": "session-end",
    "preCompact": "pre-compact",
}
 
 
def normalize(payload: dict, event: str) -> dict:
    value = dict(payload)
    value.setdefault("session_id", value.get("conversation_id") or value.get("parent_conversation_id", ""))
    roots = value.get("workspace_roots") or []
    if roots:
        value.setdefault("cwd", roots[0])
    if "tool_output" in value:
        value.setdefault("tool_response", value["tool_output"])
    if "error_message" in value:
        value.setdefault("tool_response", value["error_message"])
    if "text" in value:
        value.setdefault("last_assistant_message", value["text"])
    if "summary" in value:
        value.setdefault("last_assistant_message", value["summary"])
    return {"action": EVENTS[event], "payload": value}

Copydef _subagent_start(store, hook_input):
    context = record_subagent_start(store, hook_input)
    if context:
        return {
            "hookSpecificOutput": {
                "hookEventName": "SubagentStart",
                "additionalContext": context,
            }
        }
 
 
def _subagent_stop(store, hook_input):
    record_subagent_stop(store, hook_input)
 
 
def _post_tool(store, payload):
    response = payload.get("tool_response")
    failed = None
    if isinstance(response, dict):
        exit_code = response.get("exit_code")
        if isinstance(exit_code, int) and not isinstance(exit_code, bool):
            failed = exit_code != 0
        elif isinstance(response.get("isError"), bool):
            failed = response["isError"]
        elif isinstance(response.get("success"), bool):
            failed = not response["success"]
    if failed:
        payload = {**payload, "error": response}
    record_tool(store, payload, failed=failed)
 

Copydef read_hook_input() -> dict:
    try:
        value = json.load(sys.stdin)
        return value if isinstance(value, dict) else {}
    except (json.JSONDecodeError, OSError):
        return {}
 

Copydef default_record_stop(store: EvidenceStore, hook_input: dict):
    """Record the assistant's response without transcript parsing."""
    session_id = _session_id(hook_input)
    repo = store.repo_for_session(session_id, hook_input.get("cwd"))
    message = redact(hook_input.get("last_assistant_message", "")).strip()
    if message:
        store.record_assistant_response(repo, session_id, message)
    return repo, session_id
 

Copydef first_prompt_memory_output(store: EvidenceStore, hook_input: dict) -> dict:
    """Search once before the agent handles the first prompt in a session."""
    repo, session_id, prompt, is_first_prompt = record_user_prompt(store, hook_input)
    if not is_first_prompt:
        return {}
    try:
        minimum_query_chars = int(os.environ.get("MEM0_CODE_MIN_QUERY_CHARS", "20"))
    except ValueError:
        minimum_query_chars = 20
    if len(prompt.strip()) < max(minimum_query_chars, 1):
        return {}
    result = search_memories(
        store, repo, session_id, bounded(prompt, 6000),
        top_k=5, operation="first-prompt-search", timeout=2,
    )
    if not result.memories:
        return {}
    context = format_context(
        result.memories,
        "Mem0 found these relevant memories from earlier work in this repository:",
    )
    telemetry.record(
        "context_injected",
        repo=repo, session_id=session_id, trigger="first-prompt",
        memory_count=len(result.memories), context_chars=len(context),
        prompt_chars=len(prompt),
    )
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": context,
        },
    }

Copydef schedule_periodic_checkpoint(
    store: EvidenceStore, hook_input: dict, repo, session_id: str,
) -> bool:
    if (
        not automatic_flush_enabled()
        or not api_key()
        or not store.checkpoint_due(repo.identity, session_id)
    ):
        return False
    if store.prepare_flush(repo, session_id, "periodic") is None:
        return False
    hand_off_flush(hook_input, "periodic")
    return True

Copy"""Drop the cached key file once every configured key source is gone."""
configured = (
    os.environ.get("MEM0_API_KEY")
    or os.environ.get("PLUGIN_OPTION_API_KEY")
    or os.environ.get("CLAUDE_PLUGIN_OPTION_API_KEY")
    or os.environ.get("CLAUDE_PLUGIN_OPTION_MEM0_API_KEY")
    or ""
).strip()
if configured:
    return False
path = data_dir() / "api-key"
if not path.exists():
    return False
try:
    path.unlink()
except OSError:
    return False
return True

Copyclass EvidenceStore:
    def __init__(self, path: Path | None = None):
        directory = data_dir() if path is None else path.parent
        directory.mkdir(parents=True, exist_ok=True)
        self.path = path or directory / "evidence.sqlite3"
        try:
            self._open()
        except sqlite3.DatabaseError:
            self._quarantine()
            self._open()
 
    def _open(self) -> None:
        self.conn = sqlite3.connect(self.path, timeout=10)
        self.conn.row_factory = sqlite3.Row
        try:
            self.conn.execute("PRAGMA journal_mode=WAL")
            self.conn.execute("PRAGMA busy_timeout=10000")
            self._migrate()
        except sqlite3.DatabaseError:
            self.conn.close()
            raise

Copydef _checkpoint_message(event: dict[str, Any]) -> str:
    kind = event.get("kind")
    payload = event.get("payload") or {}
    if kind == "user_prompt":
        return redact(payload.get("text", "")).strip()
    if kind == "assistant_stop":
        transcript_messages = payload.get("transcript_messages") or []
        if isinstance(transcript_messages, list):
            text = "\n".join(
                str(message.get("content") or "")
                for message in transcript_messages
                if isinstance(message, dict) and message.get("content")
            )
            if text:
                return text
        return redact(payload.get("text", "")).strip()
    if kind in {"subagent_stop", "sidekick_stop"}:
        return redact(payload.get("final_message", "")).strip()
    return ""