#!/usr/bin/env python3"""naive.py — 素朴な解決器。serverUrl だけを見る"""import json, os, glob, yamldef load_agent(path): raw = open(path).read() _, fm, body = raw.split("---", 2) return yaml.safe_load(fm), bodydef servers_of(cfg_path): cfg = json.load(open(cfg_path)) out = {} for name, spec in cfg.get("mcpServers", {}).items(): if "serverUrl" not in spec: # ← ここが素朴な点 continue out[name] = set(spec.get("tools", [])) return outdef check(repo): fm, _ = load_agent(os.path.join(repo, ".antigravity/agents/release-notes.md")) srv = servers_of(os.path.join(repo, ".antigravity/mcp_config.json")) missing = [] for ref in fm.get("requiredMcpTools", []): s, t = ref.split(".", 1) if s not in srv or t not in srv[s]: missing.append(ref) return missingif __name__ == "__main__": for repo in sorted(glob.glob("/tmp/exp/fixture/*")): m = check(repo) print(f"{os.path.basename(repo):20s} {'OK' if not m else 'MISSING ' + ','.join(m)}")
#!/usr/bin/env python3"""preflight.py — agent.md の能力契約を移植先で解決する。解決できなければ走らせない"""from __future__ import annotationsimport json, os, sys, glob, time, yamlfrom dataclasses import dataclass, fieldURL_KEYS = ("url", "serverUrl") # 1.1.6 時点で両方が有効SERVER_ALIASES = {"gh": "github", "github": "github", "slack": "slack"}@dataclassclass Finding: kind: str # mcp-server / mcp-tool / command ref: str detail: str@dataclassclass Report: repo: str findings: list[Finding] = field(default_factory=list) @property def blocked(self) -> bool: return bool(self.findings)def read_agent(path: str) -> dict: raw = open(path, encoding="utf-8").read() if not raw.startswith("---"): raise ValueError(f"{path}: フロントマターがありません") _, fm, _body = raw.split("---", 2) return yaml.safe_load(fm) or {}def normalize_servers(cfg_path: str) -> dict[str, set[str]]: """serverUrl / url のどちらで書かれていても同じ形にそろえる""" cfg = json.load(open(cfg_path, encoding="utf-8")) out: dict[str, set[str]] = {} for raw_name, spec in cfg.get("mcpServers", {}).items(): if not any(k in spec for k in URL_KEYS): continue # stdio 起動などは対象外 name = SERVER_ALIASES.get(raw_name, raw_name) out.setdefault(name, set()).update(spec.get("tools", [])) return outdef snake(s: str) -> str: """listReleases と list_releases を同じものとして扱う""" return "".join("_" + c.lower() if c.isupper() else c for c in s).lstrip("_")def check(repo: str) -> Report: rep = Report(repo=os.path.basename(repo)) fm = read_agent(os.path.join(repo, ".antigravity/agents/release-notes.md")) servers = normalize_servers(os.path.join(repo, ".antigravity/mcp_config.json")) for ref in fm.get("requiredMcpTools", []): raw_srv, tool = ref.split(".", 1) srv = SERVER_ALIASES.get(raw_srv, raw_srv) if srv not in servers: rep.findings.append(Finding("mcp-server", ref, f"サーバー {srv} が未定義")) continue have = {snake(t) for t in servers[srv]} if snake(tool) not in have: rep.findings.append(Finding("mcp-tool", ref, f"{srv} に {tool} がない")) if fm.get("commandExecutionPolicy") == "allowlist": available = set(json.load(open(os.path.join(repo, "commands.json")))) for cmd in fm.get("allowedCommands", []): head = cmd.split()[0] if head not in available: rep.findings.append(Finding("command", cmd, f"{head} が移植先に無い")) return repdef main(paths: list[str]) -> int: t0 = time.perf_counter() reports = [check(p) for p in sorted(paths)] elapsed = (time.perf_counter() - t0) * 1000 for r in reports: if not r.blocked: print(f"✅ {r.repo:20s} 解決済み") else: print(f"⛔ {r.repo:20s} {len(r.findings)} 件") for f in r.findings: print(f" [{f.kind:11s}] {f.ref} — {f.detail}") blocked = sum(1 for r in reports if r.blocked) kinds: dict[str, int] = {} for r in reports: for f in r.findings: kinds[f.kind] = kinds.get(f.kind, 0) + 1 print(f"\n{len(reports)} リポジトリ / 停止 {blocked} / {elapsed:.1f} ms") print("種別:", ", ".join(f"{k}={v}" for k, v in sorted(kinds.items()))) return 1 if blocked else 0if __name__ == "__main__": sys.exit(main(sys.argv[1:] or glob.glob("/tmp/exp/fixture/*")))