diff --git a/build.py b/build.py
index 2578413..cf87110 100644
--- a/build.py
+++ b/build.py
@@ -10,6 +10,8 @@
import PyInstaller.__main__
+from config.platform_names import normalize_arch, normalize_system
+
sep = os.pathsep
@@ -66,17 +68,11 @@ def build_version_file():
def normalize_architecture(arch: str) -> str:
- if arch in ['x86_64', 'amd64']:
- return 'amd64'
- elif arch in ['aarch64', 'arm64']:
- return 'arm64'
- elif arch.startswith('arm'):
- return 'arm64'
- return arch
+ return normalize_arch(arch)
def normalize_system_name(system: str) -> str:
- return "macos" if system == 'darwin' else system
+ return normalize_system(system)
def get_binary_name(system: str, arch: str) -> str:
@@ -99,6 +95,11 @@ def run_pyinstaller(name: str, icon: str):
"tools.ai_scriptless.step_path",
"tools.ai_scriptless.tree",
"tools.ai_scriptless.variables",
+ "update",
+ "update.flow",
+ "update.install",
+ "update.processes",
+ "update.release",
]
PyInstaller.__main__.run([
'main.py',
diff --git a/config/http.py b/config/http.py
new file mode 100644
index 0000000..6f979bf
--- /dev/null
+++ b/config/http.py
@@ -0,0 +1,17 @@
+"""Shared HTTP client defaults for Perfecto MCP (no dependency on tools.*)."""
+
+import platform
+
+import httpx
+
+from config.version import __version__
+
+_ua_part = f"{platform.system()} {platform.release()}; {platform.machine()}"
+user_agent = f"perfecto-mcp/{__version__} ({_ua_part})"
+
+timeout = httpx.Timeout(
+ connect=15.0,
+ read=60.0,
+ write=15.0,
+ pool=60.0,
+)
diff --git a/config/platform_names.py b/config/platform_names.py
new file mode 100644
index 0000000..6f9c51b
--- /dev/null
+++ b/config/platform_names.py
@@ -0,0 +1,19 @@
+"""Platform name normalization shared by packaging and the updater."""
+
+
+def normalize_system(system: str) -> str:
+ system = system.lower()
+ if system == "darwin":
+ return "macos"
+ if system == "windows":
+ return "windows"
+ return "linux"
+
+
+def normalize_arch(machine: str) -> str:
+ machine = machine.lower()
+ if machine in {"x86_64", "amd64"}:
+ return "amd64"
+ if machine in {"aarch64", "arm64"} or machine.startswith("arm"):
+ return "arm64"
+ return machine
diff --git a/main.py b/main.py
index 016e47b..9a6d854 100644
--- a/main.py
+++ b/main.py
@@ -5,7 +5,7 @@
import sys
from typing import Literal, cast
-from mcp.server.fastmcp import FastMCP, Icon
+from mcp.server.fastmcp import FastMCP
from config.perfecto import SECURITY_TOKEN_FILE_ENV_NAME, SECURITY_TOKEN_ENV_NAME, PERFECTO_CLOUD_NAME_ENV_NAME, \
GITHUB
@@ -32,6 +32,19 @@ def init_logging(level_name: str) -> None:
)
+def _banner() -> str:
+ return (
+ " _____ __ _ \n"
+ " | __ \\ / _| | | \n"
+ " | |__) |__ _ __| |_ ___ ___| |_ ___ \n"
+ " | ___/ _ \\ '__| _/ _ \\/ __| __/ _ \\ \n"
+ " | | | __/ | | || __/ (__| || (_) |\n"
+ " |_| \\___|_| |_| \\___|\\___|\\__\\___/ \n"
+ " \n"
+ f" Perfecto MCP Server v{__version__} \n"
+ )
+
+
def get_token() -> PerfectoToken:
global PERFECTO_SECURITY_TOKEN_FILE_PATH, PERFECTO_SECURITY_TOKEN, PERFECTO_CLOUD_NAME, PERFECTO_SECURITY_TOKEN_FILE_NAME
@@ -93,6 +106,12 @@ def main():
help="Execute MCP Server"
)
+ parser.add_argument(
+ "--update",
+ action="store_true",
+ help="Run the interactive manual updater (frozen binary builds only)"
+ )
+
parser.add_argument(
"--log-level",
default="CRITICAL", # By default, only critical errors
@@ -105,18 +124,15 @@ def main():
if args.mcp:
init_logging(args.log_level)
run(log_level=args.log_level.upper())
+ elif args.update:
+ from update.flow import run_interactive_update
+
+ logo_ascii = _banner()
+ print(logo_ascii)
+ raise SystemExit(run_interactive_update())
else:
- logo_ascii = (
- " _____ __ _ \n"
- " | __ \ / _| | | \n"
- " | |__) |__ _ __| |_ ___ ___| |_ ___ \n"
- " | ___/ _ \ '__| _/ _ \/ __| __/ _ \ \n"
- " | | | __/ | | || __/ (__| || (_) |\n"
- " |_| \___|_| |_| \___|\___|\__\___/ \n"
- " \n"
- f" Perfecto MCP Server v{__version__} \n"
- )
+ logo_ascii = _banner()
print(logo_ascii)
if PERFECTO_CLOUD_NAME is None:
@@ -129,9 +145,9 @@ def main():
else:
command_path = __executable__
command = "uvx" if __uvx__ else command_path
- args = ["--mcp"]
+ mcp_args = ["--mcp"]
if __uvx__:
- args = [
+ mcp_args = [
"--from", f"git+{GITHUB}.git@v{get_version()}",
"-q", "perfecto-mcp",
"--mcp"
@@ -140,7 +156,7 @@ def main():
config_dict = {
"Perfecto MCP": {
"command": f"{command}",
- "args": args,
+ "args": mcp_args,
"env": {
f"{PERFECTO_CLOUD_NAME_ENV_NAME}": f"{perfecto_environment_str}"
}
@@ -171,7 +187,17 @@ def main():
print(" https://github.com/PerfectoCode/perfecto-mcp/")
print(" ")
- input("Press Enter to exit...")
+ if getattr(sys, "frozen", False) and not __uvx__:
+ print(" Updates:")
+ print(" To install a newer Perfecto MCP build, quit every MCP client using this server,")
+ print(" then re-run this app with --update (or choose Update below) and follow the prompts.")
+ print(" ")
+ choice = input("Press Enter to exit, or type 'u' + Enter to check for updates: ").strip().lower()
+ if choice in {"u", "update"}:
+ from update.flow import run_interactive_update
+ raise SystemExit(run_interactive_update())
+ else:
+ input("Press Enter to exit...")
if __name__ == "__main__":
diff --git a/pyproject.toml b/pyproject.toml
index 4d10b99..f075ae5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "perfecto-mcp"
-version = "1.1.1"
+version = "1.1.0"
description = "MCP server for Perfecto Cloud Platform"
readme = "README.md"
requires-python = ">=3.11"
@@ -33,7 +33,7 @@ dev = [
[tool.setuptools.packages.find]
where = ["."]
-include = ["tools*", "config", "models", "formatters", "resources"]
+include = ["tools*", "config", "models", "formatters", "resources", "update"]
[tool.setuptools.package-data]
"resources" = ["*.png"]
diff --git a/tests/test_tools_manager.py b/tests/test_tools_manager.py
index 99cf12a..424d443 100644
--- a/tests/test_tools_manager.py
+++ b/tests/test_tools_manager.py
@@ -20,10 +20,9 @@
from tools.tools_manager import (
ToolsManager,
- _match_recommended_asset,
- _normalize_arch,
- _normalize_system,
)
+from config.platform_names import normalize_arch, normalize_system
+from update.release import match_recommended_asset
def _make_ctx():
@@ -46,13 +45,13 @@ def test_version_returns_current_build_metadata(perfecto_token):
def test_normalize_platform_helpers():
- assert _normalize_system("Darwin") == "macos"
- assert _normalize_system("Windows") == "windows"
- assert _normalize_system("Linux") == "linux"
- assert _normalize_arch("x86_64") == "amd64"
- assert _normalize_arch("amd64") == "amd64"
- assert _normalize_arch("arm64") == "arm64"
- assert _normalize_arch("aarch64") == "arm64"
+ assert normalize_system("Darwin") == "macos"
+ assert normalize_system("Windows") == "windows"
+ assert normalize_system("Linux") == "linux"
+ assert normalize_arch("x86_64") == "amd64"
+ assert normalize_arch("amd64") == "amd64"
+ assert normalize_arch("arm64") == "arm64"
+ assert normalize_arch("aarch64") == "arm64"
def test_match_recommended_asset_prefers_zip():
@@ -66,7 +65,7 @@ def test_match_recommended_asset_prefers_zip():
"browser_download_url": "https://example.com/zip",
},
]
- matched = _match_recommended_asset(assets, "macos", "arm64")
+ matched = match_recommended_asset(assets, "macos", "arm64")
assert matched["name"] == "perfecto-mcp-macos-arm64.zip"
@@ -97,10 +96,21 @@ def test_check_updates_when_latest_is_newer(perfecto_token):
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
+ runtime = {
+ "frozen": True,
+ "uvx": False,
+ "docker": False,
+ "executable": "/tmp/perfecto-mcp",
+ "bundle": "/tmp/perfecto-mcp.app",
+ }
+
with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client), \
patch("tools.tools_manager.__version__", "1.0.0"), \
patch("tools.tools_manager.platform.system", return_value="Darwin"), \
- patch("tools.tools_manager.platform.machine", return_value="arm64"):
+ patch("tools.tools_manager.platform.machine", return_value="arm64"), \
+ patch("update.release.platform.system", return_value="Darwin"), \
+ patch("update.release.platform.machine", return_value="arm64"), \
+ patch("tools.tools_manager._detect_runtime", return_value=runtime):
manager = ToolsManager(perfecto_token, _make_ctx())
result = asyncio.run(manager.check_updates())
@@ -112,6 +122,42 @@ def test_check_updates_when_latest_is_newer(perfecto_token):
assert payload["release"]["body"] == "Release notes for 9.9.9"
assert payload["recommended_asset"]["name"] == "perfecto-mcp-macos-arm64.zip"
assert payload["update_guidance"]["status"] == "update_available"
+ assert "double-click" in payload["update_guidance"]["automatic"].lower()
+ assert any("update_status" in message for message in result.info)
+
+
+def test_check_updates_source_runtime_guidance(perfecto_token):
+ release = {
+ "tag_name": "v9.9.9",
+ "name": "v9.9.9",
+ "html_url": "https://github.com/PerfectoCode/perfecto-mcp/releases/tag/v9.9.9",
+ "published_at": "2026-07-14T00:00:00Z",
+ "body": "notes",
+ "assets": [],
+ }
+ mock_response = MagicMock()
+ mock_response.raise_for_status = MagicMock()
+ mock_response.json = MagicMock(return_value=release)
+ mock_client = AsyncMock()
+ mock_client.get = AsyncMock(return_value=mock_response)
+ mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+ mock_client.__aexit__ = AsyncMock(return_value=False)
+
+ runtime = {
+ "frozen": False,
+ "uvx": False,
+ "docker": False,
+ "executable": "main.py",
+ "bundle": "main.py",
+ }
+ with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client), \
+ patch("tools.tools_manager.__version__", "1.0.0"), \
+ patch("tools.tools_manager._detect_runtime", return_value=runtime):
+ result = asyncio.run(ToolsManager(perfecto_token, _make_ctx()).check_updates())
+
+ automatic = result.result[0]["update_guidance"]["automatic"].lower()
+ assert "frozen" in automatic or "source" in automatic
+ assert "double-click" not in automatic
def test_check_updates_when_up_to_date(perfecto_token):
diff --git a/tests/test_update.py b/tests/test_update.py
new file mode 100644
index 0000000..e48c263
--- /dev/null
+++ b/tests/test_update.py
@@ -0,0 +1,294 @@
+"""
+Copyright 2025 Perforce Software, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+import asyncio
+import hashlib
+import zipfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from update.install import copy_into_place_now, write_and_spawn_installer
+from update.processes import (
+ RunningProcess,
+ find_other_instances,
+ format_process_list,
+ _matches_perfecto_mcp,
+)
+from update.release import (
+ extract_update_payload,
+ match_recommended_asset,
+ verify_sha256,
+)
+from update.flow import describe_manual_update_instructions, run_interactive_update
+
+
+def test_match_recommended_asset_prefers_zip():
+ assets = [
+ {"name": "perfecto-mcp-linux-amd64", "browser_download_url": "https://example.com/bin"},
+ {"name": "perfecto-mcp-linux-amd64.zip", "browser_download_url": "https://example.com/zip"},
+ ]
+ matched = match_recommended_asset(assets, "linux", "amd64")
+ assert matched["name"] == "perfecto-mcp-linux-amd64.zip"
+
+
+def test_extract_update_payload_from_zip(tmp_path: Path):
+ archive = tmp_path / "payload.zip"
+ with zipfile.ZipFile(archive, "w") as zf:
+ zf.writestr("perfecto-mcp-linux-amd64", b"binary-bytes")
+
+ extracted = extract_update_payload(archive, tmp_path / "out")
+ assert extracted.name == "perfecto-mcp-linux-amd64"
+ assert extracted.read_bytes() == b"binary-bytes"
+
+
+def test_extract_update_payload_prefers_app_bundle(tmp_path: Path):
+ archive = tmp_path / "app.zip"
+ with zipfile.ZipFile(archive, "w") as zf:
+ zf.writestr("perfecto-mcp-arm64.app/Contents/MacOS/perfecto-mcp", b"app-bin")
+ zf.writestr("perfecto-mcp-arm64.app/Contents/Info.plist", b"")
+
+ extracted = extract_update_payload(archive, tmp_path / "out")
+ assert extracted.name.endswith(".app")
+ assert (extracted / "Contents" / "MacOS" / "perfecto-mcp").read_bytes() == b"app-bin"
+
+
+def test_extract_update_payload_ignores_macosx_appledouble(tmp_path: Path):
+ archive = tmp_path / "app.zip"
+ with zipfile.ZipFile(archive, "w") as zf:
+ zf.writestr("perfecto-mcp-arm64.app/Contents/MacOS/perfecto-mcp", b"real-bin")
+ zf.writestr("perfecto-mcp-arm64.app/Contents/Info.plist", b"")
+ zf.writestr("__MACOSX/perfecto-mcp-arm64.app/Contents/MacOS/._perfecto-mcp", b"junk")
+ zf.writestr("__MACOSX/perfecto-mcp-arm64.app/Contents/._MacOS", b"junk")
+
+ extracted = extract_update_payload(archive, tmp_path / "out")
+ assert "__MACOSX" not in extracted.parts
+ assert (extracted / "Contents" / "MacOS" / "perfecto-mcp").read_bytes() == b"real-bin"
+
+
+def test_verify_sha256_accepts_matching_digest(tmp_path: Path):
+ payload = tmp_path / "perfecto-mcp-linux-amd64"
+ payload.write_bytes(b"abc")
+ digest = hashlib.sha256(b"abc").hexdigest()
+ verify_sha256(payload, f"{digest} perfecto-mcp-linux-amd64\n")
+
+
+def test_verify_sha256_rejects_mismatch(tmp_path: Path):
+ payload = tmp_path / "bin"
+ payload.write_bytes(b"abc")
+ try:
+ verify_sha256(payload, "0" * 64 + " bin\n")
+ assert False, "expected ValueError"
+ except ValueError as exc:
+ assert "mismatch" in str(exc).lower()
+
+
+def test_format_process_list():
+ text = format_process_list([
+ RunningProcess(pid=11, name="perfecto-mcp", command="/tmp/perfecto-mcp --mcp"),
+ ])
+ assert "PID 11" in text
+ assert "perfecto-mcp --mcp" in text
+
+
+def test_find_other_instances_excludes_self():
+ fake = [
+ RunningProcess(pid=1, name="perfecto-mcp", command="perfecto-mcp --mcp"),
+ RunningProcess(pid=2, name="perfecto-mcp", command="perfecto-mcp --update"),
+ ]
+ with patch("update.processes.list_perfecto_mcp_processes", return_value=fake), \
+ patch("update.processes.os.getpid", return_value=1), \
+ patch("update.processes.os.getppid", return_value=99):
+ others = find_other_instances()
+ assert [p.pid for p in others] == [2]
+
+
+def test_matches_ignores_cursor_and_path_mentions():
+ assert _matches_perfecto_mcp(
+ "Cursor",
+ "Cursor Helper (Plugin): extension-host (user) perfecto-mcp [1-35]",
+ ) is False
+ assert _matches_perfecto_mcp(
+ "bash",
+ "bash -c cd /Users/diego/Documents/perfecto-mcp && pytest",
+ ) is False
+ assert _matches_perfecto_mcp("perfecto-mcp", "perfecto-mcp --mcp") is True
+ assert _matches_perfecto_mcp(
+ "perfecto-mcp",
+ "/Applications/perfecto-mcp-arm64.app/Contents/MacOS/perfecto-mcp --mcp",
+ ) is True
+ assert _matches_perfecto_mcp(
+ "launcher.sh",
+ "/Applications/perfecto-mcp-arm64.app/Contents/MacOS/launcher.sh",
+ ) is True
+ assert _matches_perfecto_mcp(
+ "bash",
+ "bash -lc '/opt/perfecto-mcp --mcp'",
+ ) is True
+ assert _matches_perfecto_mcp(
+ "env",
+ "env FOO=1 /usr/local/bin/perfecto-mcp --mcp",
+ ) is True
+
+
+def test_copy_into_place_replaces_file(tmp_path: Path):
+ source = tmp_path / "new-bin"
+ target = tmp_path / "old-bin"
+ source.write_bytes(b"new")
+ target.write_bytes(b"old")
+ copy_into_place_now(source, target)
+ assert target.read_bytes() == b"new"
+
+
+def test_copy_into_place_replaces_app_inner_binary(tmp_path: Path):
+ app = tmp_path / "perfecto-mcp-arm64.app"
+ inner = app / "Contents" / "MacOS" / "perfecto-mcp"
+ inner.parent.mkdir(parents=True)
+ inner.write_bytes(b"old")
+ source = tmp_path / "new-bin"
+ source.write_bytes(b"new")
+ copy_into_place_now(source, app)
+ assert inner.read_bytes() == b"new"
+
+
+def test_write_and_spawn_installer_unix(tmp_path: Path, monkeypatch):
+ source = tmp_path / "src-bin"
+ target = tmp_path / "dst-bin"
+ source.write_bytes(b"x")
+ target.write_bytes(b"y")
+ spawned = {}
+
+ def fake_mkstemp(prefix="tmp", suffix=""):
+ import os
+ path = tmp_path / f"{prefix}x{suffix}"
+ fd = os.open(path, os.O_RDWR | os.O_CREAT)
+ return fd, str(path)
+
+ def fake_popen(cmd, **kwargs):
+ spawned["cmd"] = cmd
+ spawned["kwargs"] = kwargs
+ return MagicMock()
+
+ monkeypatch.setattr("update.install.tempfile.mkstemp", fake_mkstemp)
+ monkeypatch.setattr("update.install.subprocess.Popen", fake_popen)
+ monkeypatch.setattr("update.install.platform.system", lambda: "Linux")
+
+ script = write_and_spawn_installer(source=source, target=target, wait_for_pid=12345)
+ assert script.exists()
+ content = script.read_text(encoding="utf-8")
+ assert "12345" in content
+ assert str(source.resolve()) in content
+ assert str(target.resolve()) in content
+ assert "restoring backup" in content.lower()
+ assert "Ensuring no other Perfecto MCP processes remain" in content
+ assert spawned["cmd"][0] == "nohup"
+ assert "/bin/bash" in spawned["cmd"]
+
+
+def test_write_and_spawn_installer_darwin_opens_terminal(tmp_path: Path, monkeypatch):
+ source = tmp_path / "src.app"
+ (source / "Contents" / "MacOS").mkdir(parents=True)
+ (source / "Contents" / "MacOS" / "perfecto-mcp").write_bytes(b"new")
+ target = tmp_path / "dst.app"
+ (target / "Contents" / "MacOS").mkdir(parents=True)
+ (target / "Contents" / "MacOS" / "perfecto-mcp").write_bytes(b"old")
+ spawned = {}
+
+ def fake_mkstemp(prefix="tmp", suffix=""):
+ import os
+ path = tmp_path / f"{prefix}x{suffix}"
+ fd = os.open(path, os.O_RDWR | os.O_CREAT)
+ return fd, str(path)
+
+ def fake_popen(cmd, **kwargs):
+ spawned["cmd"] = cmd
+ return MagicMock()
+
+ monkeypatch.setattr("update.install.tempfile.mkstemp", fake_mkstemp)
+ monkeypatch.setattr("update.install.subprocess.Popen", fake_popen)
+ monkeypatch.setattr("update.install.platform.system", lambda: "Darwin")
+
+ script = write_and_spawn_installer(source=source, target=target, wait_for_pid=99)
+ content = script.read_text(encoding="utf-8")
+ assert "open \"$DST\"" in content or 'open "$DST"' in content
+ assert "failed to promote staged app" in content
+ assert spawned["cmd"][:3] == ["open", "-a", "Terminal"]
+
+
+def test_describe_manual_update_instructions_lists_steps():
+ with patch("update.flow.find_other_instances", return_value=[]), \
+ patch(
+ "update.flow._runtime_supports_inplace_update",
+ return_value=(True, ""),
+ ), \
+ patch("update.flow.sys.platform", "linux"), \
+ patch("update.flow.__bundle__", "/tmp/perfecto-mcp"), \
+ patch("update.flow.__executable__", "/tmp/perfecto-mcp"):
+ guidance = describe_manual_update_instructions()
+ assert guidance["supported"] is True
+ assert guidance["other_instances_running"] == []
+ assert any("quit" in step.lower() for step in guidance["steps"])
+
+
+def test_run_interactive_update_blocks_when_not_frozen():
+ with patch("update.flow.__uvx__", False), \
+ patch(
+ "update.flow._runtime_supports_inplace_update",
+ return_value=(False, "not frozen"),
+ ):
+ code = run_interactive_update(auto_confirm=True)
+ assert code == 1
+
+
+def test_run_interactive_update_spawns_and_returns_without_sys_exit():
+ release = MagicMock()
+ release.update_available = True
+ release.current_version = "1.0.0"
+ release.latest_version = "1.1.1"
+ release.html_url = "https://example.com"
+ release.body = ""
+ release.recommended_asset = {
+ "name": "perfecto-mcp-macos-arm64.zip",
+ "browser_download_url": "https://example.com/a.zip",
+ }
+ release.assets = [release.recommended_asset]
+
+ with patch("update.flow.__uvx__", False), \
+ patch("update.flow._runtime_supports_inplace_update", return_value=(True, "")), \
+ patch("update.flow.fetch_latest_release", return_value=release), \
+ patch("update.flow.find_other_instances", return_value=[]), \
+ patch("update.flow.stage_update_from_asset", return_value=Path("/tmp/staged")), \
+ patch("update.flow.write_and_spawn_installer", return_value=Path("/tmp/install.sh")) as spawn, \
+ patch("update.flow.install_target_path", return_value=Path("/tmp/app.app")):
+ code = run_interactive_update(auto_confirm=True)
+
+ assert code == 0
+ spawn.assert_called_once()
+
+
+def test_update_status_tool(perfecto_token):
+ from tools.tools_manager import ToolsManager
+
+ manager = ToolsManager(perfecto_token, MagicMock())
+ with patch("tools.tools_manager.find_other_instances", return_value=[]), \
+ patch("tools.tools_manager.describe_manual_update_instructions", return_value={
+ "supported": True,
+ "steps": ["step"],
+ "other_instances_running": [],
+ }):
+ result = asyncio.run(manager.update_status())
+
+ assert result.error is None
+ assert result.result[0]["manual_update"]["supported"] is True
+ assert any("quit" in message.lower() for message in result.info)
diff --git a/tools/tools_manager.py b/tools/tools_manager.py
index 6bb7ced..32e7d1f 100644
--- a/tools/tools_manager.py
+++ b/tools/tools_manager.py
@@ -2,13 +2,13 @@
import platform
import sys
import traceback
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
import httpx
from mcp.server.fastmcp import Context
-from packaging.version import InvalidVersion, Version
from pydantic import Field
+from config.http import timeout, user_agent
from config.perfecto import (
GITHUB,
GITHUB_API_LATEST_RELEASE,
@@ -16,37 +16,18 @@
TOOLS_PREFIX,
WEBSITE,
)
+from config.platform_names import normalize_arch, normalize_system
from config.token import PerfectoToken
from config.version import __bundle__, __executable__, __uvx__, __version__
from models.manager import Manager
from models.result import BaseResult
from telemetry import run_tool
-from tools.utils import timeout, user_agent
-
-
-def _normalize_system(system: str) -> str:
- system = system.lower()
- if system == "darwin":
- return "macos"
- if system == "windows":
- return "windows"
- return "linux"
-
-
-def _normalize_arch(machine: str) -> str:
- machine = machine.lower()
- if machine in {"x86_64", "amd64"}:
- return "amd64"
- if machine in {"aarch64", "arm64"} or machine.startswith("arm"):
- return "arm64"
- return machine
-
-
-def _parse_version(value: str) -> Optional[Version]:
- try:
- return Version(str(value).lstrip("vV"))
- except InvalidVersion:
- return None
+from update.flow import describe_manual_update_instructions
+from update.processes import find_other_instances
+from update.release import (
+ latest_release_from_payload,
+ match_recommended_asset,
+)
def _detect_runtime() -> Dict[str, Any]:
@@ -64,29 +45,20 @@ def _platform_info() -> Dict[str, str]:
"system": platform.system(),
"release": platform.release(),
"machine": platform.machine(),
- "normalized_system": _normalize_system(platform.system()),
- "normalized_arch": _normalize_arch(platform.machine()),
+ "normalized_system": normalize_system(platform.system()),
+ "normalized_arch": normalize_arch(platform.machine()),
}
-def _match_recommended_asset(assets: List[Dict[str, Any]], system: str, arch: str) -> Optional[Dict[str, Any]]:
- prefix = f"perfecto-mcp-{system}-{arch}"
- exact = [
- asset for asset in assets
- if str(asset.get("name", "")).startswith(prefix)
- ]
- if exact:
- # Prefer zip packages when multiple assets share the same platform prefix.
- zip_assets = [asset for asset in exact if str(asset.get("name", "")).endswith(".zip")]
- return zip_assets[0] if zip_assets else exact[0]
- return None
-
-
def _releases_url() -> str:
return f"{GITHUB}/releases"
-def _update_guidance(runtime: Dict[str, Any], update_available: bool, recommended_asset: Optional[Dict[str, Any]]) -> Dict[str, str]:
+def _update_guidance(
+ runtime: Dict[str, Any],
+ update_available: bool,
+ recommended_asset: Optional[Dict[str, Any]],
+) -> Dict[str, str]:
releases_url = _releases_url()
manual = (
f"Download the package for your platform from {releases_url}, "
@@ -108,10 +80,19 @@ def _update_guidance(runtime: Dict[str, Any], update_available: bool, recommende
f"Update the MCP client config git ref to the latest release tag "
f"(for example `git+{GITHUB}.git@v`) and restart the MCP client."
)
+ elif runtime.get("frozen"):
+ automatic = (
+ "In-place update is manual and must not run while this MCP server is still attached. "
+ "Ask the user to quit Perfecto MCP in every MCP client, confirm no other Perfecto MCP "
+ "processes are running (action `update_status`), then double-click the app / run the "
+ "binary without `--mcp` (or with `--update`) and follow the on-screen installer. "
+ "MCP clients usually reconnect after the process exits and the new binary is in place."
+ )
else:
automatic = (
- "Automatic in-place update is not available yet. "
- "Use the manual download path above, or ask the AI to guide a controlled manual replace."
+ "This runtime is not a frozen binary build. Update from source by checking out the "
+ f"latest release tag from {releases_url}, reinstalling dependencies, and restarting "
+ "the MCP client. The `--update` installer only applies to PyInstaller builds."
)
if not update_available:
@@ -220,6 +201,45 @@ async def version(self) -> BaseResult:
],
)
+ async def update_status(self) -> BaseResult:
+ runtime = _detect_runtime()
+ others = find_other_instances()
+ guidance = describe_manual_update_instructions(others=others)
+ info = [
+ "This MCP session must be quit before a frozen binary can be replaced.",
+ "Use this report to guide the user through a double-click / `--update` install.",
+ ]
+ if others:
+ info.append(
+ f"Found {len(others)} other Perfecto MCP process(es). "
+ "Ask the user to quit MCP clients / close those terminals before updating."
+ )
+ else:
+ info.append(
+ "No other Perfecto MCP binaries were detected besides this session "
+ "(IDE helpers that only mention the repo name are ignored)."
+ )
+
+ if runtime.get("docker"):
+ info.append("Docker runtimes should pull a new image rather than using the binary updater.")
+ elif runtime.get("uvx"):
+ info.append("uvx runtimes should bump the git ref in MCP config rather than using the binary updater.")
+ elif not runtime.get("frozen"):
+ info.append(
+ "This runtime is not frozen; the `--update` installer does not apply. "
+ "Use a source/git update instead."
+ )
+
+ return BaseResult(
+ result=[{
+ "current_version": __version__,
+ "platform": _platform_info(),
+ "runtime": runtime,
+ "manual_update": guidance,
+ }],
+ info=info,
+ )
+
async def check_updates(self) -> BaseResult:
headers = {
"User-Agent": user_agent,
@@ -231,55 +251,43 @@ async def check_updates(self) -> BaseResult:
try:
resp = await client.get(GITHUB_API_LATEST_RELEASE, headers=headers)
resp.raise_for_status()
- release = resp.json()
+ payload = resp.json()
except httpx.HTTPError as exc:
return _github_access_failure_result(exc)
except Exception as exc:
return _github_access_failure_result(exc)
- tag_name = str(release.get("tag_name") or "")
- latest_version = tag_name.lstrip("vV") or str(release.get("name") or "")
- current = _parse_version(__version__)
- latest = _parse_version(latest_version)
-
- if current is None or latest is None:
- return BaseResult(
- error=(
- f"Unable to compare versions. "
- f"current={__version__!r}, latest={latest_version!r}."
- )
- )
-
- update_available = latest > current
- assets = [
- {
- "name": asset.get("name"),
- "browser_download_url": asset.get("browser_download_url"),
- "size": asset.get("size"),
- "content_type": asset.get("content_type"),
- "updated_at": asset.get("updated_at"),
- }
- for asset in (release.get("assets") or [])
- ]
+ try:
+ release = latest_release_from_payload(payload, current_version=__version__)
+ except ValueError as exc:
+ return BaseResult(error=str(exc))
platform_data = _platform_info()
runtime = _detect_runtime()
- recommended_asset = _match_recommended_asset(
- assets,
+ # Re-match with this host's normalized platform (same helper as the interactive updater).
+ recommended_asset = match_recommended_asset(
+ release.assets,
platform_data["normalized_system"],
platform_data["normalized_arch"],
)
- guidance = _update_guidance(runtime, update_available, recommended_asset)
+ guidance = _update_guidance(runtime, release.update_available, recommended_asset)
info = []
- if update_available:
+ if release.update_available:
info.append(
- f"Update available: current {__version__} -> latest {latest_version}."
- )
- info.append(
- "Share the release notes with the user and ask whether they want a manual "
- "update now or guidance for an automatic update path."
+ f"Update available: current {__version__} -> latest {release.latest_version}."
)
+ if runtime.get("frozen"):
+ info.append(
+ "Share the release notes with the user. Guide a manual update via "
+ "double-click / `--update` (never overwrite files while MCP is live). "
+ "Use action `update_status` to list other running Perfecto MCP processes first."
+ )
+ else:
+ info.append(
+ "Share the release notes and the manual download guidance. "
+ "The in-place `--update` installer applies only to frozen binary builds."
+ )
if recommended_asset:
info.append(
f"Recommended download for this host: {recommended_asset.get('name')}."
@@ -294,18 +302,18 @@ async def check_updates(self) -> BaseResult:
return BaseResult(
result=[{
- "current_version": __version__,
- "latest_version": latest_version,
- "update_available": update_available,
+ "current_version": release.current_version,
+ "latest_version": release.latest_version,
+ "update_available": release.update_available,
"release": {
- "tag_name": tag_name,
- "name": release.get("name"),
- "html_url": release.get("html_url"),
- "published_at": release.get("published_at"),
- "body": release.get("body") or "",
+ "tag_name": release.tag_name,
+ "name": payload.get("name"),
+ "html_url": release.html_url,
+ "published_at": payload.get("published_at"),
+ "body": release.body,
},
"recommended_asset": recommended_asset,
- "assets": assets,
+ "assets": release.assets,
"platform": platform_data,
"runtime": runtime,
"update_guidance": guidance,
@@ -324,10 +332,14 @@ def register(mcp, token: Optional[PerfectoToken]):
user-agent and `--version` / console display.
- check_updates: Query the GitHub repository for the latest release, compare it with the
current version, and return release notes plus download links when an update is available.
+- update_status: List other running Perfecto MCP processes and return step-by-step manual
+ update instructions (double-click / `--update`). Does not download or replace files.
Hints:
- Prefer `version` first when the user asks what build they are running.
- Prefer `check_updates` when the user asks whether a newer MCP release exists.
-- When an update is available, present release notes and ask before replacing the executable.
+- When an update is available for a frozen binary, present release notes, call `update_status`,
+ and guide the user to quit MCP clients then double-click the app / run `--update`.
+ Do not attempt to overwrite the running executable from inside the MCP session.
- If GitHub is unreachable (restricted network, firewall, timeout), explain that the update
check is unavailable, still share the current version, and point to the releases page for a
manual check — do not treat it as a hard failure.
@@ -349,6 +361,8 @@ async def _dispatch():
return await tools_manager.version()
case "check_updates":
return await tools_manager.check_updates()
+ case "update_status":
+ return await tools_manager.update_status()
case _:
return BaseResult(
error=f"Action {action} not found in tools manager tool"
diff --git a/tools/utils.py b/tools/utils.py
index cc2e7e0..e8f619e 100644
--- a/tools/utils.py
+++ b/tools/utils.py
@@ -17,22 +17,10 @@
from config.security import validate_http_request_endpoint
from config.token import PerfectoToken
-from config.version import __version__
+from config.http import timeout, user_agent
from models.result import BaseResult
so = platform.system() # "Windows", "Linux", "Darwin"
-version = platform.version() # kernel / build version
-release = platform.release() # ex. "10", "5.15.0-76-generic"
-machine = platform.machine() # ex. "x86_64", "AMD64", "arm64"
-
-ua_part = f"{so} {release}; {machine}"
-user_agent = f"perfecto-mcp/{__version__} ({ua_part})"
-timeout = httpx.Timeout(
- connect=15.0,
- read=60.0,
- write=15.0,
- pool=60.0
-)
project_root = Path(__file__).resolve().parent.parent
# Match Windows absolute paths (backslash or forward slash; latter may appear on POSIX).
# Negative lookbehind ensures we don't match URL protocols like https:// (where the
diff --git a/update/__init__.py b/update/__init__.py
new file mode 100644
index 0000000..b04f077
--- /dev/null
+++ b/update/__init__.py
@@ -0,0 +1,10 @@
+"""Manual in-place updater for frozen Perfecto MCP binaries (tufup-inspired)."""
+
+from update.flow import run_interactive_update
+from update.processes import RunningProcess, find_other_instances
+
+__all__ = [
+ "RunningProcess",
+ "find_other_instances",
+ "run_interactive_update",
+]
diff --git a/update/flow.py b/update/flow.py
new file mode 100644
index 0000000..f74352c
--- /dev/null
+++ b/update/flow.py
@@ -0,0 +1,191 @@
+"""Interactive (double-click / --update) manual update flow."""
+
+from __future__ import annotations
+
+import platform
+import sys
+from typing import Callable, List, Optional
+
+import httpx
+
+from config.perfecto import GITHUB
+from config.version import __bundle__, __executable__, __uvx__, __version__
+from update.install import install_target_path, update_log_path, write_and_spawn_installer
+from update.processes import RunningProcess, find_other_instances, format_process_list
+from update.release import fetch_latest_release, stage_update_from_asset
+
+
+PromptFn = Callable[[str], str]
+
+
+def _default_prompt(message: str) -> str:
+ return input(message)
+
+
+def _runtime_supports_inplace_update() -> tuple[bool, str]:
+ if __uvx__:
+ return False, (
+ "This process was started via uvx. Update by changing the git ref in your "
+ f"MCP client config (see {GITHUB}/releases), then restart the client."
+ )
+ if not getattr(sys, "frozen", False):
+ return False, (
+ "In-place binary update only applies to frozen PyInstaller builds. "
+ f"For source/dev installs, pull the latest tag from {GITHUB}."
+ )
+ return True, ""
+
+
+def _wait_until_no_other_instances(prompt: PromptFn) -> bool:
+ while True:
+ others = find_other_instances()
+ if not others:
+ print(" No other Perfecto MCP processes are running. Continuing.")
+ return True
+
+ print(" Perfecto MCP is still running elsewhere. Close it before updating:\n")
+ print(format_process_list(others))
+ print()
+ print(" Typical causes:")
+ print(" - An MCP client (Cursor, VS Code, Claude Desktop, …) still has the server attached")
+ print(" - Another Terminal window running perfecto-mcp")
+ print()
+ print(" Close those sessions, then press Enter to re-check (or type 'q' to cancel).")
+ answer = prompt("> ").strip().lower()
+ if answer in {"q", "quit", "exit", "n", "no"}:
+ return False
+
+
+def run_interactive_update(*, prompt: Optional[PromptFn] = None, auto_confirm: bool = False) -> int:
+ """
+ Run the manual update wizard.
+
+ Returns a process exit code. On success after spawning the installer, returns 0;
+ the caller must exit the process so files can be replaced.
+ """
+ prompt = prompt or _default_prompt
+
+ print(f" Perfecto MCP updater (current version {__version__})")
+ print(f" Install target: {install_target_path()}")
+ print()
+
+ supported, reason = _runtime_supports_inplace_update()
+ if not supported:
+ print(f" {reason}")
+ return 1
+
+ try:
+ release = fetch_latest_release()
+ except httpx.HTTPError as exc:
+ print(f" Could not reach GitHub to check for updates: {exc}")
+ print(f" Open {GITHUB}/releases when you have network access.")
+ return 1
+ except ValueError as exc:
+ print(f" {exc}")
+ return 1
+
+ if not release.update_available:
+ print(f" Already up to date (latest is {release.latest_version}).")
+ return 0
+
+ print(f" Update available: {release.current_version} -> {release.latest_version}")
+ print(f" Release: {release.html_url}")
+ if release.body.strip():
+ print()
+ print(" Release notes (truncated):")
+ for line in release.body.strip().splitlines()[:12]:
+ print(f" {line}")
+ print()
+
+ asset = release.recommended_asset
+ if not asset:
+ print(" No download asset matched this platform. Open the releases page and install manually:")
+ print(f" {release.html_url}")
+ return 1
+
+ print(f" Recommended package: {asset.get('name')}")
+ print()
+ print(" Before updating:")
+ print(" 1. Quit / disable Perfecto MCP in every MCP client so no server process remains.")
+ print(" 2. Keep this window open and follow the prompts.")
+ print(" 3. After install, the app relaunches; reopen the MCP client if needed.")
+ print()
+
+ if not auto_confirm:
+ answer = prompt(" Download and install this update now? [y/N]: ").strip().lower()
+ if answer not in {"y", "yes"}:
+ print(" Update cancelled.")
+ return 0
+
+ if not _wait_until_no_other_instances(prompt):
+ print(" Update cancelled while Perfecto MCP was still running.")
+ return 1
+
+ print(" Downloading update…")
+ try:
+ staged = stage_update_from_asset(asset, all_assets=release.assets)
+ except (httpx.HTTPError, OSError, ValueError, FileNotFoundError) as exc:
+ print(f" Download/extract failed: {exc}")
+ return 1
+
+ # Re-check immediately before spawn — a client may have reconnected during download.
+ if find_other_instances():
+ print(" Perfecto MCP started again during download. Close it, then re-run the updater.")
+ return 1
+
+ print(f" Staged payload: {staged}")
+ if platform.system() == "Darwin":
+ print(" Launching install helper in a new Terminal window.")
+ else:
+ print(" Launching background install helper (progress is written to the log file).")
+ print(" It waits for this process (and any other Perfecto MCP processes) to exit,")
+ print(" then replaces the app and relaunches it.")
+ script = write_and_spawn_installer(source=staged)
+ print(f" Install script: {script}")
+ print(f" Install log: {update_log_path()}")
+ print(" Exiting so the install helper can replace files…")
+ return 0
+
+
+def describe_manual_update_instructions(
+ *,
+ others: Optional[List[RunningProcess]] = None,
+) -> dict:
+ """Structured guidance for MCP tools / agents."""
+ supported, reason = _runtime_supports_inplace_update()
+ if others is None:
+ others = find_other_instances()
+ target = str(install_target_path())
+ if sys.platform == "darwin" and str(__bundle__).endswith(".app"):
+ launch_hint = (
+ f"Double-click `{__bundle__}` (or run it from Finder). "
+ "A Terminal window opens with on-screen update instructions."
+ )
+ else:
+ launch_hint = (
+ f"Quit the MCP client session, then run `{__executable__}` without `--mcp` "
+ "(or with `--update`) and follow the on-screen instructions."
+ )
+
+ return {
+ "supported": supported,
+ "unsupported_reason": reason or None,
+ "install_target": target,
+ "this_session_must_quit": True,
+ "this_session_note": (
+ "This MCP tool call is running inside a live Perfecto MCP process. "
+ "That process (and any MCP client attached to it) must be stopped before "
+ "files can be replaced. Reconnection after quit is expected and normal."
+ ),
+ "other_instances_running": [
+ {"pid": p.pid, "name": p.name, "command": p.command} for p in others
+ ],
+ "steps": [
+ "Tell the user an update requires quitting Perfecto MCP in every MCP client first "
+ "(this live session cannot overwrite its own executable).",
+ "Call out any other Perfecto MCP processes from other_instances_running and ask the user to close them.",
+ launch_hint,
+ "In the updater window, confirm no processes remain, download, and wait until install succeeds.",
+ "Re-enable / reopen the MCP client so it reconnects to the new binary.",
+ ],
+ }
diff --git a/update/install.py b/update/install.py
new file mode 100644
index 0000000..ed0b262
--- /dev/null
+++ b/update/install.py
@@ -0,0 +1,328 @@
+"""
+Platform install helpers inspired by tufup.
+
+Frozen executables cannot safely overwrite themselves while running. The pattern is:
+1. Stage the new files somewhere else.
+2. Spawn a short-lived script/process that waits for this PID to exit.
+3. Replace the install target, then relaunch (MCP clients typically reconnect afterward).
+"""
+
+from __future__ import annotations
+
+import os
+import platform
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import textwrap
+from pathlib import Path
+from typing import Optional, Union
+
+from config.version import __bundle__, __executable__
+from update.release import BINARY_NAME
+
+PathLike = Union[str, Path]
+UPDATE_LOG_NAME = f"{BINARY_NAME}-update.log"
+
+
+def install_target_path() -> Path:
+ """Directory or file that should be replaced for this runtime."""
+ if sys.platform == "darwin" and str(__bundle__).endswith(".app"):
+ return Path(__bundle__).resolve()
+ return Path(__executable__).resolve()
+
+
+def update_log_path(target: Optional[Path] = None) -> Path:
+ path = target if target is not None else install_target_path()
+ return path.parent / UPDATE_LOG_NAME
+
+
+def _is_app_bundle(path: Path) -> bool:
+ return path.suffix == ".app" or str(path).endswith(".app")
+
+
+def _unix_wait_and_log_header(*, pid: int, log_file: Path) -> str:
+ log_q = str(log_file)
+ return textwrap.dedent(
+ f"""\
+ #!/bin/bash
+ set -euo pipefail
+ LOG_FILE="{log_q}"
+ mkdir -p "$(dirname "$LOG_FILE")"
+ exec > >(tee -a "$LOG_FILE") 2>&1
+ echo "Waiting for Perfecto MCP (PID {pid}) to exit..."
+ while kill -0 {pid} 2>/dev/null; do
+ sleep 1
+ done
+ echo "Ensuring no other Perfecto MCP processes remain..."
+ SELF_PID=$$
+ while true; do
+ FOUND=0
+ while read -r OPID OCMD; do
+ [ -z "${{OPID:-}}" ] && continue
+ [ "$OPID" = "$SELF_PID" ] && continue
+ case "$OCMD" in
+ *{BINARY_NAME}-install-*) continue ;;
+ *tee\\ -a*\"$LOG_FILE\"*|*"$LOG_FILE"*) continue ;;
+ esac
+ case "$OCMD" in
+ */Contents/MacOS/{BINARY_NAME}*|*/{BINARY_NAME}\\ --*|*/{BINARY_NAME}|*/{BINARY_NAME}.exe*)
+ FOUND=1
+ break
+ ;;
+ esac
+ done < <(ps -ax -o pid=,command= 2>/dev/null || true)
+ [ "$FOUND" -eq 0 ] && break
+ echo "Waiting for remaining Perfecto MCP processes to exit..."
+ sleep 1
+ done
+ """
+ )
+
+
+def _unix_script_footer() -> str:
+ return textwrap.dedent(
+ """\
+ echo "Done. You can close this window."
+ rm -f -- "$0"
+ """
+ )
+
+
+def _macos_app_bundle_script(*, pid: int, source: Path, target: Path, log_file: Path) -> str:
+ source_q = str(source)
+ target_q = str(target)
+ header = _unix_wait_and_log_header(pid=pid, log_file=log_file)
+ body = textwrap.dedent(
+ f"""\
+ SRC="{source_q}"
+ DST="{target_q}"
+ BACKUP="${{DST}}.bak.$$"
+ STAGE="${{DST}}.new.$$"
+ echo "Installing update into $DST ..."
+ if [ ! -d "$SRC" ] || [ ! -f "$SRC/Contents/MacOS/{BINARY_NAME}" ]; then
+ echo "ERROR: staged update is not a usable .app (missing Contents/MacOS/{BINARY_NAME})."
+ echo "SRC=$SRC"
+ exit 1
+ fi
+ rm -rf "$STAGE" "$BACKUP"
+ if command -v ditto >/dev/null 2>&1; then
+ ditto "$SRC" "$STAGE"
+ else
+ cp -R "$SRC" "$STAGE"
+ fi
+ chmod 755 "$STAGE/Contents/MacOS/{BINARY_NAME}" 2>/dev/null || true
+ chmod 755 "$STAGE/Contents/MacOS/launcher.sh" 2>/dev/null || true
+ mv "$DST" "$BACKUP"
+ if ! mv "$STAGE" "$DST"; then
+ echo "ERROR: failed to promote staged app; restoring backup."
+ mv "$BACKUP" "$DST" || true
+ rm -rf "$STAGE" || true
+ exit 1
+ fi
+ rm -rf "$BACKUP"
+ echo "Update installed successfully."
+ echo "Relaunching Perfecto MCP..."
+ open "$DST" || true
+ """
+ )
+ return header + body + _unix_script_footer()
+
+
+def _macos_inner_binary_script(
+ *, pid: int, source: Path, target: Path, log_file: Path
+) -> str:
+ source_q = str(source)
+ target_q = str(target)
+ header = _unix_wait_and_log_header(pid=pid, log_file=log_file)
+ body = textwrap.dedent(
+ f"""\
+ SRC="{source_q}"
+ DST="{target_q}"
+ INNER="$DST/Contents/MacOS/{BINARY_NAME}"
+ BACKUP="${{INNER}}.bak.$$"
+ echo "Installing binary update into $INNER ..."
+ if [ ! -f "$SRC" ]; then
+ echo "ERROR: staged binary not found: $SRC"
+ exit 1
+ fi
+ if [ ! -d "$DST/Contents/MacOS" ]; then
+ echo "ERROR: install target is not a usable .app: $DST"
+ exit 1
+ fi
+ cp "$INNER" "$BACKUP"
+ if ! cp "$SRC" "$INNER"; then
+ echo "ERROR: failed to replace inner binary; restoring backup."
+ cp "$BACKUP" "$INNER" || true
+ exit 1
+ fi
+ chmod 755 "$INNER"
+ rm -f "$BACKUP"
+ echo "Update installed successfully."
+ echo "Relaunching Perfecto MCP..."
+ open "$DST" || true
+ """
+ )
+ return header + body + _unix_script_footer()
+
+
+def _linux_or_macos_file_script(
+ *, pid: int, source: Path, target: Path, log_file: Path
+) -> str:
+ source_q = str(source)
+ target_q = str(target)
+ header = _unix_wait_and_log_header(pid=pid, log_file=log_file)
+ body = textwrap.dedent(
+ f"""\
+ SRC="{source_q}"
+ DST="{target_q}"
+ BACKUP="${{DST}}.bak.$$"
+ echo "Installing update to $DST ..."
+ if [ ! -f "$SRC" ]; then
+ echo "ERROR: staged update binary not found: $SRC"
+ exit 1
+ fi
+ cp "$DST" "$BACKUP"
+ if ! cp "$SRC" "$DST"; then
+ echo "ERROR: failed to replace binary; restoring backup."
+ cp "$BACKUP" "$DST" || true
+ exit 1
+ fi
+ chmod 755 "$DST"
+ rm -f "$BACKUP"
+ echo "Update installed successfully."
+ if [ "$(uname -s)" = "Darwin" ]; then
+ open "$DST" || true
+ else
+ nohup "$DST" >/dev/null 2>&1 &
+ fi
+ """
+ )
+ return header + body + _unix_script_footer()
+
+
+def _windows_script(*, pid: int, source: Path, target: Path, log_file: Path) -> str:
+ source_q = str(source)
+ target_q = str(target)
+ log_q = str(log_file)
+ return textwrap.dedent(
+ f"""\
+ @echo off
+ setlocal
+ set "LOG_FILE={log_q}"
+ echo Waiting for Perfecto MCP (PID {pid}) to exit...>>"%LOG_FILE%"
+ :waitloop
+ tasklist /FI "PID eq {pid}" 2>NUL | find "{pid}" >NUL
+ if not errorlevel 1 (
+ timeout /t 1 /nobreak >NUL
+ goto waitloop
+ )
+ echo Ensuring no other Perfecto MCP processes remain...>>"%LOG_FILE%"
+ :waitothers
+ tasklist /FI "IMAGENAME eq {BINARY_NAME}*" 2>NUL | find /I "{BINARY_NAME}" >NUL
+ if not errorlevel 1 (
+ timeout /t 1 /nobreak >NUL
+ goto waitothers
+ )
+ set "SRC={source_q}"
+ set "DST={target_q}"
+ set "BACKUP=%DST%.bak.%RANDOM%"
+ echo Installing update to %DST% ...>>"%LOG_FILE%"
+ copy /Y "%DST%" "%BACKUP%" >NUL
+ copy /Y "%SRC%" "%DST%" >NUL
+ if errorlevel 1 (
+ echo Update failed. Restoring backup...>>"%LOG_FILE%"
+ copy /Y "%BACKUP%" "%DST%" >NUL
+ exit /b 1
+ )
+ del /F /Q "%BACKUP%" >NUL 2>&1
+ echo Update installed successfully.>>"%LOG_FILE%"
+ start "" "%DST%"
+ echo Done.>>"%LOG_FILE%"
+ (goto) 2>nul & del "%~f0"
+ """
+ )
+
+
+def write_and_spawn_installer(
+ *,
+ source: PathLike,
+ target: Optional[PathLike] = None,
+ wait_for_pid: Optional[int] = None,
+) -> Path:
+ """
+ Write a platform-specific install script and start it in a detached process.
+
+ Does not exit the current process; caller should exit after spawning.
+ """
+ source_path = Path(source).resolve()
+ target_path = Path(target).resolve() if target is not None else install_target_path()
+ pid = wait_for_pid if wait_for_pid is not None else os.getpid()
+ system = platform.system()
+ log_file = update_log_path(target_path)
+
+ if system == "Windows":
+ content = _windows_script(
+ pid=pid, source=source_path, target=target_path, log_file=log_file
+ )
+ suffix = ".bat"
+ else:
+ is_app = _is_app_bundle(target_path)
+ if is_app and system == "Darwin" and source_path.is_file():
+ content = _macos_inner_binary_script(
+ pid=pid, source=source_path, target=target_path, log_file=log_file
+ )
+ elif is_app and system == "Darwin":
+ content = _macos_app_bundle_script(
+ pid=pid, source=source_path, target=target_path, log_file=log_file
+ )
+ else:
+ content = _linux_or_macos_file_script(
+ pid=pid, source=source_path, target=target_path, log_file=log_file
+ )
+ suffix = ".sh"
+
+ fd, script_name = tempfile.mkstemp(prefix=f"{BINARY_NAME}-install-", suffix=suffix)
+ os.close(fd)
+ script_path = Path(script_name)
+ script_path.write_text(content, encoding="utf-8")
+
+ if system == "Windows":
+ creationflags = getattr(subprocess, "CREATE_NEW_CONSOLE", 0)
+ subprocess.Popen(["cmd.exe", "/c", str(script_path)], creationflags=creationflags)
+ else:
+ script_path.chmod(script_path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+ if system == "Darwin":
+ subprocess.Popen(["open", "-a", "Terminal", str(script_path)])
+ else:
+ subprocess.Popen(
+ ["nohup", "/bin/bash", str(script_path)],
+ start_new_session=True,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+
+ return script_path
+
+
+def copy_into_place_now(source: PathLike, target: Optional[PathLike] = None) -> Path:
+ """Synchronous replace used by tests / when the target is not locked."""
+ source_path = Path(source).resolve()
+ target_path = Path(target).resolve() if target is not None else install_target_path()
+ if _is_app_bundle(target_path):
+ if source_path.is_dir() and source_path.name.endswith(".app"):
+ if target_path.exists():
+ shutil.rmtree(target_path)
+ shutil.copytree(source_path, target_path)
+ else:
+ inner = target_path / "Contents" / "MacOS" / BINARY_NAME
+ inner.parent.mkdir(parents=True, exist_ok=True)
+ shutil.copy2(source_path, inner)
+ os.chmod(inner, 0o755)
+ else:
+ shutil.copy2(source_path, target_path)
+ if platform.system() != "Windows":
+ os.chmod(target_path, 0o755)
+ return target_path
diff --git a/update/processes.py b/update/processes.py
new file mode 100644
index 0000000..4e780ef
--- /dev/null
+++ b/update/processes.py
@@ -0,0 +1,153 @@
+"""Detect other running Perfecto MCP instances before applying an update."""
+
+from __future__ import annotations
+
+import os
+import platform
+import re
+import subprocess
+from dataclasses import dataclass
+from typing import List, Optional
+
+from update.release import BINARY_NAME
+
+
+@dataclass(frozen=True)
+class RunningProcess:
+ pid: int
+ name: str
+ command: str
+
+
+_APP_LAUNCHER_RE = re.compile(
+ rf"{re.escape(BINARY_NAME)}[^/\s]*\.app/Contents/MacOS/",
+ re.IGNORECASE,
+)
+# Absolute/relative path to the binary (not a bare workspace-name token).
+_BINARY_PATH_RE = re.compile(
+ rf"(?:^|[\s\"'=])((?:[A-Za-z]:)?(?:[/\\][^\s\"']*)?[/\\]{re.escape(BINARY_NAME)}"
+ rf"(?:-[A-Za-z0-9._]+)?(?:\.exe)?)(?=[\s\"']|$)",
+ re.IGNORECASE,
+)
+
+
+def _executable_basename(token: str) -> str:
+ token = token.strip().strip('"').strip("'")
+ return os.path.basename(token).lower()
+
+
+def _matches_perfecto_mcp(name: str, command: str) -> bool:
+ """
+ Match real Perfecto MCP server processes only.
+
+ Avoid false positives from IDE helpers, shells, or paths that merely mention
+ the repository name (e.g. Cursor extension-host titles containing 'perfecto-mcp').
+ """
+ name_base = _executable_basename(name.split()[0] if name else "")
+ if name_base.startswith(BINARY_NAME):
+ return True
+
+ command = command.strip()
+ if not command:
+ return False
+
+ first = _executable_basename(command.split()[0])
+ if first.startswith(BINARY_NAME):
+ return True
+
+ normalized = command.replace("\\", "/")
+ if _APP_LAUNCHER_RE.search(normalized):
+ return True
+
+ if _BINARY_PATH_RE.search(normalized):
+ # `cd /.../perfecto-mcp && pytest` mentions the repo dir, not the binary.
+ if re.search(r"\bcd\s+", command) and "--mcp" not in command and "--update" not in command:
+ return False
+ return True
+
+ return False
+
+
+def _parse_ps_line(line: str) -> Optional[RunningProcess]:
+ line = line.strip()
+ if not line:
+ return None
+ parts = line.split(None, 1)
+ if len(parts) < 2 or not parts[0].isdigit():
+ return None
+ pid = int(parts[0])
+ command = parts[1]
+ name = _executable_basename(command.split()[0]) if command else ""
+ if not _matches_perfecto_mcp(name, command):
+ return None
+ return RunningProcess(pid=pid, name=name, command=command)
+
+
+def _list_unix_processes() -> List[RunningProcess]:
+ try:
+ completed = subprocess.run(
+ ["ps", "-ax", "-o", "pid=,command="],
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+ except (OSError, subprocess.CalledProcessError):
+ return []
+
+ found: List[RunningProcess] = []
+ for line in completed.stdout.splitlines():
+ proc = _parse_ps_line(line)
+ if proc is not None:
+ found.append(proc)
+ return found
+
+
+def _list_windows_processes() -> List[RunningProcess]:
+ try:
+ completed = subprocess.run(
+ ["tasklist", "/FO", "CSV", "/NH"],
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+ except (OSError, subprocess.CalledProcessError):
+ return []
+
+ found: List[RunningProcess] = []
+ for line in completed.stdout.splitlines():
+ fields = _split_csv_fields(line)
+ if len(fields) < 2:
+ continue
+ name = fields[0]
+ pid_text = fields[1]
+ if not pid_text.isdigit():
+ continue
+ if not _matches_perfecto_mcp(name, name):
+ continue
+ found.append(RunningProcess(pid=int(pid_text), name=name, command=name))
+ return found
+
+
+def _split_csv_fields(line: str) -> List[str]:
+ return re.findall(r'"([^"]*)"', line)
+
+
+def list_perfecto_mcp_processes() -> List[RunningProcess]:
+ if platform.system() == "Windows":
+ return _list_windows_processes()
+ return _list_unix_processes()
+
+
+def find_other_instances(exclude_pid: Optional[int] = None) -> List[RunningProcess]:
+ """Return Perfecto MCP processes other than this one (and its direct parent)."""
+ if exclude_pid is None:
+ exclude_pid = os.getpid()
+ parent_pid = os.getppid()
+ excluded = {exclude_pid, parent_pid}
+ return [proc for proc in list_perfecto_mcp_processes() if proc.pid not in excluded]
+
+
+def format_process_list(processes: List[RunningProcess]) -> str:
+ if not processes:
+ return "(none)"
+ return "\n".join(f" PID {proc.pid}: {proc.command}" for proc in processes)
diff --git a/update/release.py b/update/release.py
new file mode 100644
index 0000000..6717255
--- /dev/null
+++ b/update/release.py
@@ -0,0 +1,247 @@
+"""GitHub release lookup and asset download for the manual updater."""
+
+from __future__ import annotations
+
+import hashlib
+import platform
+import re
+import tempfile
+import zipfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import httpx
+from packaging.version import InvalidVersion, Version
+
+from config.http import timeout, user_agent
+from config.perfecto import GITHUB, GITHUB_API_LATEST_RELEASE
+from config.platform_names import normalize_arch, normalize_system
+from config.version import __version__
+
+BINARY_NAME = "perfecto-mcp"
+APPLEDOUBLE_DIR = "__MACOSX"
+
+
+def parse_version(value: str) -> Optional[Version]:
+ try:
+ return Version(str(value).lstrip("vV"))
+ except InvalidVersion:
+ return None
+
+
+def match_recommended_asset(
+ assets: List[Dict[str, Any]], system: str, arch: str
+) -> Optional[Dict[str, Any]]:
+ prefix = f"{BINARY_NAME}-{system}-{arch}"
+ exact = [
+ asset for asset in assets
+ if str(asset.get("name", "")).startswith(prefix)
+ and not str(asset.get("name", "")).endswith(".sha256")
+ ]
+ if not exact:
+ return None
+ zip_assets = [asset for asset in exact if str(asset.get("name", "")).endswith(".zip")]
+ if zip_assets:
+ return zip_assets[0]
+ app_assets = [asset for asset in exact if ".app" in str(asset.get("name", "")).lower()]
+ if app_assets:
+ return app_assets[0]
+ return exact[0]
+
+
+def find_checksum_asset(
+ assets: List[Dict[str, Any]], asset: Dict[str, Any]
+) -> Optional[Dict[str, Any]]:
+ name = str(asset.get("name") or "")
+ if not name:
+ return None
+ wanted = {f"{name}.sha256", name.replace(".zip", "") + ".sha256"}
+ for candidate in assets:
+ candidate_name = str(candidate.get("name") or "")
+ if candidate_name in wanted or candidate_name == f"{name}.sha256":
+ return candidate
+ return None
+
+
+@dataclass
+class LatestRelease:
+ tag_name: str
+ latest_version: str
+ html_url: str
+ body: str
+ assets: List[Dict[str, Any]]
+ recommended_asset: Optional[Dict[str, Any]]
+ update_available: bool
+ current_version: str
+
+
+def fetch_latest_release(client: Optional[httpx.Client] = None) -> LatestRelease:
+ headers = {
+ "User-Agent": user_agent,
+ "Accept": "application/vnd.github+json",
+ "X-GitHub-Api-Version": "2022-11-28",
+ }
+ owns_client = client is None
+ if owns_client:
+ client = httpx.Client(timeout=timeout)
+
+ try:
+ resp = client.get(GITHUB_API_LATEST_RELEASE, headers=headers)
+ resp.raise_for_status()
+ release = resp.json()
+ finally:
+ if owns_client:
+ client.close()
+
+ return latest_release_from_payload(release, current_version=__version__)
+
+
+def latest_release_from_payload(
+ release: Dict[str, Any], *, current_version: str
+) -> LatestRelease:
+ tag_name = str(release.get("tag_name") or "")
+ latest_version = tag_name.lstrip("vV") or str(release.get("name") or "")
+ current = parse_version(current_version)
+ latest = parse_version(latest_version)
+ if current is None or latest is None:
+ raise ValueError(
+ f"Unable to compare versions. current={current_version!r}, latest={latest_version!r}."
+ )
+
+ assets = [
+ {
+ "name": asset.get("name"),
+ "browser_download_url": asset.get("browser_download_url"),
+ "size": asset.get("size"),
+ "content_type": asset.get("content_type"),
+ "updated_at": asset.get("updated_at"),
+ }
+ for asset in (release.get("assets") or [])
+ ]
+ recommended = match_recommended_asset(
+ assets,
+ normalize_system(platform.system()),
+ normalize_arch(platform.machine()),
+ )
+ return LatestRelease(
+ tag_name=tag_name,
+ latest_version=latest_version,
+ html_url=str(release.get("html_url") or f"{GITHUB}/releases"),
+ body=str(release.get("body") or ""),
+ assets=assets,
+ recommended_asset=recommended,
+ update_available=latest > current,
+ current_version=current_version,
+ )
+
+
+def download_asset(asset: Dict[str, Any], dest_dir: Path) -> Path:
+ url = asset.get("browser_download_url")
+ name = str(asset.get("name") or "download.bin")
+ if not url:
+ raise ValueError("Recommended release asset has no download URL.")
+
+ dest_dir.mkdir(parents=True, exist_ok=True)
+ dest_path = dest_dir / name
+ headers = {"User-Agent": user_agent, "Accept": "application/octet-stream"}
+
+ with httpx.Client(timeout=httpx.Timeout(120.0, connect=30.0), follow_redirects=True) as client:
+ with client.stream("GET", url, headers=headers) as resp:
+ resp.raise_for_status()
+ with open(dest_path, "wb") as out:
+ for chunk in resp.iter_bytes():
+ out.write(chunk)
+ return dest_path
+
+
+def verify_sha256(file_path: Path, checksum_text: str) -> None:
+ """Raise ValueError if file contents do not match a sha256sum-style checksum file."""
+ expected = None
+ for line in checksum_text.splitlines():
+ line = line.strip()
+ if not line or line.startswith("#"):
+ continue
+ match = re.match(r"^([A-Fa-f0-9]{64})(?:\s+\*?(\S+))?$", line)
+ if match:
+ expected = match.group(1).lower()
+ break
+ if expected is None:
+ raise ValueError("Checksum file did not contain a usable SHA-256 digest.")
+
+ digest = hashlib.sha256()
+ with open(file_path, "rb") as handle:
+ for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+ digest.update(chunk)
+ actual = digest.hexdigest()
+ if actual != expected:
+ raise ValueError(
+ f"SHA-256 mismatch for {file_path.name}: expected {expected}, got {actual}."
+ )
+
+
+def extract_update_payload(archive_or_file: Path, extract_dir: Path) -> Path:
+ """
+ Return the path to install: either an extracted .app, extracted binary, or the file itself.
+ """
+ extract_dir.mkdir(parents=True, exist_ok=True)
+
+ if archive_or_file.suffix.lower() == ".zip":
+ with zipfile.ZipFile(archive_or_file, "r") as zf:
+ zf.extractall(extract_dir)
+ return _select_extracted_payload(extract_dir)
+
+ return archive_or_file
+
+
+def _is_appledouble_junk(path: Path) -> bool:
+ return APPLEDOUBLE_DIR in path.parts or path.name.startswith("._")
+
+
+def _is_usable_app_bundle(path: Path) -> bool:
+ if not path.is_dir() or not path.name.endswith(".app") or _is_appledouble_junk(path):
+ return False
+ binary = path / "Contents" / "MacOS" / BINARY_NAME
+ return binary.is_file() and binary.stat().st_size > 0
+
+
+def _select_extracted_payload(extract_dir: Path) -> Path:
+ apps = [
+ path for path in extract_dir.rglob("*.app")
+ if _is_usable_app_bundle(path)
+ ]
+ if apps:
+ apps.sort(key=lambda p: (len(p.parts), str(p)))
+ return apps[0]
+
+ candidates = [
+ path for path in extract_dir.rglob("*")
+ if path.is_file()
+ and not _is_appledouble_junk(path)
+ and BINARY_NAME in path.name.lower()
+ and not path.name.endswith(".sha256")
+ and not path.name.startswith("._")
+ ]
+ if not candidates:
+ raise FileNotFoundError(
+ "Zip archive did not contain a usable Perfecto MCP binary or .app "
+ f"(ignored AppleDouble / {APPLEDOUBLE_DIR} entries)."
+ )
+ candidates.sort(key=lambda p: (0 if p.suffix.lower() in {".exe", ""} else 1, len(str(p))))
+ return candidates[0]
+
+
+def stage_update_from_asset(
+ asset: Dict[str, Any],
+ *,
+ all_assets: Optional[List[Dict[str, Any]]] = None,
+) -> Path:
+ """Download (and unzip if needed) into a temp directory; return install source path."""
+ work = Path(tempfile.mkdtemp(prefix=f"{BINARY_NAME}-update-"))
+ downloaded = download_asset(asset, work / "download")
+ if all_assets:
+ checksum_asset = find_checksum_asset(all_assets, asset)
+ if checksum_asset and checksum_asset.get("browser_download_url"):
+ checksum_path = download_asset(checksum_asset, work / "checksum")
+ verify_sha256(downloaded, checksum_path.read_text(encoding="utf-8"))
+ return extract_update_payload(downloaded, work / "extracted")