Plugin Synchronization

Copy_CLAUDE_SETTINGS = Path.home() / ".claude" / "settings.json"
_SHELL_RCS = [Path.home() / ".zshrc", Path.home() / ".bashrc", Path.home() / ".bash_profile"]
 

Copyconst CLAUDE_SETTINGS = path.join(os.homedir(), ".claude", "settings.json");
const SHELL_RCS = [
	path.join(os.homedir(), ".zshrc"),
	path.join(os.homedir(), ".bashrc"),
	path.join(os.homedir(), ".bash_profile"),
];

Copy_RC_LINE = re.compile(
    r'^([ \t]*export[ \t]+MEM0_API_KEY[ \t]*=[ \t]*)(["\']?)([^"\'\n]*)(["\']?)[ \t]*$',
    re.MULTILINE,
)

Copyconst RC_LINE_RE =
	/^([ \t]*export[ \t]+MEM0_API_KEY[ \t]*=[ \t]*)(["']?)([^"'\n]*)(["']?)[ \t]*$/m;

Copydef _atomic_write_text(path: Path, content: str) -> None:
    """Write content to path atomically (temp + rename)."""
    dirname = path.parent
    fd, tmp_path = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=dirname)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
        # Preserve mode if the original existed.
        if path.exists():
            os.chmod(tmp_path, path.stat().st_mode & 0o777)
        os.replace(tmp_path, path)

Copyfunction atomicWriteText(filePath: string, content: string): void {
	const dir = path.dirname(filePath);
	const tmp = path.join(dir, `.${path.basename(filePath)}.${process.pid}.tmp`);
	try {
		fs.writeFileSync(tmp, content, "utf-8");
		// Preserve permissions if original existed.
		if (fs.existsSync(filePath)) {
			try {
				const mode = fs.statSync(filePath).mode & 0o777;
				fs.chmodSync(tmp, mode);
			} catch {
				/* best-effort */
			}
		}
		fs.renameSync(tmp, filePath);
	} catch (err) {

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

Copy{
  "$schema": "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json",
  "mcpServers": {
    "mem0": {
      "type": "stdio",
      "command": "python3",
      "args": [
        "${PLUGIN_ROOT}/core/mcp_server.py"
      ]
    }
  }

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

Copydef main() -> int:
    for raw_line in sys.stdin:
        try:
            message = json.loads(raw_line)
            response = handle_request(message)
        except json.JSONDecodeError:
            response = {
                "jsonrpc": "2.0",
                "id": None,
                "error": {"code": -32700, "message": "Parse error"},
            }
        except Exception:
            response = {
                "jsonrpc": "2.0",
                "id": None,
                "error": {"code": -32603, "message": "Internal error"},
            }
        if response is not None:
            sys.stdout.write(json.dumps(response, separators=(",", ":")) + "\n")
            sys.stdout.flush()
    telemetry.spawn_flush()
    return 0
 
 

Copydef handle_request(message: Any) -> dict[str, Any] | None:
    if not isinstance(message, dict):
        return None
    request_id = message.get("id")
    method = message.get("method")
 
    if method == "notifications/initialized":
        return None
    if method == "initialize":
        requested = (message.get("params") or {}).get("protocolVersion")
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "protocolVersion": requested or PROTOCOL_VERSION,
                "capabilities": {"tools": {"listChanged": False}},
                "serverInfo": {"name": "mem0", "version": PLUGIN_VERSION},
            },
        }
    if method == "ping":
        return {"jsonrpc": "2.0", "id": request_id, "result": {}}

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."
)
TOOL_SCHEMA = {
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "minLength": 1,
            "maxLength": 2000,
            "description": "A direct question about earlier work in this repository.",
        },
        "top_k": {
            "type": "integer",
            "minimum": 1,
            "maximum": 20,
            "description": "Maximum memories to return. Uses Mem0's configured default when omitted.",
        },
        "category": {
            "type": "string",
            "enum": list(CODING_MEMORY_CATEGORY_NAMES),
            "description": "Optional memory category. Omit to search every category.",
        },
        "scope": {
            "type": "string",
            "enum": list(SEARCH_SCOPES),
            "description": (
                "Which memories to search. 'repo' (default) is the whole repository's "
                "shared memory plus your own preferences, 'dir' narrows the shared "
                "part to the current directory, 'mine' is your preferences alone."
            ),
        },
        "run_id": {
            "type": "string",
            "minLength": 1,
            "description": (
                "Optional coding-agent session ID. With any scope, restricts results to memories "
                "saved in that session. Omit to recall memories across sessions."
            ),
        },
    },
    "required": ["query"],
    "additionalProperties": False,
}

Copydef _validate_arguments(
    arguments: Any,
) -> tuple[str, int | None, str | None, str | None, str | None]:
    if not isinstance(arguments, dict):
        raise ToolInputError("Search arguments must be an object.")
 
    unknown = set(arguments) - {"query", "top_k", "category", "scope", "run_id"}
    if unknown:
        raise ToolInputError(f"Unknown search argument: {sorted(unknown)[0]}")
 
    query = arguments.get("query")
    if not isinstance(query, str) or not query.strip():
        raise ToolInputError("query must be a non-empty string.")
    query = query.strip()
    if len(query) > 2000:
        raise ToolInputError("query must be at most 2,000 characters.")
 
    top_k = arguments.get("top_k")
    if top_k is not None and (
        isinstance(top_k, bool) or not isinstance(top_k, int) or not 1 <= top_k <= 20
    ):
        raise ToolInputError("top_k must be an integer from 1 to 20.")
 
    category = arguments.get("category")
    if category is not None and category not in CODING_MEMORY_CATEGORY_NAMES:
        raise ToolInputError("category must be one of Mem0's supported categories.")
 
    scope = arguments.get("scope")
    if scope is not None and scope not in SEARCH_SCOPES:
        raise ToolInputError(f"scope must be one of {list(SEARCH_SCOPES)}.")
 
    run_id = arguments.get("run_id")
    if run_id is not None:
        if not isinstance(run_id, str) or not run_id.strip():
            raise ToolInputError("run_id must be a non-empty string.")
        run_id = run_id.strip()
 
    return query, top_k, category, scope, run_id
 

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

Copyconfigure_harness("codex", data_dir_name="codex-plugin", source_tag="codex_plugin")
telemetry.init(harness="codex", source_tag="CODEX_PLUGIN")