お問い合わせ
AUTOMATION NOTE — 154

Claude Batch APIでGWS監査ログを月次分析する

PUBLISHED2026.06.14
READ9 分
CATEGORYAUTOMATION

AnthropicのMessage Batches API（以下 Batch API）は、標準APIの50%のコストでリクエストを非同期処理する機能として2024年後半に正式リリースされました。この記事では、GWS（Google Workspace）のReports APIで取得した監査ログをBatch APIで月次一括分析する設計パターンを整理します。

この記事を読んだほうが良い人

  • GWSの監査ログをスプレッドシートで手動確認している情シス担当者
  • GASで簡易集計しているが、AI分析のAPIコストが心配な方
  • 月次セキュリティレポートの作成工数を削減したい方
  • Claudeを定型業務に組み込みたいが、コスト予測が難しいと感じている方

Batch APIとリアルタイムAPIの選択基準

AnthropicのAPIには大きく2種類の呼び出し方があります。

リアルタイムのMessages APIは、リクエストを送るとほぼ即座に応答を返します。対話型ツールや、結果が即座に必要な処理に向いています。

一方、Message Batches APIは非同期処理です。複数のリクエストをまとめて投入し、処理が完了したら結果を取得する設計です。Anthropic公式ドキュメントによると、ほとんどのバッチは1時間以内に完了します（最大24時間でタイムアウト）。コストはリアルタイムAPIの50%です。

月次監査ログ分析は「急がない定型処理」の典型です。毎月1回、前月分のログをまとめて送信し、翌朝に結果を確認するフローなら、Batch APIが最適な選択肢になります。リアルタイムAPIをループで呼び出し続ける構成より、コストが確定的で管理しやすい点も情シスの実務に合っています。

Batch APIが特に向いている条件は3点です。まず、処理結果が24時間以内に返れば十分な定型業務であること。次に、毎月・毎週など決まったサイクルで繰り返す処理であること。そして、APIコストを月次の確定額として管理したい状況—稟議で固定費化したい場面など—です。逆に、インシデント発生時の即時分析や、担当者が画面から手動で都度実行するアドホック分析にはBatch APIは向きません。そうした用途には引き続きリアルタイムのMessages APIを使います。

月次バッチ分析の全体フロー

GWS監査ログのBatch API分析を月次で自動化する場合、以下の4ステップが現実的です。

  1. ログ取得: 毎月1日の早朝、GASが前月分の監査ログをReports APIで取得しスプレッドシートに保存
  2. バッチ投入: PythonスクリプトがログデータをBatch APIに一括送信しバッチIDを記録
  3. 結果取得: 翌朝のGASトリガーが保存済みバッチIDで処理結果を取得
  4. 通知: 分析サマリをSlack Incoming Webhookで特定チャンネルに投稿

このフローにすると、情シス担当者は月初の朝にSlackを確認するだけで前月の監査ログ異常有無を把握できます。APIコストも月次の固定費として管理できるため、稟議・費用予測がしやすくなります。

Step 1：Reports APIでGWS監査ログを取得する

GWSのReports API（Admin SDK）では、管理者が監査ログにアクセスできます。取得できる主なアプリケーションログは次の通りです。

  • admin: 管理者操作ログ（設定変更・ユーザー管理操作）
  • login: ログイン・ログアウト、不審なログインの記録
  • drive: ファイルの共有・アクセス・削除操作

以下は、GAS（Google Apps Script）から前月分のログを取得してスプレッドシートに書き出すコードです。実行には管理者権限と、appsscript.json への https://www.googleapis.com/auth/admin.reports.audit.readonly スコープの追加が必要です。

ただし、GASには1回の実行で最大6分（Workspaceの実行コンテキストによっては最大30分）のタイムアウト制限があります。社員数が多い組織でドライブログやログインログを全件取得しようとすると、ページネーションの繰り返しで制限に達するケースがあります。取得件数が1万件を超えるようであれば、取得対象期間を週単位で分割して複数トリガーで実行するか、GASではなくGoogle Cloud Run Jobsにログ取得処理を移す構成を検討してください。

/**
 * 前月のGWS監査ログを取得してスプレッドシートに記録する
 *
 * appsscript.json の oauthScopes に以下を追加してください:
 *   "https://www.googleapis.com/auth/admin.reports.audit.readonly"
 *   "https://www.googleapis.com/auth/script.external_request"
 */
