Cloud Run で Gemini と MCP を使用して自然言語を実際の Google Cloud アクションに変換する

1. はじめに

この Codelab では、Python を使用してカスタム MCP（Model Context Protocol）サーバーを構築し、Google Cloud Run にデプロイして、Gemini CLI と接続し、自然言語を使用して実際の Google Cloud Storage オペレーションを実行する方法について説明します。

アーキテクチャ フロー: Gemini CLI → Cloud Run → MCP

ターミナルを開き、次のような簡単なプロンプトを AI エージェントに入力します。

• List my GCS buckets
• Create a GCS bucket named <bucket-name>
• Tell me about the metadata of my GCS object

数秒以内に、クラウドがリッスンして実行されます。複雑なコマンドは必要ありません。タブを切り替える必要もありません。平易な言葉が実際のクラウド アクションに変わります。

演習内容

Gemini CLI を Google Cloud Storage に接続するカスタム MCP サーバーを構築してデプロイします。

次のことを行います。

• Python ベースの MCP サーバーを構築する
• アプリケーションをコンテナ化する
• Cloud Run にデプロイする
• IAM と ID トークンを使用して保護する
• Gemini CLI と接続する
• 自然言語を使用してライブ GCS オペレーションを実行する

学習内容

• MCP（Model Context Protocol）とは何か、その仕組み
• Python を使用してツール呼び出し機能を構築する方法
• コンテナ化されたアプリケーションを Cloud Run にデプロイする方法
• Gemini CLI を外部 MCP サーバーと統合する方法
• Cloud Run サービスを安全に認証する方法
• AI を使用して実際の Google Cloud Storage オペレーションを実行する方法

必要なもの

• Chrome ウェブブラウザ
• Gmail アカウント
• 課金が有効になっている Google Cloud プロジェクト
• Gemini CLI（Google Cloud Shell にプリインストールされています）
• Python と Google Cloud の基本的な知識

この Codelab では、ユーザーが Python の基本的な知識を持っていることを前提としています。

2. 始める前に

プロジェクトを作成する

1. Google Cloud コンソールのプロジェクト選択ページで、Google Cloud プロジェクトを選択または作成します。
2. Cloud プロジェクトに対して課金が有効になっていることを確認します。プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。
3. Cloud Shell（Google Cloud で動作するコマンドライン環境）を使用します。この環境には bq がプリロードされています。Google Cloud コンソールの上部にある「Cloud Shell をアクティブにする」アイコン をクリックします。

4. Cloud Shell に接続したら、次のコマンドを使用して、すでに認証済みであることと、プロジェクトがプロジェクト ID に設定されていることを確認します。

gcloud auth list

5. Cloud Shell で次のコマンドを実行して、gcloud コマンドがプロジェクトを認識していることを確認します。

gcloud config list project

6. プロジェクトが設定されていない場合は、次のコマンドを使用して設定します。

gcloud config set project <YOUR_PROJECT_ID>

7. 次のコマンドを使用して、必要な API を有効にします。これには数分かかる場合があります。

gcloud services enable \
  run.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com

承認を求められたら、[承認] をクリックして続行します。

コマンドが正常に実行されると、次のようなメッセージが表示されます。

Operation "operations/..." finished successfully.

API が見つからない場合は、実装中にいつでも有効にできます。

gcloud コマンドとその使用方法については、ドキュメントをご覧ください。

Python プロジェクトを準備する

このセクションでは、MCP サーバーをホストする Python プロジェクトを作成し、Cloud Run にデプロイするための依存関係を構成します。

プロジェクト ディレクトリを作成する

まず、ソースコードを保存する mcp-on-cloudrun という名前の新しいフォルダを作成します。

mkdir gcs-mcp-server && cd gcs-mcp-server

requirements.txt を作成する

touch requirements.txt
cloudshell edit ~/gcs-mcp-server/requirements.txt

次の内容をファイルに追加します。

fastmcp
google-cloud-storage
google-api-core
pydantic

ファイルを保存します。

3. MCP サーバーを作成する

このセクションでは、Google Cloud Storage オペレーションを呼び出し可能なツールとして公開する MCP サーバーを作成します。

