監査ログ・スキーマ検証・タイムアウト・呼び出し制限を加える

目標

前のラボで防御したサーバーに、運用に必要な4つのものを加えます。すべての呼び出しを記録する監査ログ、実行前の引数スキーマ検証、ツールごとの実行時間の上限、破壊的ツールの呼び出し回数の制限です。そして、そのログを読んで集計するスクリプトまで書きます。

なぜ重要なのか

事故が起きたときの最初の問いは、「誰が、いつ、何を、どんな引数で呼んだのか」です。監査ログがなければ、この問いに答えられず、そうなると同じ事故がまた起きます。仕様は、サーバーがすべてのツール入力を検証し、呼び出し頻度を制限しなければならない(MUST)こと、クライアントはツール呼び出しにタイムアウトを設け、監査のために使用を記録すべきである(SHOULD)ことを書いています。その文をそのままコードに移すのが、このラボです。とくにタイムアウトは、クライアントだけを信じてはいけません。サーバーが自分で打ち切らなければ、止まったツール1つがセッション全体を占有します。

ステップ

シードSQL(/root/mcp/audit/seed.sql)を保存し、DB(/root/mcp/audit/shop.db)へ読み込んでください。customersが5行、ordersが8行になっている必要があります。
サーバー(/root/mcp/audit/server.py)を作成してください。initializeに返答し、tools/listにlist_customers・count_orders・delete_order(confirmが必要)の3つのツールを返します。DBのパスはMCP_DB(デフォルトは/root/mcp/audit/shop.db)です。
すべてのtools/callを、1行のJSONとして、ファイル(/root/mcp/audit/audit.jsonl)に残してください。パスは環境変数MCP_AUDIT_LOGで変更できるようにします。キーはts、tool、arguments、ok、duration_msで、ツールがisErrorを返したときは、okはfalseです。
ツールを実行する前に、inputSchemaで引数を検査してください。requiredが欠けていたり、typeが合っていなかったりする場合(例: statusに数値、idに文字列)は、実行せずにJSON-RPCエラー-32602で返答します。
ツールslow_report(引数はseconds、integer)を追加し、ツール1つの実行時間を2秒に制限してください(環境変数MCP_TOOL_TIMEOUT、デフォルトは2)。超えたときは、isError: trueとtimeoutを含むテキストで返答し、監査ログにok: falseとして残ります。seconds: 1は正常に終わる必要があります。
initializeのcapabilitiesにloggingを宣言し、ツール呼び出しごとに、レスポンスの前にnotifications/message通知(levelはRFC 5424の段階のうちの1つ、logger、data)をstdoutに出力してください。失敗した呼び出しはerrorのレベルです。
delete_orderは、1セッションにつき3回までだけ許可してください。4回目の呼び出しは削除せず、isError: trueとrate limitを含むテキストで返答します。
集計スクリプト(/root/mcp/audit/summary.py)を作成してください。audit.jsonlを読み、ツールごとの{"calls": n, "failed": m}を、ファイル(/root/mcp/audit/summary.json)に書き込みます。環境変数MCP_SUMMARY_OUTがあれば、そのパスに書きます。実際のaudit.jsonlに、呼び出しの記録が5行以上たまっている必要があります。

参考

監査の1行は、json.dumps({...})をopen(path, "a")で追記すれば足ります。tsはdatetime.now(timezone.utc).isoformat()、duration_msはtime.monotonic()の差です。
スキーマの検査は、requiredとproperties[*].typeの2つだけを見れば十分です。Pythonではboolがintのサブタイプなので、integerの位置にtrueが入ってくる場合は、別に防ぐ必要があります。
実行時間の上限は、signal.signal(signal.SIGALRM, ...)とsignal.alarm(초)(プレースホルダーは秒数です)でかけ、終わったらsignal.alarm(0)で解除します。サーバーはシングルスレッドなので、この方式が最も単純です。
ログ通知はレスポンスではなく通知です。idを入れないでください。レベルは、debug・info・notice・warning・error・critical・alert・emergencyのうちの1つです。
採点ツールは、破壊的な呼び出しを、学生のDBの一時コピー(MCP_DB)と一時的な監査ログ(MCP_AUDIT_LOG)で試します。ステップ8だけは、学生の実際のaudit.jsonlを見ます。
よくある間違い1: タイムアウトになった呼び出しを、監査ログに残さないことです。失敗した呼び出しこそ残さなければなりません。
よくある間違い2: スキーマ検証の失敗をisError: trueで返してしまうことです。引数が間違っているリクエストは、実行される前に拒否されるプロトコルエラーです。

店のDBを作る