function fetchAuditLogs() {
  const now = new Date();
  // 前月の開始・終了日時を算出（RFC 3339形式）
  const startTime = new Date(now.getFullYear(), now.getMonth() - 1, 1);
  const endTime   = new Date(now.getFullYear(), now.getMonth(), 0, 23, 59, 59);

  const applications = ['admin', 'login', 'drive'];
  const token = ScriptApp.getOAuthToken();
  const ss = SpreadsheetApp.getActiveSpreadsheet();

  applications.forEach(app => {
    let pageToken = null;
    const rows = [];

    do {
      const params = {
        startTime: startTime.toISOString(),
        endTime:   endTime.toISOString(),
        maxResults: '1000',
        ...(pageToken ? { pageToken } : {})
      };
      const qs = Object.keys(params)
        .map(k => `${k}=${encodeURIComponent(params[k])}`)
        .join('&');
      const url =
        `https://admin.googleapis.com/admin/reports/v1/activity/users/all/applications/${app}?${qs}`;

      const res = UrlFetchApp.fetch(url, {
        headers: { Authorization: `Bearer ${token}` },
        muteHttpExceptions: true
      });
      const data = JSON.parse(res.getContentText());

      (data.items || []).forEach(item => {
        rows.push([
          item.id?.time   || '',
          item.actor?.email || '',
          app,
          (item.events || []).map(e => e.name).join(', ')
        ]);
      });
      pageToken = data.nextPageToken || null;
    } while (pageToken);

    // アプリ名ごとのシートに書き出す
    const sheet = ss.getSheetByName(app) || ss.insertSheet(app);
    sheet.clearContents();
    if (rows.length > 0) {
      sheet.getRange(1, 1, rows.length, 4).setValues(rows);
    }
    Logger.log(`${app}: ${rows.length} 件取得`);
  });
}

pageToken を使ったページネーション処理がポイントです。1回のAPIレスポンスで返ってくる件数は最大1,000件なので、月間ログが多い場合は繰り返し取得する必要があります。muteHttpExceptions: true にしておくと、APIエラー時にスクリプトが止まらず原因をJSONで確認できます。

Step 2：Batch APIで一括分析するPythonスクリプト

取得したログデータをPythonでBatch APIに投入します。以下は、アプリケーションごとに1リクエストを生成してバッチ送信し、翌朝に結果を回収するサンプルです。

事前に pip install anthropic が必要です。

import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

def build_batch_requests(log_data: dict) -> list:
    """
    log_data: {"admin": [{"time":..., "actor":..., "event":...}, ...], ...}
    各アプリのログを1リクエストにまとめてBatch APIのリクエストリストを作成する
    """
    requests = []
    for app_name, entries in log_data.items():
        # トークン上限を考慮して最大300件に絞る
        log_lines = "\n".join(
            f"{e.get('time', '')} | {e.get('actor', '')} | {e.get('event', '')}"
            for e in entries[:300]
        )
        prompt = (
            f"以下は {app_name} アプリの先月のGWS監査ログです。\n"
            "不審な操作・通常と異なるアクセスパターンがあれば箇条書きで列挙してください。\n"
            "問題が見当たらなければ「特記事項なし」と返してください。\n\n"
            f"---\n{log_lines}\n---"
        )
        requests.append(
            Request(
                custom_id=f"audit-{app_name}",
                params=MessageCreateParamsNonStreaming(
                    model="claude-haiku-4-5",
                    max_tokens=1024,
                    messages=[{"role": "user", "content": prompt}],
                ),
            )
        )
    return requests

def submit_batch(log_data: dict) -> str:
    """バッチを送信しバッチIDを返す"""
    client = anthropic.Anthropic()
    batch = client.messages.batches.create(
        requests=build_batch_requests(log_data)
    )
    print(f"バッチ送信完了。ID: {batch.id}")
    return batch.id

def retrieve_results(batch_id: str) -> dict:
    """バッチ処理完了後に結果を回収する（翌朝呼び出す想定）"""
    client = anthropic.Anthropic()
    batch = client.messages.batches.retrieve(batch_id)

    if batch.processing_status != "ended":
        raise RuntimeError(f"バッチがまだ処理中です: {batch.processing_status}")

    results = {}
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            results[result.custom_id] = result.result.message.content[0].text  # content[0] が TextBlock であることを前提。型確認は SDK ドキュメント参照
    return results

