#!/usr/bin/env python3 """Build real PyPI MCP environments without editing any installed server source.""" import argparse import datetime import hashlib import io import json import os from pathlib import Path import platform import shutil import subprocess import sys import tempfile import urllib.request import zipfile from probe import probe, sanitize SERVER = "mcp-server-git==2026.7.10" class Experiment: def __init__(self, root, output): self.root = root.resolve() self.output = output.resolve() allowed_prefixes = ("/private/tmp/asn-real-mcp-", "/private/tmp/asn-journey-2026-10-13") if not str(self.root).startswith(allowed_prefixes): raise ValueError("Run root must use an authorized asn-real-mcp or dated journey temporary prefix") if self.output.exists() and ( not self.output.is_dir() or any(self.output.iterdir()) ): raise FileExistsError("Refusing to overwrite evidence: choose a new empty output directory") self.root.mkdir(parents=True, exist_ok=True) self.output.mkdir(parents=True, exist_ok=True) self.commands = [] self.env = { "PATH": os.environ.get("PATH", os.defpath), "LANG": "C.UTF-8", "PYTHONNOUSERSITE": "1", "PYTHONDONTWRITEBYTECODE": "1", "PIP_CONFIG_FILE": os.devnull, "PIP_DISABLE_PIP_VERSION_CHECK": "1", "PIP_CACHE_DIR": str(self.root / "pip-cache"), "UV_CACHE_DIR": str(self.root / "uv-cache"), "UV_TOOL_DIR": str(self.root / "uv-tools"), "UV_TOOL_BIN_DIR": str(self.root / "uv-bin"), "UV_PYTHON_INSTALL_DIR": str(self.root / "uv-python"), "UV_NO_CONFIG": "1", "UV_PYTHON_DOWNLOADS": "never", "UV_DEFAULT_INDEX": "https://pypi.org/simple", "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull, } self.repo = self.root / "empty-repo" if not self.repo.exists(): self.run(["git", "init", "--initial-branch=main", str(self.repo)]) def save(self, name, value): (self.output / name).write_text(json.dumps(sanitize(value, self.root), indent=2, ensure_ascii=False) + "\n") def run(self, command, timeout=240): completed = subprocess.run(list(map(str, command)), env=self.env, capture_output=True, text=True, cwd=self.root, timeout=timeout) entry = {"command": list(map(str, command)), "exit_code": completed.returncode, "stdout": completed.stdout[-18000:], "stderr": completed.stderr[-18000:]} self.commands.append(entry) self.save("commands.json", self.commands) if completed.returncode: print(json.dumps(sanitize(entry, self.root), indent=2), flush=True) raise RuntimeError(f"Command failed with exit {completed.returncode}") return completed.stdout def metadata(self): packages = [("mcp-server-git", "2026.7.10"), ("mcp", "2.0.0")] provenance = [] for package, version in packages: url = f"https://pypi.org/pypi/{package}/{version}/json" with urllib.request.urlopen(url, timeout=40) as response: data = json.load(response) wheel = next(item for item in data["urls"] if item["packagetype"] == "bdist_wheel") with urllib.request.urlopen(wheel["url"], timeout=60) as response: blob = response.read() digest = hashlib.sha256(blob).hexdigest() if digest != wheel["digests"]["sha256"]: raise ValueError("PyPI wheel hash mismatch") (self.root / wheel["filename"]).write_bytes(blob) item = {"name": package, "version": version, "metadata_url": url, "requires_python": data["info"]["requires_python"], "requires_dist": data["info"]["requires_dist"], "filename": wheel["filename"], "url": wheel["url"], "sha256": digest, "pypi_hash_verified": True, "upload_time": wheel["upload_time_iso_8601"]} if package == "mcp-server-git": with zipfile.ZipFile(io.BytesIO(blob)) as archive: source = archive.read("mcp_server_git/server.py") self.server_source_hash = hashlib.sha256(source).hexdigest() item["server_py_sha256"] = self.server_source_hash text = source.decode() item["decorator_source_excerpt"] = "\n".join( f"{number}: {line}" for number, line in enumerate(text.splitlines(), 1) if "@server.list_tools()" in line or "@server.call_tool()" in line) provenance.append(item) self.save("provenance.json", provenance) def venv(self, name): target = self.root / name if target.exists(): raise ValueError(f"Refusing to reuse existing environment: {name}") self.run([sys.executable, "-m", "venv", str(target)]) return target / "bin" / "python" def install(self, python, label, requirements): report = self.root / f"{label}-pip-report.json" self.run([python, "-m", "pip", "--isolated", "--cache-dir", self.root / "pip-cache", "install", "--index-url", "https://pypi.org/simple", "--report", report, *requirements]) data = json.loads(report.read_text()) self.save(f"{label}-distributions.json", [{ "name": item["metadata"]["name"], "version": item["metadata"]["version"], "url": item["download_info"]["url"], "hashes": item["download_info"]["archive_info"].get("hashes", {}) } for item in data["install"]]) def snapshot(self, label, python): code = ( "import importlib.metadata as m,hashlib,json,pathlib,platform;" "d=m.distribution('mcp-server-git');" "p=pathlib.Path(d.locate_file('mcp_server_git/server.py'));" "print(json.dumps({'python':platform.python_version()," "'packages':{d.metadata['Name']:d.version for d in m.distributions()}," "'server_py_sha256':hashlib.sha256(p.read_bytes()).hexdigest()}))" ) environment = json.loads(self.run([python, "-c", code])) environment["server_source_matches_pypi_wheel"] = ( environment["server_py_sha256"] == self.server_source_hash) if not environment["server_source_matches_pypi_wheel"]: raise ValueError("Installed server source differs from official wheel") self.save(f"{label}-environment.json", environment) freeze = "\n".join(f"{name}=={version}" for name, version in sorted(environment["packages"].items(), key=lambda x: x[0].lower())) (self.output / f"{label}-requirements.txt").write_text(freeze + "\n") checked = probe([str(python), "-m", "mcp_server_git", "--repository", str(self.repo)], self.repo) self.save(f"{label}-protocol.json", checked) print(json.dumps({"scenario": label, "mcp": environment["packages"].get("mcp"), "success": checked["success"], "list_tools_attribute_error": checked["list_tools_attribute_error"], "initialize_ok": checked["initialize_ok"], "tools_list_ok": checked["tools_list_ok"], "tool_call_ok": checked["tool_call_ok"]}), flush=True) return checked def broken(self): python = self.venv("broken") self.install(python, "broken", [SERVER, "mcp==2.0.0"]) checked = self.snapshot("broken", python) assert checked["list_tools_attribute_error"] and not checked["initialize_ok"] assert checked["exit_code"] == 1 and not checked["mcp_error_import_failure"] def natural(self): python = self.venv("natural") self.install(python, "natural", [SERVER]) self.snapshot("natural", python) def pip_repair(self): python = self.venv("pip-repaired") self.install(python, "pip-before", [SERVER, "mcp==2.0.0"]) before = self.snapshot("pip-before", python) assert before["list_tools_attribute_error"] self.install(python, "pip-after", [SERVER, "mcp<2"]) self.run([python, "-m", "pip", "check"]) after = self.snapshot("pip-after", python) assert after["success"] and after["exit_code"] == 0 def uv_repair(self): uv = shutil.which("uv") if not uv: raise RuntimeError("uv required for uv tool scenario") python = self.root / "uv-tools" / "mcp-server-git" / "bin" / "python" self.run([uv, "tool", "install", "--python", sys.executable, "--with", "mcp==2.0.0", SERVER]) before = self.snapshot("uv-before", python) assert before["list_tools_attribute_error"] self.run([uv, "tool", "install", "--python", sys.executable, "--force", "--with", "mcp<2", SERVER]) self.run([uv, "pip", "check", "--python", python]) after = self.snapshot("uv-after", python) assert after["success"] and after["exit_code"] == 0 entrypoint = probe([str(self.root / "uv-bin" / "mcp-server-git"), "--repository", str(self.repo)], self.repo) self.save("uv-entrypoint-protocol.json", entrypoint) assert entrypoint["success"] and entrypoint["exit_code"] == 0 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--scenario", choices=["all", "broken", "natural", "pip", "uv"], default="all") parser.add_argument("--run-root", type=Path) parser.add_argument("--output-dir", type=Path, help="New or empty evidence directory; defaults to /evidence") args = parser.parse_args() root = args.run_root or Path(tempfile.mkdtemp(prefix="asn-real-mcp-", dir="/private/tmp")) experiment = Experiment(root, args.output_dir or root / "evidence") experiment.save("host.json", {"captured_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "platform": platform.system(), "platform_release": platform.release(), "machine": platform.machine(), "python": platform.python_version(), "uv": subprocess.check_output(["uv", "--version"], text=True).strip(), "git": subprocess.check_output(["git", "--version"], text=True).strip(), "isolated_environment_variables": sorted(experiment.env), "temporary_path_token": "$RUN_ROOT"}) experiment.metadata() for name, method in [("broken", experiment.broken), ("natural", experiment.natural), ("pip", experiment.pip_repair), ("uv", experiment.uv_repair)]: if args.scenario in ("all", name): method() print("Selected reproduction scenarios completed and checked.", flush=True) if __name__ == "__main__": main()