MCPサーバーは繋がっているのに、ツールが0件で返る — 無人実行の権限劣化をツール表の差分で止める

{
  "mcpServers": {
    "report": {
      "command": "python3",
      "args": ["./report_server.py"],
      "env": { "REPORT_API_TOKEN": "${REPORT_API_TOKEN}" }
    }
  }
}

#!/usr/bin/env python3
"""実効ツール表がロックファイルと違えば fail closed する preflight。"""
import json, os, subprocess, sys, time
 
CONFIG = os.environ.get("MCP_CONFIG", ".mcp.json")
LOCK = os.environ.get("MCP_TOOL_LOCK", "mcp-tools.lock.json")
TIMEOUT_S = float(os.environ.get("MCP_PROBE_TIMEOUT_S", "10"))
 
 
def expand(value: str) -> str:
    # 設定内の ${VAR} は「起動時の環境」から解決される
    if value.startswith("${") and value.endswith("}"):
        return os.environ.get(value[2:-1], "")
    return value
 
 
def probe(name, spec):
    env = {k: v for k, v in os.environ.items()}
    for k, v in (spec.get("env") or {}).items():
        env[k] = expand(v)
    try:
        # Popen は try の内側に置く。コマンド名の誤りは FileNotFoundError として
        # ここで飛び、外に漏らすと「設定ミス」が「preflightの障害」に化ける
        proc = subprocess.Popen([spec["command"], *spec.get("args", [])],
                                stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                stderr=subprocess.DEVNULL, text=True, env=env, bufsize=1)
 
        def call(method, params=None, rid=None):
            msg = {"jsonrpc": "2.0", "method": method}
            if params is not None:
                msg["params"] = params
            if rid is not None:
                msg["id"] = rid
            proc.stdin.write(json.dumps(msg) + "\n"); proc.stdin.flush()
            return json.loads(proc.stdout.readline()) if rid is not None else None
 
        call("initialize", {"protocolVersion": "2026-07-28", "capabilities": {},
                            "clientInfo": {"name": "mcp-preflight", "version": "1.0"}}, rid=1)
        call("notifications/initialized")
        listed = call("tools/list", {}, rid=2)
        call("shutdown", None, rid=3)
        proc.wait(timeout=TIMEOUT_S)
        return sorted(t["name"] for t in listed.get("result", {}).get("tools", []))
    except Exception as exc:          # 起動失敗・壊れたフレーム・タイムアウト
        if "proc" in dir():
            proc.kill()
        return {"error": f"{type(exc).__name__}: {exc}"}
 
 
def main():
    started = time.perf_counter()
    servers = json.load(open(CONFIG))["mcpServers"]
    observed = {name: probe(name, spec) for name, spec in servers.items()}
 
    if "--update-lock" in sys.argv:
        json.dump(observed, open(LOCK, "w"), indent=2, sort_keys=True)
        print(f"wrote {LOCK}")
        return 0
 
    expected = json.load(open(LOCK))
    failures = []
    for name, want in expected.items():
        got = observed.get(name)
        if isinstance(got, dict):
            failures.append(f"{name}: probe failed ({got['error']})")
            continue
        missing = [t for t in want if t not in (got or [])]
        if missing:
            failures.append(f"{name}: {len(missing)}/{len(want)} tools missing -> {missing}")
 
    took = (time.perf_counter() - started) * 1000
    if failures:
        print(f"MCP preflight FAILED in {took:.0f} ms")
        for f in failures:
            print(f"  - {f}")
        return 78                     # EX_CONFIG: 実行時障害ではなく配線の問題
    print(f"MCP preflight OK in {took:.0f} ms ({sum(len(v) for v in expected.values())} tools verified)")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main())

$ REPORT_API_TOKEN=test-token python3 preflight.py --update-lock
wrote mcp-tools.lock.json
 
$ cat mcp-tools.lock.json
{
  "report": [
    "fetch_daily_report",
    "list_report_dates"
  ]
}
 
$ REPORT_API_TOKEN=test-token python3 preflight.py; echo "exit=$?"
MCP preflight OK in 23 ms (2 tools verified)
exit=0
 
$ env -u REPORT_API_TOKEN python3 preflight.py; echo "exit=$?"
MCP preflight FAILED in 23 ms
  - report: 2/2 tools missing -> ['fetch_daily_report', 'list_report_dates']
exit=78

FileNotFoundError: [Errno 2] No such file or directory: './no-such-binary'
exit=1

MCP preflight FAILED in 1 ms
  - ghost: probe failed (FileNotFoundError: [Errno 2] No such file or directory: './no-such-binary')
exit=78

$ REPORT_API_TOKEN=test-token MCP_PROBE_TIMEOUT_S=10 timeout 30 python3 preflight.py
exit=124  elapsed=30s        # 124 は timeout(1) による強制終了

#!/usr/bin/env python3
"""stdio プローブのための、期限つき1行読み取り。"""
import json, selectors, subprocess, time
 
 
class ProbeTimeout(Exception):
    pass
 
 
def read_line_before(stream, deadline):
    """1行返す。期限を過ぎたら ProbeTimeout を送出する。"""
    sel = selectors.DefaultSelector()
    sel.register(stream, selectors.EVENT_READ)
    try:
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise ProbeTimeout("no response before deadline")
            if not sel.select(timeout=remaining):
                continue                      # 早く起きただけ。ループ先頭で時計を見直す
            line = stream.readline()          # 読める状態なので、ここではもうブロックしない
            if line == "":
                raise ProbeTimeout("server closed the stream")
            return line
    finally:
        sel.unregister(stream)
        sel.close()
 
 