custom_id にアプリ名を埋め込むことで、結果回収時にどのアプリの分析結果かを識別できます。バッチIDは送信後にスプレッドシートのセルに保存しておき、翌日の別スクリプトから retrieve_results() を呼ぶ設計にします。

Anthropic APIキーは環境変数 ANTHROPIC_API_KEY から読み込む設計にしてください。Cloud Run Jobsで実行する場合はGoogle Cloud Secret Managerに保存し、ローカルの開発環境では .env ファイルから読み込む運用が標準的です。APIキーをコードに直書きすると、誤ってリポジトリに公開するリスクがあります。

バッチ送信後に返ってくるバッチIDは確実に保存しておく必要があります。Anthropicの管理コンソールから再確認できる期間は29日間ですが、GASやCloud Runの実行ログに埋もれやすいため、スプレッドシートの専用セルまたはCloud StorageのJSONファイルにバッチIDを書き出す運用を推奨します。

月次定時実行の設計パターン

実運用での月次自動化は、2段構成が現実的です。

第1段：ログ取得＋バッチ送信（毎月1日 2:00）

GASの時刻ベーストリガーで fetchAuditLogs() を実行し、スプレッドシートにログを記録します。その後、PythonスクリプトをGoogle Cloud Run Jobsまたはcronジョブとして起動してBatch APIに送信し、返却されたバッチIDをスプレッドシートのセルに書き込みます。

第2段：結果取得＋Slack通知（翌日 8:00）

GASの別トリガーがスプレッドシートからバッチIDを読み込み、retrieve_results() の結果をSlack Incoming Webhookで特定チャンネルに投稿します。SlackへのHTTPリクエストはGASの UrlFetchApp.fetch() で送れるため、追加のサーバー構成は不要です。

Slack Incoming Webhookへの通知処理は次のように書きます。Webhook URLはコードに直書きせず、GASのスクリプトプロパティに SLACK_WEBHOOK_URL として保存しておきます。

/**
 * Batch API の分析結果を Slack に通知する
 * スクリプトプロパティに SLACK_WEBHOOK_URL を事前に設定すること
 * results: { "audit-admin": "...", "audit-login": "...", ... }
 */
function postAnalysisToSlack(results) {
  const props = PropertiesService.getScriptProperties();
  const webhookUrl = props.getProperty('SLACK_WEBHOOK_URL');
  if (!webhookUrl) {
    Logger.log('スクリプトプロパティに SLACK_WEBHOOK_URL が未設定です');
    return;
  }
  const lines = Object.entries(results)
    .map(([key, summary]) => `*【${key.replace('audit-', '')}】*\n${summary}`)
    .join('\n\n');

  const payload = {
    text: `*📊 GWS 監査ログ 月次分析レポート (${new Date().toLocaleDateString('ja-JP')})*\n\n${lines}`
  };
  UrlFetchApp.fetch(webhookUrl, {
    method: 'post',
    contentType: 'application/json',
    payload: JSON.stringify(payload),
    muteHttpExceptions: true
  });
  Logger.log('Slack 通知を送信しました');
}

PropertiesService.getScriptProperties() はGASのスクリプトプロパティ管理の標準機能です。Webhook URLをここに保存しておくと、コードを変更せずに通知先のチャンネルを切り替えられます。

このフローの優れている点は、業務時間内のAPI負荷がほぼゼロになることです。月次の深夜バッチで処理を完結させるため、日中のレート制限を気にする必要がありません。

コスト削減：Batch APIとリアルタイムAPIの比較

以下はAnthropicの公式価格をもとにした、Claude Haiku 4.5使用時の単価比較です。

方式 入力単価 出力単価
リアルタイムAPI（標準） $1.00/MTok $5.00/MTok
Batch API $0.50/MTok $2.50/MTok
削減率 50% 50%

参考試算として、admin・login・drive の3アプリを月次分析（1アプリにつき1リクエスト）し、1リクエストあたり入力20,000トークン・出力3,000トークンと仮定すると以下の通りです。

方式 月次コスト（参考）
リアルタイムAPI $0.105
Batch API $0.053

