夜間に回すMCPコネクタの健康状態を自前で見える化する — 個人運用のための軽量ヘルス台帳

# health-ledger.sh — 1回のMCPツール呼び出しを1行で記録する
LEDGER="${LEDGER:-$HOME/ops/mcp-health.jsonl}"
mkdir -p "$(dirname "$LEDGER")"
 
record_call() {
  # 使い方: record_call <connector> <tool> -- <実行するコマンド...>
  local connector="$1" tool="$2"; shift 2
  [ "$1" = "--" ] && shift
  local start_ms end_ms status err latency
  start_ms=$(date +%s%3N)                 # GNU date のミリ秒エポック
  if "$@"; then
    status="ok"; err=""
  else
    status="fail"; err="exit_$?"          # 失敗したコマンドの終了コードを残す
  fi
  end_ms=$(date +%s%3N)
  latency=$(( end_ms - start_ms ))
  printf '{"ts":"%s","connector":"%s","tool":"%s","latency_ms":%d,"status":"%s","err":"%s"}\n' \
    "$(TZ=Asia/Tokyo date -Iseconds)" "$connector" "$tool" "$latency" "$status" "$err" \
    >> "$LEDGER"
}

source health-ledger.sh
 
record_call github issue_list -- gh issue list --repo me/app --limit 20
record_call stripe recent_charges -- node scripts/fetch-charges.mjs
record_call notion sync_pages   -- node scripts/notion-sync.mjs

#!/usr/bin/env python3
# rollup.py — ヘルス台帳をコネクタ単位のエラー率とp95レイテンシに丸める
import json, sys, collections
 
def percentile(values, p):
    if not values:
        return 0
    s = sorted(values)
    k = (len(s) - 1) * p
    lo = int(k)
    hi = min(lo + 1, len(s) - 1)
    return round(s[lo] + (s[hi] - s[lo]) * (k - lo))  # 線形補間
 
emit_json = "--json" in sys.argv
rows = collections.defaultdict(lambda: {"total": 0, "fail": 0, "lat": []})
 
for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    r = json.loads(line)
    b = rows[r["connector"]]
    b["total"] += 1
    if r["status"] != "ok":
        b["fail"] += 1
    b["lat"].append(r["latency_ms"])
 
if emit_json:
    for name, b in sorted(rows.items()):
        print(json.dumps({
            "connector": name, "calls": b["total"],
            "err_pct": round(100 * b["fail"] / b["total"], 1),
            "p50": percentile(b["lat"], 0.5),
            "p95": percentile(b["lat"], 0.95),
        }, ensure_ascii=False))
else:
    print(f'{"connector":<12}{"calls":>7}{"err%":>8}{"p50":>9}{"p95":>9}')
    for name, b in sorted(rows.items()):
        err = 100 * b["fail"] / b["total"]
        print(f'{name:<12}{b["total"]:>7}{err:>7.1f}%'
              f'{percentile(b["lat"],0.5):>7}ms{percentile(b["lat"],0.95):>7}ms')

cat ~/ops/mcp-health.jsonl | python3 rollup.py

#!/usr/bin/env python3
# regress.py — 今週の集計を前週ベースラインと比べ、悪化だけを知らせる
import json, sys
 
THRESHOLDS = {"err_pct": 5.0, "p95_ratio": 1.5}  # エラー率5%超 or p95が1.5倍
 
def load(path):
    with open(path) as f:
        return {r["connector"]: r
                for r in (json.loads(l) for l in f if l.strip())}
 
cur  = load(sys.argv[1])   # 今週の集計（rollup.py --json の出力）
base = load(sys.argv[2])   # 前週のベースライン
 
alerts = []
for name, c in cur.items():
    if c["err_pct"] > THRESHOLDS["err_pct"]:
        alerts.append(f'{name}: エラー率 {c["err_pct"]:.1f}%'
                      f'（しきい値 {THRESHOLDS["err_pct"]}%）')
    b = base.get(name)
    if b and b["p95"] > 0 and c["p95"] / b["p95"] >= THRESHOLDS["p95_ratio"]:
        alerts.append(f'{name}: p95 {b["p95"]}ms→{c["p95"]}ms'
                      f'（{c["p95"] / b["p95"]:.1f}倍）')
 
if alerts:
    print("⚠️ 悪化を検知しました")
    for a in alerts:
        print(" -", a)
    sys.exit(1)   # 非ゼロ終了でスケジュールタスク側に失敗を伝える
print("✅ しきい値内です")

# 今週分を集計してJSONで保存
cat ~/ops/mcp-health.jsonl | python3 rollup.py --json > ~/ops/summary-cur.jsonl
 
# 前週ベースラインと比較（悪化があれば非ゼロ終了）
python3 regress.py ~/ops/summary-cur.jsonl ~/ops/summary-base.jsonl \
  || echo "要確認: 今週のコネクタ健康状態に悪化あり"
 
# 問題なければ今週を次回のベースラインに繰り上げる
cp ~/ops/summary-cur.jsonl ~/ops/summary-base.jsonl