シードSQL(/root/mcp/audit/seed.sql)を保存し、DB(/root/mcp/audit/shop.db)へ読み込んでください。customersが5行、ordersが8行になっている必要があります。

sqlite3では、sqlite3 shop.db < seed.sqlでファイルをまるごと実行します。Pythonで行うなら、sqlite3.connect(...).executescript(open(...).read())です。すでにあるDBに再度読み込むと、テーブルが存在するというエラーになるので、先に削除してください。

基本のサーバーを立てる

サーバー(/root/mcp/audit/server.py)を作成してください。initializeに返答し、tools/listにlist_customers・count_orders・delete_order(confirmが必要)の3つのツールを返します。DBのパスはMCP_DB(デフォルトは/root/mcp/audit/shop.db)です。

前のラボのv3から、許可リストだけを取り除けば足ります(このラボでは3つのツールをすべて有効にします)。delete_orderは、confirmがtrueでなければ削除しません。

すべての呼び出しを記録する

すべてのtools/callを、1行のJSONとして、ファイル(/root/mcp/audit/audit.jsonl)に残してください。パスは環境変数MCP_AUDIT_LOGで変更できるようにします。キーはts、tool、arguments、ok、duration_msで、ツールがisErrorを返したときは、okはfalseです。

ツールを実行する関数を1つ包めば足ります。開始時刻を測り、結果のisErrorでokを決め、1行をappendします。1行の形は{"ts": "2026-01-01T00:00:00+00:00", "tool": "get_weather", "arguments": {"location": "Seoul"}, "ok": true, "duration_ms": 12, "error": null}です。採点ツールは、MCP_AUDIT_LOGに一時的なパスを渡し、呼び出しを2回(正常1回、未知の状態値1回)行ったあとに、2行あるかどうかを確認します。

実行前に引数を検査する

ツールを実行する前に、inputSchemaで引数を検査してください。requiredが欠けていたり、typeが合っていなかったりする場合(例: statusに数値、idに文字列)は、実行せずにJSON-RPCエラー-32602で返答します。

スキーマのrequiredの一覧と、propertiesのtypeを、Pythonの型に対応付ける小さな関数で足ります(string→str、integer→int、boolean→bool)。検査の失敗は、ツールが実行される前なので、プロトコルエラー(-32602、Invalid params)です。

ツールの実行時間に上限をかける

ツールslow_report(引数はseconds、integer)を追加し、ツール1つの実行時間を2秒に制限してください(環境変数MCP_TOOL_TIMEOUT、デフォルトは2)。超えたときは、isError: trueとtimeoutを含むテキストで返答し、監査ログにok: falseとして残ります。seconds: 1は正常に終わる必要があります。

signal.alarm(TOOL_TIMEOUT)をかけ、SIGALRMのハンドラーで例外を投げれば、time.sleepの最中でも抜け出せます。終わったらsignal.alarm(0)です。採点ツールはseconds: 6を送り、5秒以内にisErrorのレスポンスが返ってくるかを測ります。

ログをプロトコルで送る

initializeのcapabilitiesにloggingを宣言し、ツール呼び出しごとに、レスポンスの前にnotifications/message通知(levelはRFC 5424の段階のうちの1つ、logger、data)をstdoutに出力してください。失敗した呼び出しはerrorのレベルです。

通知は{"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"...","data":{...}}}で、idはありません。レスポンスを書く前に、1行先に書けば足ります。stderrのログと違って、これはクライアントが構造化された形で受け取るログです。

削除するツールは回数を制限する

delete_orderは、1セッションにつき3回までだけ許可してください。4回目の呼び出しは削除せず、isError: trueとrate limitを含むテキストで返答します。

セッションはプロセス1つなので、モジュールのグローバルな辞書でツールごとの呼び出し数を数えれば足ります。採点ツールは、一時コピーのDBにconfirm=trueで4回呼び出し、行がちょうど3つだけ減ったかどうかを確認します。

監査ログを読んで集計する

集計スクリプト(/root/mcp/audit/summary.py)を作成してください。audit.jsonlを読み、ツールごとの{"calls": n, "failed": m}を、ファイル(/root/mcp/audit/summary.json)に書き込みます。環境変数MCP_SUMMARY_OUTがあれば、そのパスに書きます。実際のaudit.jsonlに、呼び出しの記録が5行以上たまっている必要があります。

1行ずつjson.loadsしてtoolごとに数え、okが偽ならfailedを加算します。最初の引数でログのパスを受け取れるようにすれば、採点ツールが一時的なログで検算できます。前のステップまでを実行していれば、audit.jsonlにはすでに複数の行があります。