Claude Code × Mackerel MCP でレポート自動化を実現

Claude → MCPツール呼び出し → Mackerel API → データ取得
      ← 全データ返却 ←         ←
Claude → 次のMCPツール呼び出し → Mackerel API
      ← データ返却 ←            ←
（この往復が数十〜数百回繰り返される）

Claude → Pythonスクリプト起動（1回のみ）
       ← 最終結果のJSON返却のみ

（Pythonスクリプト内部で）
Python → MCP呼び出し → Mackerel API
       ← データ取得 ←
Python → 内部処理（集計・Excel生成）

/mackerel-report

レポート生成開始: 過去7日間
期間: 2024-12-19 ~ 2024-12-25
モニター情報を取得中...
  15個の関連モニターを検出
ホスト情報を取得中...
  8個のホストを検出
アラート情報を取得中...
  42個の関連アラートを検出
アラートログを取得中...
  10/42 処理済み
  20/42 処理済み
  ...
合計 127 件のログエントリを取得
Excelファイルを生成中...
==================================================
レポート生成完了
==================================================
{
  "success": true,
  "output_path": "メトリクス監視_20241219_20241225.xlsx",
  "stats": {
    "period_start": "2024-12-19",
    "period_end": "2024-12-25",
    "warning_count": 68,
    "ok_count": 59,
    "check_targets_count": 5,
    "total_logs": 127
  }
}

---
description: devサービスのMackerelメトリクス監視Excelレポートを生成（トークン最適化版）
argument-hint: [日数]
allowed-tools:
  - Bash(python3:*)
  - Bash(pip3:*)
---

# Mackerel メトリクス監視レポート生成

devサービスのMackerel監視アラートのExcelレポートを生成します。

**このコマンドはコード実行アプローチを採用しています。**
MCP Python SDKを使用してMCPツールの呼び出しをPythonコード内で完結させ、
Claudeには最終結果のみを返すことで、トークン消費を大幅に削減します。

## 引数

ユーザーが入力したコマンド引数: `$ARGUMENTS`

- 引数がある場合: 日数として使用（例: `7`, `14`, `30`）
- 引数が空または数値でない場合: デフォルトの7日間を使用
- 期間: **今日を含まない**直近N日分（今日の00:00:00 JSTが終了境界）

## 実装手順

### ステップ1: 依存パッケージの確認

実行: `python3 -c "import openpyxl, mcp"`

失敗した場合:
- 実行: `pip3 install openpyxl mcp`

### ステップ2: レポート生成スクリプトの実行

実行: `python3 .claude/scripts/mackerel/generate_report.py $ARGUMENTS`

※ APIキーは`config.json`の`op_secret_ref`を参照して自動的に1Passwordから取得されます

### ステップ3: 結果の報告

スクリプトの標準出力（JSON形式）を確認し、以下を報告:
- ファイルの完全なパス
- 対象期間（開始日と終了日）
- 処理したアラート数
- 統計サマリー（WARNING数、OK数）
- 確認対象が存在する場合はその旨を記載

