はじめに
本記事では、MCPを利用して、LLM/エージェントが実行したツール呼び出しを監査ログとして記録する「MCP監査ログサーバ」をPythonで実装します。FastMCPと stdio を使い、ツールが呼ばれたタイミングで JSONL 形式のログをファイルに残すところまでを扱います。
MCP監査ログサーバを開発しようと考えたきっかけ
エージェントが増えると業務効率は上がる一方、以下のような問題が発生します。
- 許可されていないエージェントが動く
- いつ、だれが、どんなツールを使ったかわからない
- 事故が起きても「証跡」は残らない
このような問題に対して、まず取り組むべきはブロックや検知ではなく、「ログをとること」です。誰が・いつ・どのツールを呼んだかを記録する土台があってはじめて、のちの検知や分析が可能になります。その第一歩として、本記事ではMCPサーバとして動作する監査ログ基盤を構築します。
MCP(Model Context Protocol)とは
MCPとは、LLMが外部ツールや外部のデータソースにアクセスするための標準的な仕組みです。LLMやエージェントは、MCPの決まりに沿ってツールを呼び出すことで、実装ごとの差を吸収できます。従来は「LLMにツールを使わせる仕組み」がばらばらになりがちでしたが、MCPにより「ツールを呼び出すための共通インターフェース」が整備されつつあります。
MCPクライアントとMCPサーバ
MCPは「MCPサーバ」と「MCPクライアント」の2つで構成されます。
MCPサーバは、LLM/エージェントから呼び出されるツール（関数）を提供します。
MCPクライアントは、MCPサーバに対して tools/list や tools/call を送信し、ツールを実行します。
本記事では stdio 方式を利用するため、HTTPのようにポートを公開せず、標準入力/標準出力を通じて両者が通信します。まず役割、次に接続から実行までの流れで整理します。
役割の関係
次の図は、MCPにおける両者の役割の関係を示しています。左がツールを呼び出す側の MCPクライアント、右がツールを実行して返す側の MCPサーバ であり、その間はポートを使わず、標準入出力（stdio）で双方向に通信します。
接続から実行までの流れ
次の図は、接続から実行までの流れを示しています。クライアントとサーバの間では、いずれも標準入出力（stdio）上で、まず接続（initialize）とその応答、次にツール一覧の要求（tools/list）と一覧の返却、そしてツール実行の要求（例: ping の tools/call）と実行結果の返却、という順にやりとりが行われます。図中の括弧内は、プロトコル上のメソッド名や呼び出し例です。
FastMCPとは
PythonでMCPサーバを実装するためのフレームワークです。
MCPサーバをゼロから実装するのではなく、FastMCPを使うことで、ツール定義やリクエスト処理をシンプルに記述することができます。
開発環境
- OS: Windows 11 + WSL2
- Python 3.12.3
実装の準備
ディレクトリ構成
.
├── server.py
├── client.py
├── audit_log.py
└── audit.log.jsonl
-
server.py
- MCPサーバのエントリポイント
- FastMCPでツールを定義し、
stdioで待ち受ける
-
client.py
- 検証用の MCP クライアント
- サーバを子プロセスで起動し、
initialize/tools/list/tools/callを送って
動作確認する
-
audit_log.py
- 監査ログを JSONL 形式でファイルに追記するモジュール
-
write_audit_log()をツール実行時に呼ぶ
-
audit.log.jsonl
- 監査ログの出力先ファイル
- ツール実行のたびに追記され、クライアント実行後に初めて生成される
仮想環境(venv)を作成する
python3 -m venv .venv && source .venv/bin/activate
依存関係をインストールする
今回はMCPサーバを実装するために fastmcp を利用します。また、入力やログの形式を崩さないために pydantic を導入します。加えて、最低限の品質担保としてテスト用に pytest、静的解析用に ruff を導入します。
※今回のコードで pydantic は使いませんが、のちの拡張でリクエスト/ログの型を扱うときに使う想定です。
pip install fastmcp pydantic ruff pytest
MCPサーバを起動する
ここでは、監査ログを実装する前提として、FastMCPサーバが起動できることを確認します。
この時点では監査ログ機能はまだありません。
server.py を作成する
このファイルでは、FastMCPで ping という1つのツールだけを定義し、stdio で待ち受けるようにします。
touch server.py
from fastmcp import FastMCP
mcp = FastMCP("mcp-audit-log")
@mcp.tool
def ping() -> str:
return "pong"
if __name__ == "__main__":
mcp.run()
起動確認
MCPサーバを起動します。
python server.py
このログが表示されれば、FastMCPサーバが起動し、stdio で待ち受けている状態になっています。
stdio とは、HTTPのようにポートを開くのではなく、標準入力/標準出力を通じてMCPクライアントと通信する方式です。
ただし、この時点ではツールが実際に呼び出されたわけではないため、ping ツールが正しく動作するかどうかはまだ確認できません。
次のセクションでは監査ログを実装し、ツール呼び出しが発生したことをログから確認します。
監査ログを記録する
ここでは、本題である「監査ログ」を実装します。
ツール呼び出しが実行されたタイミングで、監査ログをJSONL形式でファイルに追記します。
JSONL形式とは「1行に1つのJSONを出力するログ形式」です。
追記が容易で、後から検索・集計もしやすいため、監査ログ用途に向いています。
ログ出力用のモジュールを作成する
audit_log.py を作成します。
touch audit_log.py
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict
LOG_PATH = Path("audit.log.jsonl")
def write_audit_log(event: Dict[str, Any]) -> None:
"""
監査ログをJSONL形式で追記する。
"""
record = {
"ts": datetime.now(timezone.utc).isoformat(),
**event,
}
with LOG_PATH.open("a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
ツール実行時に監査ログを書き込む処理を追加する
ここまででMCPサーバ自体は起動することができましたが、この状態では「いつツールが呼ばれたか」「どのツールが実行されたか」といった証跡が残りません。
そこで、ツール関数の先頭で write_audit_log() を呼び出し、ツール実行のタイミングで監査ログを追記します。
今回は以下をログとして残します。
- 実行時刻(UTC)
- ツール名
- 実行結果（成功/ 失敗の種別）
ログ形式はJSONLとし、後から集計・検索しやすい形にします。
上記の方針に沿って、server.py を修正します。
from fastmcp import FastMCP
from audit_log import write_audit_log
mcp = FastMCP("mcp-audit-log")
@mcp.tool
def ping() -> str:
write_audit_log(
{
"tool": "ping",
"status": "ok",
}
)
return "pong"
if __name__ == "__main__":
mcp.run()
stdioクライアントでツールを実行し、ログ出力を確認する
本記事では、stdio 方式のMCPサーバを「クライアントが子プロセスとして起動する」構成で動作確認します。
そのため、サーバは別ターミナルで起動せず、クライアントを実行することでサーバも同時に起動します。
また、stdio 方式ではサーバを起動しただけではツールは実行されません。
ツールを実行するには、MCPクライアントがJSON-RPC で tools/call を送る必要があります。
そこで、stdioクライアントを作成し、ping を呼び出して監査ログが出力されることを確認します。
client.py を作成します。
touch client.py
import asyncio
import json
import subprocess
from typing import Any
def send(req: dict[str, Any], proc: subprocess.Popen) -> None:
proc.stdin.write((json.dumps(req) + "\n").encode("utf-8"))
proc.stdin.flush()
def recv(proc: subprocess.Popen) -> dict[str, Any]:
line = proc.stdout.readline()
if not line:
raise RuntimeError("server closed stdout")
return json.loads(line.decode("utf-8"))
async def main() -> None:
# MCPサーバを subprocess として起動する
proc = subprocess.Popen(
["python", "-m", "server"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=False,
)
try:
# 1) initialize
send(
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "mcp-audit-log-client", "version": "0.1.0"},
},
},
proc,
)
print("initialize response:", recv(proc))
# 2) initialized
send(
{
"jsonrpc": "2.0",
"method": "initialized",
"params": {},
},
proc,
)
# 3) tools/list
send(
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {},
},
proc,
)
print("tools/list response:", recv(proc))
# 4) tools/call ping
send(
{
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "ping",
"arguments": {},
},
},
proc,
)
print("tools/call response:", recv(proc))
finally:
proc.terminate()
if __name__ == "__main__":
asyncio.run(main())
stdio クライアントを実行する
stdio 方式のMCPサーバは、HTTPサーバのようにブラウザやcurlで呼び出すことができません。ツールを実行するには、MCPクライアントからJSON-RPCで tools/call を送る必要があります。
今回は動作確認を簡単にするため、クライアントがMCPサーバを子プロセスとして起動し、そのまま ping を呼び出します。
これにより「ツールが呼ばれたこと」と「監査ログが出力されたこと」を一度に確認できます。
python client.py
監査ログが出力されることを確認する
ツールが呼ばれると、audit.log.jsonl が作成されます。
ls -la audit.log.jsonl
cat audit.log.jsonl
ログは以下のような形式で出力されます。
{"ts": "2026-02-06T19:26:11.015087+00:00", "tool": "ping", "status": "ok"}
{"ts": "2026-02-06T19:27:33.238063+00:00", "tool": "ping", "status": "ok"}
まとめ
本記事では、FastMCPを使ってMCPサーバを起動し、ツール呼び出しを監査ログとして記録する実装をしました。
また、stdio 方式ではサーバを起動しただけではツールは実行されず、MCPクライアントからJSON-RPCで tools/call を送る必要がある点も確認しました。
現時点では ping ツールのみですが、次回以降は監査ログとして実用レベルにするため、以下を拡張します。
- ツール呼び出しの入力（arguments）をログに含める
- 失敗時のエラー内容をログに含める
- 改ざん検知を導入する