G-gen の勝島です。当記事では、Gemini Enterprise Agent Platform と Cloud Monitoring の MCP サーバーを組み合わせて、エラーログの検知から AI による原因分析、Slack 通知までを自動化します。 はじめに Gemini Enterprise Agent Platform とは MCP(Model Context Protocol)とは 当記事について 背景と構成 本構成の狙い システムの構成 環境構築 環境変数の設定 API の有効化 ログルーティングの設定 サービスアカウントと IAM ロール アプリケーションの実装 ディレクトリ構成 requirements.txt main.py ソースコードの解説 Cloud Run functions へのデプロイ デプロイコマンドの実行 Cloud Run の呼び出し権限の付与 動作確認 はじめに Gemini Enterprise Agent Platform とは 2026年4月現在、 Gemini Enterprise Agent Platform (以下、Agent Platform)は、Google Cloud が提供する AI エージェントの構築・運用のための統合プラットフォーム(プロダクト群)です。 同年4月の Google Cloud Next '26 で、従来の Vertex AI から名称変更されました。Agent Platform は、エージェントの開発、スケール、ガバナンス、最適化のためのプロダクト群であるといえます。 MCP(Model Context Protocol)とは Model Context Protocol (以下、 MCP )は、AI モデルが外部ツールを呼び出すための標準プロトコルです。 Google Cloud では、Cloud Monitoring や Cloud Logging などの主要サービス向けに、フルマネージドなリモート MCP サーバーである Google Cloud MCP Servers が提供されています。当記事の構成では、AI モデルがこの MCP サーバー経由で Cloud Monitoring のメトリクスをツールとして自律的に呼び出し、原因分析に使用します。 参考 : Google Cloud MCP servers overview 参考 : MCP Reference: monitoring.googleapis.com Google Cloud MCP Servers の概要や認証方式の詳細については、以下の記事も参照してください。 blog.g-gen.co.jp 当記事について 当記事では、Cloud Logging で severity >= ERROR のログを検知した際に、Gemini モデルが MCP サーバー経由で関連メトリクスを取得し、Cloud Logging の関連ログも横断的に検索したうえで、原因の仮説と対処アクションを Slack に通知する AI エージェントを構築します。 なお、当記事の構成で使用する Cloud Run functions の全体像については以下の記事も参照してください。 blog.g-gen.co.jp 背景と構成 本構成の狙い Cloud Monitoring の標準の アラート 機能でも、しきい値ベースでの通知や Error Reporting によるエラー集計は可能です。しかし、これらは「何が起きたか」を伝えてくれるものの、「なぜ起きたのか」「どう対処すべきか」までは教えてくれません。 エラー発生時にメトリクスとログを横断的に確認し、根本原因の仮説を立てるという作業は、依然としてエンジニアの手作業に依存しています。当構成では、この一次切り分けの作業を AI エージェントに委譲することで、対応のリードタイム短縮を狙います。 システムの構成 ユーザー側で severity >= ERROR のログが Cloud Logging に書き込まれると、ログシンクを経由して Pub/Sub にメッセージが転送されます。 Pub/Sub のメッセージを受信した Cloud Run functions は、Agent Platform 経由で Gemini モデルを呼び出します。 Gemini モデルは MCP サーバー経由で Cloud Monitoring からメトリクスを取得し、さらに Cloud Logging から関連ログを検索しながら、原因の仮説と対処アクションを生成し Slack へ通知します。 構成図 なお、Pub/Sub を中心とした疎結合アーキテクチャの考え方や、Cloud Logging のログルーター(シンク)の仕組みの詳細については、以下の記事を参照してください。 blog.g-gen.co.jp blog.g-gen.co.jp 環境構築 環境変数の設定 以降のコマンドで使用する環境変数を設定します。 PROJECT_ID と SLACK_WEBHOOK_URL は、実際の環境に合わせて変更してください。 export PROJECT_ID = " your-project-id " export REGION = " asia-northeast1 " export TOPIC_NAME = " error-alerts-topic " export SINK_NAME = " error-logs-sink " export SA_NAME = " ai-ops-agent-sa " export SA_EMAIL = " ${SA_NAME} @ ${PROJECT_ID} .iam.gserviceaccount.com " export SLACK_WEBHOOK_URL = " https://hooks.slack.com/services/xxxx/xxxx/xxxx " API の有効化 対象プロジェクトをセットし、必要な API を有効化します。 gcloud config set project $PROJECT_ID gcloud services enable \ logging.googleapis.com \ pubsub.googleapis.com \ cloudfunctions.googleapis.com \ run.googleapis.com \ aiplatform.googleapis.com \ monitoring.googleapis.com \ cloudbuild.googleapis.com \ eventarc.googleapis.com 主要な API の役割は以下の通りです。 API 役割 logging.googleapis.com エラーログを検知し、Pub/Sub へルーティングする pubsub.googleapis.com エラーログを受け取り、Cloud Run functions に通知する cloudfunctions.googleapis.com 通知をトリガーに分析処理を実行する aiplatform.googleapis.com Gemini モデルでエラーの原因分析を行う monitoring.googleapis.com MCP サーバー経由でメトリクスを参照する ログルーティングの設定 Cloud Logging から Pub/Sub へエラーログを転送するための ログシンク と Pub/Sub トピック を作成します。 # Pub/Sub トピックの作成 gcloud pubsub topics create $TOPIC_NAME # ログシンクの作成(テスト用ログのみ転送) gcloud logging sinks create $SINK_NAME \ pubsub.googleapis.com/projects/ $PROJECT_ID /topics/ $TOPIC_NAME \ --log-filter =" severity>=ERROR AND logName= \" projects/ ${PROJECT_ID} /logs/my-test-log \" " --log-filter で logName を my-test-log に限定することで、動作確認用のログだけを Pub/Sub に転送する構成にしています。本番運用ではこのフィルタを実際のサービスログに合わせて変更してください。 続いて、ログシンクが Pub/Sub に書き込めるよう、ログシンクのサービスアカウントに Publisher 権限を付与します。 SINK_SA = $( gcloud logging sinks describe $SINK_NAME --format =' value(writerIdentity) ' ) gcloud pubsub topics add-iam-policy-binding $TOPIC_NAME \ --member = $SINK_SA \ --role = roles/pubsub.publisher サービスアカウントと IAM ロール Cloud Run functions が Agent Platform、Cloud Monitoring、MCP を使用するためのサービスアカウントを作成し、必要なロールを付与します。 # サービスアカウントの作成 gcloud iam service-accounts create $SA_NAME \ --display-name =" AI Ops Agent " || true # Agent Platform ユーザー gcloud projects add-iam-policy-binding $PROJECT_ID \ --member =" serviceAccount: ${SA_EMAIL} " \ --role =" roles/aiplatform.user " # Monitoring 閲覧者 gcloud projects add-iam-policy-binding $PROJECT_ID \ --member =" serviceAccount: ${SA_EMAIL} " \ --role =" roles/monitoring.viewer " # ログ閲覧者 gcloud projects add-iam-policy-binding $PROJECT_ID \ --member =" serviceAccount: ${SA_EMAIL} " \ --role =" roles/logging.viewer " # MCP ツールユーザー gcloud projects add-iam-policy-binding $PROJECT_ID \ --member =" serviceAccount: ${SA_EMAIL} " \ --role =" roles/mcp.toolUser " 各ロールの目的は以下の通りです。 ロール 目的 Agent Platform ユーザー( roles/aiplatform.user ) Gemini モデルの呼び出し Monitoring 閲覧者( roles/monitoring.viewer ) MCP 経由でのメトリクス取得 ログ閲覧者( roles/logging.viewer ) 関連ログの参照 MCP ツールユーザー( roles/mcp.toolUser ) MCP ツールの呼び出し なお、Google Cloud の MCP サーバーは「MCP プロトコル自体を呼び出す権限( roles/mcp.toolUser )」と「対象サービスのデータを参照する権限( roles/monitoring.viewer など)」の二段階の認可で保護されています。両方を付与する必要がある点に注意してください。 アプリケーションの実装 ディレクトリ構成 以下の構成でファイルを作成します。 ai-ops-function(任意のフォルダ名) ├── main.py └── requirements.txt requirements.txt 必要なライブラリを定義します。 functions-framework==3.* google-cloud-pubsub google-cloud-logging google-genai google-auth requests main.py main.py は、Pub/Sub から受け取ったエラーログを Gemini に解析させ、Slack に通知するアプリケーション本体です。 import base64 import json import os from datetime import datetime, timedelta, timezone import requests from google import genai from google.cloud import logging_v2 import google.auth from google.auth.transport.requests import Request SLACK_WEBHOOK_URL = os.environ.get( "SLACK_WEBHOOK_URL" ) PROJECT_ID = os.environ.get( "PROJECT_ID" ) MCP_SERVER_URL = "https://monitoring.googleapis.com/mcp" def get_mcp_headers (): scopes = [ "https://www.googleapis.com/auth/cloud-platform" ] credentials, _ = google.auth.default(scopes=scopes) credentials.refresh(Request()) return { "Authorization" : f "Bearer {credentials.token}" , "Content-Type" : "application/json" } def list_monitoring_mcp_tools () -> str : """MCP サーバーから使用可能なツール一覧を取得する""" payload = { "jsonrpc" : "2.0" , "method" : "tools/list" , "id" : 1 } res = requests.post(MCP_SERVER_URL, json=payload, headers=get_mcp_headers()) if not res.ok: return f "MCP tools/list API エラー (HTTP {res.status_code}): {res.text[:1000]}" result_data = res.json().get( "result" , {}) simplified_tools = [] for tool in result_data.get( "tools" , []): schema = tool.get( "inputSchema" , {}) simplified_props = {} for k, v in schema.get( "properties" , {}).items(): simplified_props[k] = { "type" : v.get( "type" , "unknown" ), "description" : v.get( "description" , "" )[: 100 ] } simplified_tools.append({ "name" : tool.get( "name" ), "description" : tool.get( "description" , "" )[: 200 ], "required_args" : schema.get( "required" , []), "properties" : simplified_props }) return json.dumps({ "tools" : simplified_tools}, indent= 2 , ensure_ascii= False ) def call_monitoring_mcp_tool (tool_name: str , arguments_json_str: str ) -> str : """指定した MCP ツールを実行してメトリクスを取得する""" arguments = json.loads(arguments_json_str) payload = { "jsonrpc" : "2.0" , "method" : "tools/call" , "id" : 2 , "params" : { "name" : tool_name, "arguments" : arguments} } res = requests.post(MCP_SERVER_URL, json=payload, headers=get_mcp_headers()) if not res.ok: return f "MCP tools/call API エラー (HTTP {res.status_code}): {res.text[:1000]}" response_data = res.json() if "result" in response_data and "content" in response_data[ "result" ]: text_result = " \n " .join( [item.get( "text" , "" ) for item in response_data[ "result" ][ "content" ]] ) return text_result[: 3000 ] + " \n ...(省略)" if len (text_result) > 3000 else text_result return f "MCP エラー: {response_data.get('error', '不明なレスポンス')}" def search_cloud_logs (filter_str: str , hours: int = 2 ) -> str : """Cloud Logging で過去 N 時間のログを検索する""" client = logging_v2.Client(project=PROJECT_ID) start_time = datetime.now(timezone.utc) - timedelta(hours=hours) full_filter = f '{filter_str} AND timestamp>="{start_time.isoformat()}"' entries = client.list_entries(filter_=full_filter, max_results= 20 ) results = [] for entry in entries: results.append({ "timestamp" : entry.timestamp.isoformat() if entry.timestamp else "" , "severity" : str (entry.severity), "resource" : entry.resource.type if entry.resource else "" , "payload" : str (entry.payload)[: 500 ] }) if not results: return "該当するログは見つかりませんでした。" text = json.dumps(results, indent= 2 , ensure_ascii= False ) return text[: 3000 ] + " \n ...(省略)" if len (text) > 3000 else text def analyze_error (event, context): """Pub/Sub からエラーログを受け取り、Gemini で分析して Slack に通知する""" pubsub_message = base64.b64decode(event[ 'data' ]).decode( 'utf-8' ) log_data = json.loads(pubsub_message) error_msg = log_data.get( "textPayload" ) or log_data.get( "jsonPayload" ) client = genai.Client(vertexai= True , project=PROJECT_ID, location= "us-central1" ) log_str = json.dumps(log_data, indent= 2 )[: 5000 ] prompt = f """ 以下のエラーログが検知されました。MCP サーバーおよび Cloud Logging と連携して調査してください。 【ログ内容】 {log_str} 【厳守事項】 原因分析にあたり、以下のステップを必ずすべて実行してください。推測(ハルシネーション)による回答や、一部のツール呼び出しの省略は許可されません。 1. `list_monitoring_mcp_tools` でツール一覧を確認してください。 2. `call_monitoring_mcp_tool` でプロジェクト {PROJECT_ID} の直近 10 分のメトリクスを取得してください。取得対象はログの文脈(「リクエストが処理しきれません」等)から判断し、Cloud Run のリクエスト数(例: metric.type="run.googleapis.com/request_count")など、負荷状況がわかる確実な標準メトリクスを指定してください。無効なクエリは避けてください。 3. `search_cloud_logs` で直近 10 分の関連するログを検索してください(severity>=WARNING など)。 【出力フォーマット】 分析結果は、必ず以下の Markdown 構造に厳密に従って出力してください。ツール名は記載せず、自然な日本語で記載してください。 ### 調査結果 1. **メトリクス分析:** (実際に取得したメトリクスの数値やスパイクの有無など、客観的な事実のみを記載) 2. **ログ分析:** (実際に検索した関連ログの件数や内容など、客観的な事実のみを記載) ### 原因の仮説 (上記の客観的データに基づき、なぜエラーが発生したのかの考察を記載) ### 対処アクション (具体的な解決策を記載) """ res = client.models.generate_content( model= 'gemini-2.5-flash' , contents=prompt, config={ "tools" : [list_monitoring_mcp_tools, call_monitoring_mcp_tool, search_cloud_logs]} ) requests.post(SLACK_WEBHOOK_URL, json={ "text" : f "🚨 *【AI 自動分析】* \n *ログ:* \n ``` \n {str(error_msg)[:1000]} \n ``` \n *分析:* \n {res.text}" }) ソースコードの解説 上記のソースコードは、大きく分けて「Gemini に渡すツール関数群」と「Pub/Sub をトリガーにエージェントを起動するエントリーポイント」の2つのパートで構成されます。 まずは、ツール関数群について解説します。このパートは、 list_monitoring_mcp_tools 、 call_monitoring_mcp_tool 、 search_cloud_logs で構成されます。 list_monitoring_mcp_tools MCP サーバーから使用可能なツール一覧を取得します。Cloud Monitoring が返すスキーマは大きく、そのまま Gemini に渡すとコンテキスト上限を超える恐れがあるため、プロパティ情報を必要最小限に絞り込んでいます。 call_monitoring_mcp_tool 指定された MCP ツールを実行し、メトリクスを取得します。 search_cloud_logs Cloud Logging から関連ログを検索します。Gemini がメトリクスだけでは原因を判断できない場合に、追加の調査手段として呼び出されます。 次に、エントリーポイント( analyze_error() )のパートについて解説します。 Pub/Sub イベントの受信 Pub/Sub から渡されたメッセージをデコードし、含まれるエラーログの内容を取り出します。 Gemini モデルの呼び出し プロンプトと使用可能ツールの一覧を渡して generate_content を実行します。プロンプトには、ツールの呼び出し順序と、最終的に出力すべき内容(原因の仮説と対処アクション)を明記しています。 Slack への通知 Gemini から返された応答を、Slack Webhook 経由で指定チャンネルに POST します。 Cloud Run functions へのデプロイ デプロイコマンドの実行 ターミナルで ai-ops-function ディレクトリに移動し、Cloud Run functions にデプロイします。 # プロジェクト番号の取得 export PROJECT_NUMBER = $( gcloud projects describe $PROJECT_ID --format =' value(projectNumber) ' ) # デプロイの実行(クリーン版) gcloud functions deploy ai-ops-analyzer \ --gen2 \ --runtime = python311 \ --region = $REGION \ --source = . \ --entry-point = analyze_error \ --trigger-topic = $TOPIC_NAME \ --service-account = $SA_EMAIL \ --trigger-service-account = ${PROJECT_NUMBER} -compute@developer.gserviceaccount.com \ --set-env-vars = SLACK_WEBHOOK_URL = $SLACK_WEBHOOK_URL , PROJECT_ID = $PROJECT_ID \ --quiet Cloud Run の呼び出し権限の付与 Cloud Run functions は、内部的には Cloud Run service として展開されます。Pub/Sub 経由でのトリガー時に正しく認証が通るよう、サービスアカウントに Cloud Run の呼び出し権限を付与します。 # プロジェクト番号の取得 export PROJECT_NUMBER = $( gcloud projects describe $PROJECT_ID --format =' value(projectNumber) ' ) # Compute Engine デフォルトサービスアカウントに Cloud Run 起動権限を付与 gcloud run services add-iam-policy-binding ai-ops-analyzer \ --region = $REGION \ --member =" serviceAccount: ${PROJECT_NUMBER} -compute@developer.gserviceaccount.com " \ --role =" roles/run.invoker " 動作確認 デプロイしたエージェントの動作確認として、以下の手順で疑似的なインシデント状況を作り出してテストします。意図的にメトリクスの負荷スパイクを発生させ、テスト用エラーログを書き込むことで、AI による原因分析が正しく実行されるかを確認します。 1.Cloud Shell から、デプロイした関数に対してリクエストを送り、メトリクス上にスパイクを発生させます。 # 自分の関数の URL を取得 URL = $( gcloud run services describe ai-ops-analyzer --region = $REGION --format =' value(status.url) ' ) TOKEN = $( gcloud auth print-identity-token ) # 1 分間、並列でリクエストを送り続ける(スパイクを作成) echo " 負荷を発生させています(約 1 分間)... " for i in { 1 .. 100 } ; do curl -s -H " Authorization: Bearer $TOKEN " $URL > /dev/null & done sleep 30 for i in { 1 .. 100 } ; do curl -s -H " Authorization: Bearer $TOKEN " $URL > /dev/null & done wait echo " 負荷生成完了。 " 2.負荷をかけてから2〜3分のタイミングで、テスト用のエラーログを書き込みます。 gcloud logging write my-test-log \ " CRITICAL: サービス応答遅延が発生しています。リクエストが処理しきれません。 " \ --severity = ERROR 3.Slack 上で、AI による分析結果が通知されることを確認します。 勝島 祐太郎 (記事一覧) クラウドソリューション部 ソリューションアーキテクト課 2025年1月G-genにジョイン!飲食業界からIT業界に転身したエンジニア。 コーヒーが好きです。