このサーバーは次のようになります。

• MCP ツールを登録する
• Google Cloud Storage に接続する
• HTTP で実行する
• Cloud Run にデプロイ可能にする

次に、main.py 内にコア MCP ロジックを作成します。

以下は、バケットのリスト表示と作成から、BLOB のアップロード、ダウンロード、管理まで、Google Cloud Storage を管理するための複数のツールを定義する完全なコードです。

メイン アプリケーション ファイルを作成する

mcp-on-cloudrun ディレクトリに、main.py という名前の新しいファイルを作成します。

touch main.py

Cloud Shell エディタを使用してファイルを開きます。

cloudshell edit ~/gcs-mcp-server/main.py

main.py ファイルの内容に次のソースを追加します。

import asyncio
import logging
import os
from datetime import timedelta
from typing import List, Dict, Any
from fastmcp import FastMCP
from google.cloud import storage
from google.api_core import exceptions

# ---------------------------------------------------------
# 🌐 Initialize MCP
# ---------------------------------------------------------
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
logger = logging.getLogger(__name__)

mcp = FastMCP(name="MyEnhancedGCSMCPServer")
# ---------------------------------------------------------
# 1️⃣ Simple Greeting
# ---------------------------------------------------------
@mcp.tool
def sayhi(name: str) -> str:
  """Returns a friendly greetings"""
  return f"Hello {name}! It's a pleasure to connect from your enhanced MCP Server."

# ---------------------------------------------------------
# 2️⃣ List all GCS buckets
# ---------------------------------------------------------
@mcp.tool
def list_gcs_buckets() -> List[str]:
  """Lists all GCS buckets in the project."""
  try:
      storage_client = storage.Client()
      buckets = storage_client.list_buckets()
      return [bucket.name for bucket in buckets]
  except exceptions.Forbidden as e:
      return [f"Error: Permission denied to list buckets. Details: {e}"]
  except Exception as e:
      return [f"An unexpected error occurred: {e}"]

# ---------------------------------------------------------
# 3️⃣ Create a new bucket
# ---------------------------------------------------------
@mcp.tool
def create_bucket(bucket_name: str, location: str = "US") -> str:
  """Creates a new GCS bucket. Bucket names must be globally unique."""
  try:
      storage_client = storage.Client()
      bucket = storage_client.bucket(bucket_name)
      bucket.location = location
      storage_client.create_bucket(bucket)
      return f"✅ Bucket '{bucket_name}' created successfully in '{location}'."
  except exceptions.Conflict:
      return f"⚠️ Error: Bucket '{bucket_name}' already exists."
  except exceptions.Forbidden as e:
      return f"❌ Error: Permission denied to create bucket. Details: {e}"
  except Exception as e:
      return f"❌ Unexpected error: {e}"

# ---------------------------------------------------------
# 4️⃣ Delete a bucket
# ---------------------------------------------------------
@mcp.tool
def delete_bucket(bucket_name: str) -> str:
  """Deletes a GCS bucket."""
  try:
      storage_client = storage.Client()
      bucket = storage_client.bucket(bucket_name)
      bucket.delete(force=True)
      return f"🗑️ Bucket '{bucket_name}' deleted successfully."
  except exceptions.NotFound:
      return f"⚠️ Error: Bucket '{bucket_name}' not found."
  except exceptions.Forbidden as e:
      return f"❌ Error: Permission denied to delete bucket. Details: {e}"
  except Exception as e:
      return f"❌ Unexpected error: {e}"

# ---------------------------------------------------------
# 5️⃣ List objects in a bucket
# ---------------------------------------------------------
@mcp.tool
def list_objects(bucket_name: str) -> List[str]:
  """Lists all objects in a specified GCS bucket."""
  try:
      storage_client = storage.Client()
      blobs = storage_client.list_blobs(bucket_name)
      return [blob.name for blob in blobs]
  except exceptions.NotFound:
      return [f"⚠️ Error: Bucket '{bucket_name}' not found."]
  except Exception as e:
      return [f"❌ Unexpected error: {e}"]
