MCP のツール出力をモデルに渡す前に正規化する — PostToolUse フックパターン

from datetime import datetime, timezone
import re

def post_tool_use_hook(tool_name: str, tool_output: dict) -> dict:
    """MCP ツール出力のタイムスタンプを ISO 8601 に正規化する"""
    return normalize_timestamps(tool_output)

def normalize_timestamp(value: str) -> str:
    """個別の値を ISO 8601 に変換する"""
    # Unix epoch (例: "1718409600", "1718409600.123")
    if re.fullmatch(r"\d{10}(\.\d+)?", value):
        dt = datetime.fromtimestamp(float(value), tz=timezone.utc)
        return dt.isoformat()

    # MM/DD/YYYY (例: "06/15/2026")
    try:
        dt = datetime.strptime(value, "%m/%d/%Y").replace(tzinfo=timezone.utc)
        return dt.isoformat()
    except ValueError:
        pass

    # ISO 8601 はそのまま通す (例: "2026-06-15T09:00:00Z")
    try:
        datetime.fromisoformat(value.replace("Z", "+00:00"))
        return value
    except ValueError:
        pass

    # 未知の形式はそのまま返す
    return value

def normalize_timestamps(obj):
    """再帰的にツール出力内の全フィールドを走査して正規化する"""
    if isinstance(obj, str):
        return normalize_timestamp(obj)
    if isinstance(obj, dict):
        return {k: normalize_timestamps(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [normalize_timestamps(item) for item in obj]
    return obj

# 正規化前: 各 MCP サーバーからの生の出力
raw_outputs = {
    "crm": {"customer": "田中太郎", "last_contact": "1718409600"},
    "tickets": {"id": "TK-1234", "created": "2026-06-15T09:00:00Z"},
    "legacy": {"order_date": "06/15/2026", "status": "shipped"},
}

# 正規化後: モデルに渡るデータ
for source, output in raw_outputs.items():
    normalized = post_tool_use_hook(source, output)
    print(f"{source}: {normalized}")

# crm:     {"customer": "田中太郎", "last_contact": "2024-06-15T00:00:00+00:00"}
# tickets: {"id": "TK-1234", "created": "2026-06-15T09:00:00Z"}
# legacy:  {"order_date": "2026-06-15T00:00:00+00:00", "status": "shipped"}