JSON出力例:
\`\`\`json
{
  "success": true,
  "output_path": "メトリクス監視_20251201_20251207.xlsx",
  "stats": {
    "period_start": "2024-12-01",
    "period_end": "2024-12-07",
    "warning_count": 15,
    "ok_count": 12,
    "check_targets_count": 2,
    "total_logs": 27
  }
}
\`\`\`

## エラーハンドリング

- スクリプトが失敗した場合（success: false）、標準エラー出力のエラーメッセージを報告
- 依存パッケージがない場合、インストールを促す

## 重要な注意事項

- APIキーは`config.json`の`op_secret_ref`から1Password経由で自動取得されます
- 生成されたファイルはカレントディレクトリに保存されます
- 期間計算はJSTタイムゾーン（UTC+9）を使用します
- CRITICAL、WARNINGとOKのアラートが含まれます
- 期間は今日を含みません（今日の00:00:00が上限境界）
- 確認対象は、同じ時刻（HH:MM）に同じモニターで3回以上発生したWARNINGです

.claude/
├── commands/
│   └── mackerel-report.md      # コマンド定義ファイル
└── scripts/
    └── mackerel/
        ├── __init__.py
        ├── config.json          # 設定ファイル（1Password参照）
        ├── mcp_client.py        # MCP汎用クライアント
        ├── mackerel_client.py   # Mackerel専用クライアント
        ├── excel_generator.py   # Excel生成モジュール
        └── generate_report.py   # メインスクリプト

{
  "op_secret_ref": "op://XXXXXXXXXXXX/credential"
}

#!/usr/bin/env python3
"""
Mackerelメトリクス監視レポート生成スクリプト（エントリポイント）

使用方法:
    python generate_report.py [日数]

引数:
    日数: レポート対象期間（省略時: 7日間）
"""

import asyncio
import sys
import json
from datetime import datetime, timezone, timedelta
from mackerel_client import MackerelClient
from excel_generator import MackerelExcelGenerator

SERVICE_NAME = "dev"
JST = timezone(timedelta(hours=9))

def parse_days(args):
    """コマンドライン引数から日数を解析"""
    if len(args) > 1:
        try:
            days = int(args[1])
            if days > 0:
                return days
        except ValueError:
            pass
    return 7  # デフォルト

async def main():
    """メイン処理"""
    # 引数解析
    days = parse_days(sys.argv)
    print(f"レポート生成開始: 過去{days}日間", file=sys.stderr)

    # 期間計算（JST）
    today = datetime.now(JST).replace(hour=0, minute=0, second=0, microsecond=0)
    start_dt = today - timedelta(days=days)
    end_dt = today
    start_ts = start_dt.timestamp()
    end_ts = end_dt.timestamp()

    print(f"期間: {start_dt.strftime('%Y-%m-%d')} ~ {(end_dt - timedelta(seconds=1)).strftime('%Y-%m-%d')}", file=sys.stderr)

    try:
        async with MackerelClient() as client:
            # モニター取得
            print("モニター情報を取得中...", file=sys.stderr)
            monitors = await client.list_monitors()

            # 対象モニターをフィルタリング
            monitor_map = {}
            monitor_ids = []
            for m in monitors:
                scopes = m.get("scopes", [])
                if any(SERVICE_NAME in s for s in scopes):
                    monitor_map[m["id"]] = m.get("name", m["id"])
                    monitor_ids.append(m["id"])

            print(f"  {len(monitor_ids)}個の関連モニターを検出", file=sys.stderr)

            # ホスト取得
            print("ホスト情報を取得中...", file=sys.stderr)
            hosts = await client.list_hosts(SERVICE_NAME)
            host_map = {}
            for h in hosts:
                host_map[h["id"]] = h.get("displayName") or h.get("name", h["id"])

            print(f"  {len(host_map)}個のホストを検出", file=sys.stderr)

            # アラート取得
            print("アラート情報を取得中...", file=sys.stderr)
            all_alerts = await client.get_all_alerts()
            relevant_alerts = [a for a in all_alerts if a.get("monitorId") in monitor_ids]

            print(f"  {len(relevant_alerts)}個の関連アラートを検出", file=sys.stderr)

            # アラートログ収集
            print("アラートログを取得中...", file=sys.stderr)
            alert_logs = []

            for i, alert in enumerate(relevant_alerts):
                logs = await client.get_alert_logs(alert["id"])
                host_id = alert.get("hostId", "")
                monitor_id = alert.get("monitorId", "")
                host_name = host_map.get(host_id, host_id)
                monitor_name = monitor_map.get(monitor_id, monitor_id)

                for log in logs:
                    status = log.get("status", "")
                    created_at = log.get("createdAt", 0)
                    # 期間フィルタリング
                    if not (start_ts <= created_at < end_ts):
                        continue

                    value = log.get("value", "")
                    alert_logs.append((created_at, status, host_name, monitor_name, value))

                if (i + 1) % 10 == 0:
                    print(f"  {i + 1}/{len(relevant_alerts)} 処理済み", file=sys.stderr)

            print(f"合計 {len(alert_logs)} 件のログエントリを取得", file=sys.stderr)

        # Excel生成
        print("Excelファイルを生成中...", file=sys.stderr)
        generator = MackerelExcelGenerator(start_dt, end_dt)
        result = generator.generate(alert_logs)

        # 結果出力（JSON形式）
        output = {
            "success": True,
            "output_path": result["output_path"],
            "stats": {
                "period_start": start_dt.strftime('%Y-%m-%d'),
                "period_end": (end_dt - timedelta(seconds=1)).strftime('%Y-%m-%d'),
                "warning_count": result["warning_count"],
                "critical_count": result["critical_count"],
                "ok_count": result["ok_count"],
                "other_count": result["other_count"],
                "check_targets_count": result["check_targets_count"],
                "total_logs": len(alert_logs),
            }
        }

        print("\n" + "="*50, file=sys.stderr)
        print("レポート生成完了", file=sys.stderr)
        print("="*50, file=sys.stderr)
        print(json.dumps(output, ensure_ascii=False, indent=2))

        return 0

    except Exception as e:
        error_output = {
            "success": False,
            "error": str(e),
        }
        print(json.dumps(error_output, ensure_ascii=False, indent=2))
        import traceback
        traceback.print_exc(file=sys.stderr)
        return 1

if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

"""
Mackerel専用クライアント
MCPClientをラップしてMackerel APIの呼び出しを簡単にする
Python実行時にMCPサーバーを起動する方式
"""

import os
import json
import subprocess
from pathlib import Path
from typing import Dict, List, Optional
from mcp_client import MCPClient

# 設定ファイルのパス
CONFIG_PATH = Path(__file__).parent / "config.json"

def load_config() -> dict:
    """設定ファイルを読み込む"""
    if CONFIG_PATH.exists():
        with open(CONFIG_PATH) as f:
            return json.load(f)
    return {}

def get_api_key_from_1password() -> Optional[str]:
    """config.jsonのop_secret_refを参照して1PasswordからAPIキーを取得"""
    config = load_config()
    op_ref = config.get("op_secret_ref")
    if not op_ref:
        return None

    try:
        result = subprocess.run(
            ["op", "read", op_ref],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass
    return None

class MackerelClient:
    """Mackerel専用クライアント"""

    def __init__(
        self,
        api_key: Optional[str] = None,
        command: str = "pnpm",
        args: Optional[List[str]] = None,
    ):
        """
        初期化

        Args:
            api_key: Mackerel APIキー（Noneの場合は環境変数または1Passwordから取得）
            command: MCPサーバー起動コマンド（デフォルト: pnpm）
            args: MCPサーバー起動引数（デフォルト: ['dlx', '@mackerel/mcp-server']）
        """
        self.api_key = (
            api_key
            or os.environ.get("MACKEREL_API_KEY")
            or os.environ.get("MACKEREL_APIKEY")
            or get_api_key_from_1password()
        )
        if not self.api_key:
            raise ValueError(
                "Mackerel APIキーが見つかりません。\n"
                "config.jsonのop_secret_refを設定するか、環境変数MACKEREL_APIKEYを設定してください。"
            )

        if args is None:
            args = ["dlx", "@mackerel/mcp-server"]

        env = {"MACKEREL_APIKEY": self.api_key}
        self.client = MCPClient(command=command, args=args, env=env)

    async def __aenter__(self):
        """非同期コンテキストマネージャー（開始）"""
        await self.client.__aenter__()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """非同期コンテキストマネージャー（終了）"""
        await self.client.__aexit__(exc_type, exc_val, exc_tb)

    async def list_monitors(self) -> List[Dict]:
        """
        モニター一覧を取得

        Returns:
            モニターのリスト
        """
        result = await self.client.call_tool("list_monitors", {})
        if isinstance(result.content, list) and len(result.content) > 0:
            data = json.loads(result.content[0].text)
            return data.get("monitors", [])
        return []

    async def list_hosts(self, service: str) -> List[Dict]:
        """
        ホスト一覧を取得

        Args:
            service: サービス名

        Returns:
            ホストのリスト
        """
        result = await self.client.call_tool("list_hosts", {"service": service})
        if isinstance(result.content, list) and len(result.content) > 0:
            data = json.loads(result.content[0].text)
            return data.get("hosts", [])
        return []

    async def list_alerts(
        self, limit: int = 100, with_closed: bool = True, next_id: Optional[str] = None
    ) -> Dict:
        """
        アラート一覧を取得

        Args:
            limit: 取得件数上限
            with_closed: クローズ済みアラートも含めるか
            next_id: ページネーション用ID

        Returns:
            アラート情報（alerts配列とnextIdを含む）
        """
        args = {"limit": limit, "withClosed": with_closed}
        if next_id:
            args["nextId"] = next_id

        result = await self.client.call_tool("list_alerts", args)
        if isinstance(result.content, list) and len(result.content) > 0:
            return json.loads(result.content[0].text)
        return {"alerts": []}

    async def get_alert_logs(self, alert_id: str) -> List[Dict]:
        """
        アラートログを取得

        Args:
            alert_id: アラートID

        Returns:
            ログのリスト
        """
        result = await self.client.call_tool("get_alert_logs", {"alertId": alert_id})
        if isinstance(result.content, list) and len(result.content) > 0:
            data = json.loads(result.content[0].text)
            return data.get("logs", [])
        return []

    async def get_all_alerts(self) -> List[Dict]:
        """
        すべてのアラートを取得（ページネーション対応）

        Returns:
            すべてのアラートのリスト
        """
        all_alerts = []
        next_id = None

        while True:
            data = await self.list_alerts(limit=100, with_closed=True, next_id=next_id)
            alerts = data.get("alerts", [])
            all_alerts.extend(alerts)

            next_id = data.get("nextId")
            if not next_id:
                break

        return all_alerts

"""
MCP SDK汎用クライアント
MCP Python SDKを使用してMCPサーバーと通信する
"""

import os
from contextlib import AsyncExitStack
from typing import Any, Dict, List, Optional
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPClient:
    """MCP SDK汎用クライアント"""

    def __init__(
        self,
        command: str,
        args: Optional[List[str]] = None,
        env: Optional[Dict[str, str]] = None,
    ):
        """
        初期化

        Args:
            command: MCPサーバーを起動するコマンド
            args: コマンドライン引数
            env: 追加の環境変数（既存の環境変数とマージされる）
        """
        merged_env = os.environ.copy()
        if env:
            merged_env.update(env)

        self.server_params = StdioServerParameters(
            command=command,
            args=args or [],
            env=merged_env,
        )
        self.session: Optional[ClientSession] = None
        self._exit_stack: Optional[AsyncExitStack] = None

    async def __aenter__(self):
        """非同期コンテキストマネージャー（開始）"""
        self._exit_stack = AsyncExitStack()
        await self._exit_stack.__aenter__()

        # stdio_clientをコンテキストスタックに追加
        read, write = await self._exit_stack.enter_async_context(
            stdio_client(self.server_params)
        )

        # ClientSessionをコンテキストスタックに追加
        self.session = await self._exit_stack.enter_async_context(
            ClientSession(read, write)
        )

        await self.session.initialize()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """非同期コンテキストマネージャー（終了）"""
        if self._exit_stack:
            await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)

    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
        """
        MCPツールを呼び出す

        Args:
            tool_name: ツール名
            arguments: ツールに渡す引数

        Returns:
            ツールの実行結果
        """
        if not self.session:
            raise RuntimeError("MCPセッションが初期化されていません")

        result = await self.session.call_tool(tool_name, arguments=arguments)
        return result

"""
Excel生成モジュール
openpyxlを使用してMackerelレポートをExcel形式で生成
"""

from datetime import datetime, timezone, timedelta
from typing import List, Tuple, Dict
from collections import defaultdict
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side

class MackerelExcelGenerator:
    """Mackerelレポート用Excel生成クラス"""

    def __init__(self, start_dt: datetime, end_dt: datetime):
        """
        初期化

        Args:
            start_dt: 期間開始日時（JSTタイムゾーン）
            end_dt: 期間終了日時（JSTタイムゾーン）
        """
        self.start_dt = start_dt
        self.end_dt = end_dt
        self.jst = timezone(timedelta(hours=9))

    def generate(self, alert_logs: List[Tuple]) -> Dict:
        """
        Excelレポートを生成

        Args:
            alert_logs: [(timestamp, status, host_name, monitor_name, value), ...]

        Returns:
            {
                "output_path": str,
                "check_targets_count": int,
                "warning_count": int,
                "critical_count": int,
                "ok_count": int,
                "other_count": int
            }
        """
        # タイムスタンプでフィルタリング・ソート
        start_ts = self.start_dt.timestamp()
        end_ts = self.end_dt.timestamp()

        filtered_logs = []
        for ts, status, host, monitor, value in alert_logs:
            if start_ts <= ts < end_ts:
                dt = datetime.fromtimestamp(ts, tz=self.jst)
                filtered_logs.append({
                    "timestamp": ts,
                    "datetime": dt,
                    "status": status,
                    "host": host,
                    "monitor": monitor,
                    "value": value,
                })

        filtered_logs.sort(key=lambda x: x["timestamp"])

        # 確認対象の特定（同モニター、異なる日で時刻が15分以内、3日以上のWARNINGのみ）
        alerts_by_monitor = defaultdict(list)
        for log in filtered_logs:
            if log["status"] == "WARNING":
                alerts_by_monitor[log["monitor"]].append(log)

        def within_15_minutes(dt1, dt2):
            """2つの時刻が15分以内かどうか（日付は無視）"""
            m1 = dt1.hour * 60 + dt1.minute
            m2 = dt2.hour * 60 + dt2.minute
            diff = abs(m1 - m2)
            return min(diff, 1440 - diff) <= 15

        check_targets = {}
        for monitor, logs in alerts_by_monitor.items():
            # 時刻でグループ化（15分以内のものを同じグループに）
            groups = []
            for log in logs:
                found_group = False
                for group in groups:
                    if within_15_minutes(group[0]["datetime"], log["datetime"]):
                        group.append(log)
                        found_group = True
                        break
                if not found_group:
                    groups.append([log])

            # 各グループで異なる日が3回以上あれば確認対象
            for group in groups:
                unique_dates = set(log["datetime"].strftime("%m/%d") for log in group)
                if len(unique_dates) >= 3:
                    representative_time = group[0]["datetime"].strftime("%H:%M")
                    check_targets[(representative_time, monitor)] = group

        # Excelワークブック作成
        wb = Workbook()
        ws = wb.active
        ws.title = "メトリクス監視"

        # スタイル定義とデータ書き込み処理...
        # （省略：実際のファイルには全て含まれます）

        # ファイル保存
        file_start = self.start_dt.strftime('%Y%m%d')
        file_end = (self.end_dt - timedelta(days=1)).strftime('%Y%m%d')
        output_path = f'メトリクス監視_{file_start}_{file_end}.xlsx'
        wb.save(output_path)

        return {
            "output_path": output_path,
            "check_targets_count": len(check_targets),
            "warning_count": warning_count,
            "critical_count": critical_count,
            "ok_count": ok_count,
            "other_count": other_count,
        }

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# MCPサーバーに接続
server_params = StdioServerParameters(
    command="npx",
    args=["-y", "@mackerel/mcp-server"],
    env={"MACKEREL_APIKEY": api_key}
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        # MCPツールを呼び出し
        result = await session.call_tool(...)