# ---------------------------------------------------------
# Delete file from a bucket
# ---------------------------------------------------------
@mcp.tool
def delete_blob(bucket_name: str, blob_name: str) -> str:
  """Deletes a blob from a GCS bucket."""
  try:
      storage_client = storage.Client()
      bucket = storage_client.bucket(bucket_name)
      blob = bucket.blob(blob_name)
      blob.delete()
      return f"🗑️ Blob '{blob_name}' deleted from bucket '{bucket_name}'."
  except exceptions.NotFound:
      return f"⚠️ Error: Bucket '{bucket_name}' or blob '{blob_name}' not found."
  except exceptions.Forbidden as e:
      return f" Permission denied. Details: {e}"
  except Exception as e:
      return f" Unexpected error: {e}"

# ---------------------------------------------------------
# Get bucket metadata
# ---------------------------------------------------------
@mcp.tool
def get_bucket_metadata(bucket_name: str) -> Dict[str, Any]:
  """Retrieves metadata for a GCS bucket."""
  try:
      storage_client = storage.Client()
      bucket = storage_client.get_bucket(bucket_name)
      return {
          "id": bucket.id,
          "name": bucket.name,
          "location": bucket.location,
          "storage_class": bucket.storage_class,
          "created": bucket.time_created.isoformat() if bucket.time_created else None,
          "updated": bucket.updated.isoformat() if bucket.updated else None,
          "versioning_enabled": bucket.versioning_enabled,
      }
  except exceptions.NotFound:
      return {"error": f" Bucket '{bucket_name}' not found."}
  except Exception as e:
      return {"error": f" Unexpected error: {e}"}

# ---------------------------------------------------------
# Get object metadata
# ---------------------------------------------------------
@mcp.tool
def get_blob_metadata(bucket_name: str, blob_name: str) -> Dict[str, Any]:
  """Retrieves metadata for a specific blob."""
  try:
      storage_client = storage.Client()
      bucket = storage_client.bucket(bucket_name)
      blob = bucket.get_blob(blob_name)
      if not blob:
          return {"error": f" Blob '{blob_name}' not found in '{bucket_name}'."}
      return {
          "name": blob.name,
          "bucket": blob.bucket.name,
          "size": blob.size,
          "content_type": blob.content_type,
          "updated": blob.updated.isoformat() if blob.updated else None,
          "storage_class": blob.storage_class,
          "crc32c": blob.crc32c,
          "md5_hash": blob.md5_hash,
      }
  except exceptions.NotFound:
      return {"error": f" Bucket '{bucket_name}' not found."}
  except Exception as e:
      return {"error": f" Unexpected error: {e}"}

# ---------------------------------------------------------
# 🚀 Entry Point
# ---------------------------------------------------------
if __name__ == "__main__":
  port = int(os.getenv("PORT", 8080))
  logger.info(f"🚀 Starting Enhanced GCS MCP Server on port {port}")
  asyncio.run(
      mcp.run_async(
          transport="http",
          host="0.0.0.0",
          port=port,
      )
  )

コードを追加したら、ファイルを保存します。

プロジェクト構造は次のようになります。

gcs-mcp-server/
├── requirements.txt
└── main.py

コードについて簡単に説明します。

インポートと設定:

コードは、必要なライブラリをインポートすることから始まります。

• 標準ライブラリ: 非同期実行用の asyncio、ステータス メッセージの出力用の logging、環境変数用の os。
• FastMCP: Model Context Protocol サーバーの作成に使用されるコア フレームワーク。
• Google Cloud Storage: GCS とやり取りするために google.cloud.storage ライブラリをインポートし、エラー処理用に exceptions をインポートします。

初期化:

サーバーの ID をデバッグして追跡できるように、ロギング形式を構成します。また、MyEnhancedGCSMCPServer という名前の FastMCP のインスタンスを構成します。このオブジェクト（mcp）は、サーバーが公開するすべてのツール（関数）を登録するために使用されます。次のツールを定義します。