def probe(command, args, env, budget_s=10.0):
    deadline = time.monotonic() + budget_s
    proc = subprocess.Popen([command, *args], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                            stderr=subprocess.DEVNULL, text=True, env=env, bufsize=1)
    try:
        def call(method, params=None, rid=None):
            msg = {"jsonrpc": "2.0", "method": method}
            if params is not None:
                msg["params"] = params
            if rid is not None:
                msg["id"] = rid
            proc.stdin.write(json.dumps(msg) + "\n")
            proc.stdin.flush()
            if rid is None:
                return None
            return json.loads(read_line_before(proc.stdout, deadline))
 
        call("initialize", {"protocolVersion": "2026-07-28", "capabilities": {},
                            "clientInfo": {"name": "mcp-preflight", "version": "1.1"}}, rid=1)
        call("notifications/initialized")
        listed = call("tools/list", {}, rid=2)
        return sorted(t["name"] for t in listed["result"]["tools"])
    finally:
        proc.kill()                            # 期限を使い切った相手を、行儀よく待つ理由はない
        proc.wait(timeout=2)

healthy: tools=['fetch_daily_report', 'list_report_dates'] in 24 ms
stalled: ProbeTimeout(no response before deadline) after 2004 ms (budget 2.0s)

def probe_remote(spec, budget_s):
    """リモート（HTTP）のMCPサーバーを叩く。保持するセッションはない。"""
    url = spec["url"]
    headers = {"Content-Type": "application/json"}
    headers.update({k: expand(v) for k, v in (spec.get("headers") or {}).items()})
    deadline = time.monotonic() + budget_s
 
    def rpc(method, params, rid):
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise ProbeTimeout("budget spent before request")
        req = urllib.request.Request(url, data=json.dumps(
            {"jsonrpc": "2.0", "id": rid, "method": method, "params": params}).encode(),
            method="POST", headers=headers)
        with urllib.request.urlopen(req, timeout=remaining) as resp:
            return json.loads(resp.read())
 
    rpc("initialize", {"protocolVersion": "2026-07-28", "capabilities": {},
                       "clientInfo": {"name": "mcp-preflight", "version": "1.1"}}, 1)
    listed = rpc("tools/list", {}, 2)
    return sorted(t["name"] for t in listed["result"]["tools"])
 
 
def probe(spec, budget_s=10.0):
    """ツール名の並びを返す。失敗した場合は理由を持つ dict を返す。"""
    handler = probe_remote if spec.get("type") in ("http", "sse") or "url" in spec else probe_stdio
    try:
        return handler(spec, budget_s)
    except Exception as exc:
        return {"error": f"{type(exc).__name__}: {exc}"}

token present   tool_count=2  names=['fetch_daily_report', 'list_report_dates']  elapsed_ms=[4.0, 1.7, 1.5]
token missing   tool_count=0  names=[]                                           elapsed_ms=[1.3, 1.3, 1.3]

exact-match expand (published version): header -> 'Bearer ${REPORT_API_TOKEN}'
regex expand (fixed):                   header -> 'Bearer test-token'

VAR = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
 
 
def expand(value: str) -> str:
    """値の一部に埋め込まれた ${VAR} も展開する。"""
    return VAR.sub(lambda m: os.environ.get(m.group(1), ""), value)

{
  "report": ["fetch_daily_report", "list_report_dates"],
  "report-remote": ["fetch_daily_report", "list_report_dates"],
  "stalled": { "error": "ProbeTimeout: no response before deadline" }
}

if "--update-lock" in sys.argv:
    broken = {n: v["error"] for n, v in observed.items() if isinstance(v, dict)}
    if broken:
        print("refusing to write the lock: some probes failed")
        for n, e in broken.items():
            print(f"  - {n}: {e}")
        return 78
    json.dump(observed, open(LOCK, "w"), indent=2, sort_keys=True)
    print(f"wrote {LOCK}")
    return 0

$ ... --update-lock            # 応答のないサーバーを含む構成
refusing to write the lock: some probes failed
  - stalled: ProbeTimeout: no response before deadline
exit=78
 
$ ... --update-lock            # stdio + リモートの2台
wrote mcp-tools.lock.json
exit=0
 
$ REPORT_API_TOKEN=test-token python3 preflight.py; echo "exit=$?"
MCP preflight OK in 29 ms (4 tools verified)
exit=0
 
$ env -u REPORT_API_TOKEN python3 preflight.py; echo "exit=$?"
MCP preflight FAILED in 27 ms
  - report: 2/2 tools missing -> ['fetch_daily_report', 'list_report_dates']
  - report-remote: 2/2 tools missing -> ['fetch_daily_report', 'list_report_dates']
exit=78

MCP preflight FAILED in 2032 ms
  - stalled: probe failed (ProbeTimeout: no response before deadline)
exit=78

- name: MCP preflight
  env:
    REPORT_API_TOKEN: ${{ secrets.REPORT_API_TOKEN }}
  run: python3 tools/preflight.py
 
- name: Run unattended agent
  run: claude --print --output-format json "$(cat prompts/nightly.md)"