AWS MCP Server が東京リージョンをサポートしたので試してみた

#!/usr/bin/env python3
"""AWS MCP Server の東京/バージニア エンドポイント応答時間を、同一ホストから交互に測定する。

- 標準ライブラリのみ。mcp-proxy-for-aws を stdio で起動し、JSON-RPC で tools/call を送る。
- 認証はホストの default プロファイル（環境変数・ファイルの持ち出しなし）。
- 1 呼び出しごとに JSONL を即時追記する（途中で止まっても記録が残る）。
- 呼び出し順は反復ごとに入れ替え、順序による偏りを避ける。
使い方: measure-mcp-latency.py <出力jsonl> <反復回数>
"""
import json, select, subprocess, sys, time

OUT, N = sys.argv[1], int(sys.argv[2])
UVX = "uvx"
ENDPOINTS = ["ap-northeast-1", "us-east-1"]
STS_CODE = ('r = await call_boto3(service_name="sts", operation_name="GetCallerIdentity")\n'
            'result = {"has_arn": "Arn" in r}\nresult')
TOOLS = [
    ("sts_run_script", "aws___run_script", {"code": STS_CODE}),
    ("list_regions", "aws___list_regions", {}),
]

class Proxy:
    def __init__(self, ep):
        self.ep, self.seq = ep, 0
        self.p = subprocess.Popen(
            [UVX, "mcp-proxy-for-aws-cli@latest", f"https://aws-mcp.{ep}.api.aws/mcp",
             "--metadata", "AWS_REGION=ap-northeast-1"],
            stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
            text=True, bufsize=1)

    def send(self, o):
        self.p.stdin.write(json.dumps(o) + "\n")
        self.p.stdin.flush()

    def recv(self, id_, timeout=90):
        end = time.time() + timeout
        while time.time() < end:
            r, _, _ = select.select([self.p.stdout], [], [], 1)
            if not r:
                continue
            line = self.p.stdout.readline()
            if not line:
                return None
            try:
                m = json.loads(line)
            except ValueError:
                continue
            if m.get("id") == id_:
                return m
        return None

    def init(self):
        self.send({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
            "protocolVersion": "2025-03-26", "capabilities": {},
            "clientInfo": {"name": "latency-probe", "version": "1"}}})
        m = self.recv(1, 120)
        self.send({"jsonrpc": "2.0", "method": "notifications/initialized"})
        self.seq = 1
        return bool(m and "result" in m)

    def call(self, name, args):
        self.seq += 1
        t0 = time.perf_counter()
        self.send({"jsonrpc": "2.0", "id": self.seq, "method": "tools/call",
                   "params": {"name": name, "arguments": args}})
        m = self.recv(self.seq)
        dt = time.perf_counter() - t0
        ok = bool(m and "result" in m and not m["result"].get("isError"))
        err = None if ok else json.dumps(m, ensure_ascii=False)[:300]
        return dt, ok, err

def main():
    px = {ep: Proxy(ep) for ep in ENDPOINTS}
    with open(OUT, "a", buffering=1) as f:
        def rec(**kw):
            kw["wall_time"] = time.strftime("%Y-%m-%dT%H:%M:%S%z")
            f.write(json.dumps(kw, ensure_ascii=False) + "\n")

        for ep in ENDPOINTS:
            t = time.perf_counter()
            ok = px[ep].init()
            rec(kind="init", endpoint=ep, seconds=round(time.perf_counter() - t, 4), ok=ok)
        for label, tool, args in TOOLS:
            for i in range(N):
                order = ENDPOINTS if i % 2 == 0 else ENDPOINTS[::-1]
                for ep in order:
                    dt, ok, err = px[ep].call(tool, args)
                    rec(kind="call", test=label, endpoint=ep, iter=i,
                        seconds=round(dt, 4), ok=ok, error=err)
        for q in px.values():
            q.p.terminate()

main()

import json,sys,statistics as s
rows=[json.loads(l) for l in open(sys.argv[1])]
print("init:",[(r["endpoint"],r["seconds"],r["ok"]) for r in rows if r["kind"]=="init"])
calls=[r for r in rows if r["kind"]=="call"]
print("失敗:",[r for r in calls if not r["ok"]] or "なし")
def q(v,p): v=sorted(v); return v[min(len(v)-1,int(round(p*(len(v)-1))))]
for t in ("sts_run_script","list_regions"):
    for ep in ("ap-northeast-1","us-east-1"):
        a=[r for r in calls if r["test"]==t and r["endpoint"]==ep]
        first=a[0]["seconds"]; rest=[r["seconds"] for r in a[1:]]
        print(f"{t:15s} {ep:15s} 初回={first:.2f}s  2回目以降 n={len(rest)} min={min(rest):.2f} med={s.median(rest):.2f} p90={q(rest,.9):.2f} max={max(rest):.2f} mean={s.mean(rest):.2f}")
    # 対応ペア差（同一反復の東京-バージニア）
    d=[]
    for i in range(1,30):
        tk=[r for r in calls if r["test"]==t and r["iter"]==i and r["endpoint"]=="ap-northeast-1"][0]["seconds"]
        us=[r for r in calls if r["test"]==t and r["iter"]==i and r["endpoint"]=="us-east-1"][0]["seconds"]
        d.append(us-tk)
    print(f"{t:15s} ペア差(バージニア-東京) med={s.median(d):+.2f}s mean={s.mean(d):+.2f}s 東京が速い回={sum(x>0 for x in d)}/{len(d)}")