月次コストの絶対額は小さく見えますが、分析頻度を週次に上げたり、対象アプリを増やすと差額はその分拡大します。また、絶対額より重要なのはコスト予測性です。リアルタイムAPIでスクリプトを都度実行する運用では、手動実行のたびにコストが積み重なり、月末の費用が読みにくくなります。Batch APIで月次バッチに集約することで、APIコストが確定値になり、稟議・予算管理がしやすくなります。

分析精度を重視する場合はClaude Sonnet 4.5（Batch価格: 入力$1.50/MTok・出力$7.50/MTok）も選択肢に入ります。Haikuより高精度な分析が期待できますが、コストは約3倍になります。

導入前に確認すべき制約

実装を始める前に、以下の点を組織ポリシーと照合してください。

  • データ送信の可否: 監査ログにはメールアドレスなどの個人情報が含まれます。外部APIへの送信が社内セキュリティポリシーで許可されているか、情報セキュリティ担当と確認してください
  • ZDR非対象: Anthropic公式仕様によると、Batch APIはZDR（Zero Data Retention）の対象外です。送信データはAnthropicの標準データ保持ポリシーに従い管理されます
  • 処理時間の保証なし: Anthropic公式では「ほとんどは1時間以内に完了するが、最大24時間でタイムアウトする」とされています。当日中に結果が必要な場面にはBatch APIは向いていません
  • 1バッチの上限: 1バッチあたり最大100,000リクエスト、256MBまでです。月次ログ分析の規模では通常この上限には達しません
  • 結果の有効期限: バッチ結果はAnthropicのサーバーに29日間保存されます。自社ストレージへのダウンロードと削除の自動化を運用設計に含めてください
  • ログのマスキング検討: 機密性の高いログ（人事関連の操作履歴など）をそのまま送信する前に、メールアドレスの匿名化や不要フィールドの除外を検討してください

実装前に前月のログ件数を確認しておくことも推奨します。Reports APIで取得できるログ件数はアプリ・組織規模によって大きく異なるため、試験的に1アプリだけ取得して件数を把握しておくのが安全です。APIタイムアウトやBatch APIの256MB上限に引っかからないかの事前見積もりは、本番稼働時のトラブルを防ぐうえで重要です。ログ件数が想定より多い場合は、Pythonスクリプト側でログを分割して複数バッチに分けて送信する方式に切り替えます。

まとめ：月次バッチ化で何が変わるか

GWS監査ログをClaude Batch APIで月次処理するパターンのポイントを整理します。

  • Reports APIで前月分のログを取得し、GASで自動化できる
  • Batch APIに一括投入することで、リアルタイムAPIより50%コストを削減できる（Anthropic公式価格より）
  • 「月初深夜にログ送信→翌朝結果確認→Slack通知」のフローで定型化でき、担当者の手作業を最小化できる
  • APIコストが月次の確定値になるため、稟議・費用計画が立てやすくなる

このパターンはすべての情シスチームに向くわけではありません。リアルタイムのインシデント検知が必要な場面や、機密性の高いログを外部APIに出せない環境では、SIEM連携や社内LLM基盤を検討してください。「月次セキュリティレビューの工数を削減し、コストも予測可能にしたい」という目的に対しては、費用対効果の高い現実的な選択肢です。

最初の1ヶ月は手動でスクリプトを動かし、Slackに届いた分析結果が実際のログと整合するかを目視で確認してから自動化への移行を判断してください。AIが「不審」と判断したイベントについては、実際のログと突き合わせて精度を確認するのが最初のステップです。誤検知や見落としのパターンが分かったらプロンプトを調整し、「深夜帯の管理者操作はアラート対象」「組織外ドメインへのファイル共有はフラグを立てる」のような自社固有のポリシーを書き込むと効果的です。自動化が安定して動き出せば、月次のセキュリティレビュー会議をSlackの通知確認から始めるフローが定着します。

GWS監査ログ分析の設計・実装支援について詳しくは、DRASENAS 公式サイトのサービスページもご覧ください。

コーポレートITのご相談はお気軽に

この記事で書いたような業務改善・自動化の設計から実装まで、DRASENASではコーポレートITの現場に寄り添った支援を行っています。 「まず相談だけ」でも大歓迎です。DRASENAS 公式サイトからお気軽にどうぞ。

CONTACT

御社の IT 部門、ここにあります。

「ITのことはあまりわからない」── そのような状態からで、まったく問題ございません。まずはお気軽にご相談ください。

無料相談を予約する → サービス詳細を見る
一社ずつ、一から。