MCP を束ねると read_file が 3 つになる — ツール名の衝突を起動前に落とす

#!/usr/bin/env python3
"""MCP ツール名の衝突を起動前に検出し、決定的な別名表を出力する。
 
exit 0: 衝突なし / exit 1: 完全一致の衝突あり / exit 2: 設定の読み込み失敗
"""
from __future__ import annotations
 
import hashlib
import json
import re
import sys
from collections import defaultdict
 
NAME_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
MAX_LEN = 64
 
# 語形の揺れを吸収する。左が正規形。
SYNONYMS = {
    "get": "read", "fetch": "read", "load": "read",
    "find": "search", "query": "search", "grep": "search",
    "ls": "list", "dir": "list", "directory": "list",
    "put": "write", "save": "write", "update": "write",
    "rm": "delete", "remove": "delete",
    "file": "file", "files": "file", "contents": "file", "content": "file",
    "page": "page", "pages": "page",
}
 
 
def normalize(name: str) -> str:
    """比較用の正規形。区切り記号を落とし、語ごとに同義語を畳む。"""
    words = [w for w in re.split(r"[^a-z0-9]+", name.lower()) if w]
    folded = sorted({SYNONYMS.get(w, w) for w in words})
    return "_".join(folded)
 
 
def short(server: str, budget: int) -> str:
    """サーバー名を安全な接頭辞へ。長い場合は決定的に縮める。"""
    slug = re.sub(r"[^A-Za-z0-9]+", "_", server).strip("_").lower() or "srv"
    if len(slug) <= budget:
        return slug
    digest = hashlib.sha1(server.encode("utf-8")).hexdigest()[:4]
    return slug[: max(1, budget - 5)] + "_" + digest
 
 
def load(path: str) -> dict[str, list[str]]:
    try:
        with open(path, encoding="utf-8") as fh:
            raw = json.load(fh)
    except FileNotFoundError:
        print(f"error: 設定ファイルが見つかりません: {path}", file=sys.stderr)
        raise SystemExit(2)
    except json.JSONDecodeError as exc:
        # 設定を手で編集していると、ここで落ちるのが一番多い
        print(f"error: JSON を解析できません ({exc.lineno}行目): {exc.msg}", file=sys.stderr)
        raise SystemExit(2)
 
    servers = raw.get("mcpServers")
    if not isinstance(servers, dict):
        print("error: mcpServers オブジェクトがありません", file=sys.stderr)
        raise SystemExit(2)
 
    catalog: dict[str, list[str]] = {}
    for server, spec in sorted(servers.items()):
        tools = spec.get("tools") if isinstance(spec, dict) else None
        if not tools:
            # tools を静的に持たないサーバーは起動して tools/list を叩く必要がある。
            # 黙って飛ばさず、確認漏れとして明示する。
            print(f"warn: {server}: tools 未記載のため静的検査から除外", file=sys.stderr)
            continue
        catalog[server] = list(dict.fromkeys(str(t) for t in tools))
    return catalog
 
 
def audit(catalog: dict[str, list[str]]):
    exact: dict[str, list[str]] = defaultdict(list)
    near: dict[str, list[tuple[str, str]]] = defaultdict(list)
    invalid: list[tuple[str, str]] = []
 
    for server, tools in catalog.items():
        for tool in tools:
            if not NAME_RE.match(tool):
                invalid.append((server, tool))
            exact[tool].append(server)
            near[normalize(tool)].append((server, tool))
 
    exact_hits = {n: s for n, s in exact.items() if len(s) > 1}
    near_hits = {}
    for key, pairs in near.items():
        distinct_names = {t for _, t in pairs}
        distinct_servers = {s for s, _ in pairs}
        if len(distinct_servers) > 1 and len(distinct_names) > 1:
            near_hits[key] = sorted(pairs)
    return exact_hits, near_hits, invalid
 
 
def alias_map(catalog: dict[str, list[str]], exact_hits: dict[str, list[str]]):
    """衝突した名前にだけ接頭辞を付ける。無衝突の名前は触らない。"""
    taken = {t for tools in catalog.values() for t in tools}
    mapping: dict[tuple[str, str], str] = {}
    for server in sorted(catalog):
        for tool in catalog[server]:
            if tool not in exact_hits:
                continue
            budget = MAX_LEN - len(tool) - 1
            if budget < 3:
                # 名前自体が長すぎる。切り詰めではなく明示的な失敗にする
                raise ValueError(f"{server}.{tool}: 別名が {MAX_LEN} 文字に収まりません")
            candidate = f"{short(server, budget)}_{tool}"
            if candidate in taken:
                tail = hashlib.sha1(f"{server}/{tool}".encode()).hexdigest()[:4]
                candidate = f"{candidate[: MAX_LEN - 5]}_{tail}"
            taken.add(candidate)
            mapping[(server, tool)] = candidate
    return mapping
 
 
def main(argv: list[str]) -> int:
    path = argv[1] if len(argv) > 1 else "mcp_config.json"
    catalog = load(path)
    total = sum(len(v) for v in catalog.values())
    exact_hits, near_hits, invalid = audit(catalog)
 
    print(f"検査対象: {len(catalog)} サーバー / {total} ツール")
 
    for server, tool in invalid:
        print(f"NG  形式違反 {server}.{tool}（^[A-Za-z0-9_-]{{1,64}}$ を満たしません）")
 
    if exact_hits:
        print(f"\n完全一致の衝突: {len(exact_hits)} 件")
        for name in sorted(exact_hits):
            print(f"  {name}  <- {', '.join(sorted(exact_hits[name]))}")
        try:
            mapping = alias_map(catalog, exact_hits)
        except ValueError as exc:
            print(f"NG  別名を生成できません: {exc}")
            return 1
        print("\n別名表（決定的・この順序で固定）:")
        for (server, tool), alias in sorted(mapping.items()):
            print(f"  {server}.{tool} -> {alias}")
 
    if near_hits:
        print(f"\n意味的に近い名前: {len(near_hits)} 件（自動改名の対象外・目視確認）")
        for key in sorted(near_hits):
            pairs = ", ".join(f"{s}.{t}" for s, t in near_hits[key])
            print(f"  [{key}]  {pairs}")
 
    if not exact_hits and not near_hits and not invalid:
        print("衝突なし")
    return 1 if (exact_hits or invalid) else 0
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv))

#!/usr/bin/env bash
# agy-run — 衝突検査を通してから Antigravity CLI を起動する
set -euo pipefail
 
CONFIG="${MCP_CONFIG:-$HOME/.antigravity/mcp_config.json}"
 
if ! python3 "$(dirname "$0")/mcp_preflight.py" "$CONFIG"; then
  echo "---" >&2
  echo "ツール名が衝突しています。別名表を反映してから再実行してください。" >&2
  exit 1
fi
 
exec antigravity "$@"

- name: MCP tool name preflight
  run: python3 tools/mcp_preflight.py .antigravity/mcp_config.json