• list_gcs_buckets: 関連付けられた Google Cloud プロジェクト内のすべてのストレージ バケットのリストを取得します。
• create_bucket: 特定の名前とロケーションで新しいバケットを作成します。
• delete_bucket: 既存のバケットを削除します。
• list_objects: 特定のバケット内のすべてのファイル（BLOB）を一覧表示します。
• delete_blob: バケットから特定のファイルを 1 つ削除します。
• get_bucket_metadata: バケットに関する技術的な詳細（ロケーション、ストレージ クラス、バージョニング ステータス、作成時間）を返します。
• get_blob_metadata: 特定のファイルに関する技術的な詳細（サイズ、コンテンツ タイプ、MD5 ハッシュ、最終更新日）を返します。

エントリ ポイント:

これにより、ポートが構成されます。設定されていない場合は、デフォルトで 8080 になります。次に、asyncio.run() を使用して、mcp.run_async でサーバーを非同期で起動します。最後に、HTTP（host 0.0.0.0）で実行するようにサーバーを構成し、受信ネットワーク リクエストにアクセスできるようにします。

4. MCP サーバーをコンテナ化する

このセクションでは、MCP サーバーを Cloud Run にデプロイできるように Dockerfile を作成します。

Cloud Run にはコンテナ化されたアプリケーションが必要です。アプリケーションのビルド方法と起動方法を定義します。

Dockerfile を作成する

Dockerfile という名前の新しいファイルを作成します。

touch Dockerfile

Cloud Shell エディタで開きます。

cloudshell edit ~/gcs-mcp-server/Dockerfile

Docker 構成を追加する

次の内容を Dockerfile に貼り付けます。

FROM python:3.11-slim
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
WORKDIR /app
RUN apt-get update && apt-get install -y \
   build-essential \
   gcc \
   && rm -rf /var/lib/apt/lists/*
RUN pip install --upgrade pip
COPY . .
RUN pip install -r requirements.txt
ENV PORT=8080
EXPOSE 8080
CMD ["python", "main.py"]

コンテンツを追加したら、ファイルを保存します。プロジェクト構造は次のようになります。

gcs-mcp-server/
├── requirements.txt
├── main.py
└── Dockerfile

5. Cloud Run にデプロイする

ソースから直接 MCP サーバーをデプロイします。

Cloud Shell で次のコマンドを実行します。

gcloud run deploy gcs-mcp-server \
   --no-allow-unauthenticated \
   --region=us-central1 \
   --source=. \
   --labels=session=buildersdayblr

次のように求められたら

• 未認証の呼び出しを許可しますか？→ [いいえ]

Cloud Build は以下を行います。

• コンテナ イメージをビルドする
• Artifact Registry に push する
• Cloud Run にデプロイする

[Y] と入力して、Artifact Registry リポジトリを作成できることを確認します。

Deploying from source requires an Artifact Registry Docker repository to store built containers. A repository named [cloud-run-source-deploy] in region [us-central1] will be created.

Do you want to continue (Y/n)?  Y

⚠️ 注: 本番環境では、不正アクセスを防ぐため、常に --no-allow-unauthenticated フラグを使用してください。次のステップで、ID トークンを使用して Gemini CLI を認証します。

デプロイが成功すると、Cloud Run サービスの URL を含む成功メッセージが表示されます。

[Google Cloud コンソール] の Cloud Run → Servicesでデプロイを確認することもできます。

6. Gemini CLI の構成

これまでのところ、MCP サーバーを構築して Cloud Run にデプロイしました。

ここからが楽しいところです。Gemini CLI と接続し、自然言語プロンプトを実際のクラウド アクションに変えます。

Cloud Run 起動元の権限を付与する

Cloud Run サービスは非公開であるため、ID トークンを使用して認証し、適切な IAM 権限を割り当てます。

サービスは --no-allow-unauthenticated でデプロイされているため、呼び出す権限を付与する必要があります。

プロジェクト ID を設定します。

export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project)

Cloud Run 起動元 ロールを自分に付与します。

gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
  --member=user:$(gcloud config get-value account) \
  --role='roles/run.invoker'

これにより、アカウントは Cloud Run サービスを安全に呼び出すことができます。

ID トークンを生成する

Cloud Run では、認証されたアクセスに ID トークンが必要です。

生成します。

export PROJECT_NUMBER=$(gcloud projects describe $GOOGLE_CLOUD_PROJECT --format="value(projectNumber)")
export ID_TOKEN=$(gcloud auth print-identity-token)

確認します。

echo $PROJECT_NUMBER
echo $ID_TOKEN

このトークンは Gemini CLI の構成で使用します。

Gemini CLI で MCP サーバーを構成する

Gemini CLI 設定ファイルを開きます。

cloudshell edit ~/.gemini/settings.json

次の構成を追加します。

{
 "ide": {
   "enabled": true,
   "hasSeenNudge": true
 },
 "mcpServers": {
   "my-cloudrun-server": {
     "httpUrl": "https://gcs-mcp-server-$PROJECT_NUMBER.asia-south1.run.app/mcp",
     "headers": {
       "Authorization": "Bearer $ID_TOKEN"
     }
   }
 },
 "security": {
   "auth": {
     "selectedType": "cloud-shell"
   }
 }
}

Gemini CLI で構成された MCP サーバーを検証する

次のコマンドを使用して、Cloud Shell ターミナルで Gemini CLI を起動します。

gemini

次の出力が表示されます。

Gemini CLI で次のコマンドを実行します。

/mcp refresh
/mcp list

gcs-cloudrun-server が登録されていることがわかります。サンプル スクリーンショットを以下に示します。

7. 自然言語で Google Storage オペレーションを呼び出す

⚠️ Google Cloud Storage ではグローバルに一意の名前のみが許可されるため、以下のプロンプトの my-ai-bucket 名をグローバルに一意のバケット名に置き換えてください。バケットがすでに存在するというエラーが表示された場合は、別の名前を試してください。

バケットを作成する

Create a bucket named my-ai-bucket in asia-south1 region

これにより、MCP サーバーから create_bucket ツールを呼び出す権限を求めるプロンプトが表示されます。

[Allow once] をクリックすると、リクエストした特定のリージョンにバケットが正常に作成されます。

バケットを一覧表示する

バケットを一覧表示するには、次のプロンプトを入力します。

List all my GCS buckets

バケットを削除する

バケットを削除するには、次のプロンプトを入力します（<your_bucket_name> はバケット名に置き換えます）。

Delete the bucket <your_bucket_name>

バケットのメタデータを取得する

バケットのメタデータを取得するには、次のプロンプトを入力します（<your_bucket_name> はバケット名に置き換えます）。

Give me metadata of the <your_bucket_name>

8. クリーンアップ

Google Cloud プロジェクトの削除は通常元に戻せないため、このセクション全体を読んでから削除するかどうかを決定してください。

この Codelab で使用したリソースについて、Google Cloud アカウントに課金されないようにするには、次の手順を行います。

• Google Cloud コンソールで、[リソースの管理] ページに移動します。
• プロジェクト リストで、削除するプロジェクトを選択します。
• [削除] をクリックします。

ダイアログで**プロジェクト ID** を入力し、[シャットダウン] をクリックしてプロジェクトを完全に削除します。

プロジェクトを削除すると、そのプロジェクト内で使用されているすべてのリソース（Cloud Run サービスや Artifact Registry に保存されているコンテナ イメージなど）に対する課金が停止します。

または、プロジェクトを保持したままデプロイされたサービスを削除する場合は、次の操作を行います。

1. Google Cloud コンソールで [Cloud Run] に移動します。
2. [gcs-mcp-server サービス] を選択します。
3. [削除] をクリックしてサービスを削除します。

または、Cloud Shell ターミナルで次の gcloud コマンドを実行します。

gcloud run services delete gcs-mcp-server --region=us-central1

9. まとめ

🎉 お疲れさまでした。AI を活用した最初のクラウド ワークフローを構築しました。

次のことを実装しました。

• カスタム Python ベースの MCP サーバー
• Google Cloud Storage のツール呼び出し機能
• Docker を使用したコンテナ化
• Cloud Run への安全なデプロイ
• ID ベースの認証
• Gemini CLI との統合

このアーキテクチャを拡張して、BigQuery、Pub/Sub、Compute Engine などの追加の Google Cloud サービスをサポートできます。

このパターンは、AI システムが構造化されたツール呼び出しを通じてクラウド インフラストラクチャと安全にやり取りする方法を示しています。