הפיכת שפה טבעית לפעולות אמיתיות ב-Google Cloud באמצעות Gemini + MCP ב-Cloud Run

1. מבוא

ב-Codelab הזה נסביר איך לבנות שרת MCP (Model Context Protocol) מותאם אישית באמצעות Python, לפרוס אותו ב-Google Cloud Run ולקשר אותו ל-Gemini CLI כדי לבצע פעולות אמיתיות ב-Google Cloud Storage באמצעות שפה טבעית.

זרימה ארכיטקטונית: Gemini CLI ‏→ Cloud Run ‏→ MCP

e149713a547f4157.png

תארו לעצמכם את התרחיש הבא: אתם פותחים את הטרמינל ומקלידים הנחיה פשוטה בסוכן AI, כמו ההנחיות שמוצגות בהמשך:

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

תוך שניות, הענן מקשיב ומבצע את הפעולה. בלי פקודות מסובכות. לא צריך לעבור בין כרטיסיות בלי סוף. פשוט שפה רגילה שהופכת לפעולות אמיתיות בענן.

מה תעשו

תבנו ותפרסו שרת MCP מותאם אישית שמקשר בין Gemini CLI לבין Google Cloud Storage.

תצטרכו:

  • פיתוח שרת MCP מבוסס-Python
  • העברת האפליקציה לקונטיינר
  • פריסה ב-Cloud Run
  • אבטחה באמצעות IAM ואסימוני זהות
  • קישור ל-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 שנטען מראש. לוחצים על 'הפעלת Cloud Shell' בחלק העליון של מסוף Google Cloud.

תמונה של לחצן ההפעלה של Cloud Shell

  1. אחרי שמתחברים ל-Cloud Shell, בודקים שכבר בוצע אימות ושהפרויקט מוגדר למזהה הפרויקט באמצעות הפקודה הבאה:
gcloud auth list
  1. מריצים את הפקודה הבאה ב-Cloud Shell כדי לוודא שפקודת gcloud מכירה את הפרויקט.
gcloud config list project
  1. אם הפרויקט לא מוגדר, משתמשים בפקודה הבאה כדי להגדיר אותו:
gcloud config set project <YOUR_PROJECT_ID>
  1. מפעילים את ממשקי ה-API הנדרשים באמצעות הפקודה שמוצגת למטה. זה יימשך כמה דקות, אז כדאי לחכות בסבלנות.
gcloud services enable \
  run.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com

אם מתבקשים לאשר, לוחצים על 'אישור' כדי להמשיך.

5e681903144bdfbe.png

אם הפקודה תפעל בהצלחה, תוצג הודעה שדומה לזו שמופיעה בהמשך:

Operation "operations/..." finished successfully.

אם פספסתם API כלשהו, תמיד תוכלו להפעיל אותו במהלך ההטמעה.

אפשר לעיין במאמרי העזרה בנושא פקודות gcloud ושימוש בהן.

הכנת פרויקט Python

בקטע הזה ניצור את פרויקט Python שיארח את שרת ה-MCP ונגדיר את התלות שלו לצורך פריסה ב-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

בקטע הזה ניצור את שרת ה-MCP שחושף פעולות של Google Cloud Storage ככלים שאפשר להפעיל.

השרת הזה:

  • רישום כלי MCP
  • חיבור ל-Google Cloud Storage
  • הרצה באמצעות HTTP
  • אפשר לפרוס אותם ב-Cloud Run

עכשיו ניצור את הלוגיקה המרכזית של MCP בתוך main.py.

בהמשך מופיע הקוד המלא שמגדיר כמה כלים לניהול Google Cloud Storage – החל מרישום ויצירה של קטגוריות ועד להעלאה, להורדה ולניהול של אובייקטים מסוג Blob

יצירת קובץ האפליקציה הראשי

בתוך הספרייה mcp-on-cloudrun, יוצרים קובץ חדש בשם main.py:

touch main.py

פותחים את הקובץ באמצעות Cloud Shell Editor:

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: הספרייה google.cloud.storage מיובאת כדי ליצור אינטראקציה עם GCS, יחד עם exceptions לטיפול בשגיאות.

הפעלה ראשונית:

אנחנו מגדירים את פורמט הרישום ביומן כדי לעזור בניפוי באגים ובמעקב אחרי הזהות של השרת. בנוסף, אנחנו מגדירים מופע של FastMCP בשם MyEnhancedGCSMCPServer. האובייקט הזה (mcp) ישמש לרישום של כל הכלים (הפונקציות) שהשרת חושף. הגדרנו את הכלים הבאים:

  • ‫list_gcs_buckets: מאחזר רשימה של כל מאגרי האחסון בפרויקט בענן המשויך ב-Google Cloud.
  • ‫create_bucket: יוצר קטגוריה חדשה עם שם ומיקום ספציפיים.
  • ‫delete_bucket: מחיקת קטגוריה קיימת.
  • list_objects: מציגה רשימה של כל הקבצים (blobs) בתוך קטגוריה ספציפית.
  • ‫delete_blob: מחיקת קובץ ספציפי יחיד מקטגוריה.
  • ‫get_bucket_metadata: מחזירה פרטים טכניים על קטגוריה (מיקום, סוג אחסון, סטטוס ניהול גרסאות, זמן יצירה).
  • ‫get_blob_metadata: מחזירה פרטים טכניים על קובץ ספציפי (גודל, סוג תוכן, גיבוב MD5, עדכון אחרון).

נקודת כניסה:

ההגדרה הזו קובעת את היציאה, וברירת המחדל היא 8080 אם לא הוגדרה יציאה. לאחר מכן, הוא משתמש ב-asyncio.run() כדי להפעיל את השרת באופן אסינכרוני עם mcp.run_async. לבסוף, הוא מגדיר את השרת להרצה באמצעות HTTP ‏ (host 0.0.0.0), כך שניתן לגשת אליו לבקשות רשת נכנסות.

4. העברת שרת ה-MCP למאגר

בקטע הזה, יוצרים Dockerfile כדי שיהיה אפשר לפרוס את שרת ה-MCP ב-Cloud Run.

ב-Cloud Run נדרשת אפליקציה בקונטיינר. תגדירו איך האפליקציה שלכם בנויה ואיך היא מופעלת.

יצירת קובץ Dockerfile

יוצרים קובץ חדש בשם Dockerfile:

touch Dockerfile

פותחים אותו ב-Cloud Shell Editor:

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

כשמופיעה בקשה

  • לאפשר הפעלות לא מאומתות? ‫→ No

‫Cloud Build יבצע את הפעולות הבאות:

  • יוצרים את קובץ האימג' של הקונטיינר
  • העלאה ל-Artifact Registry
  • פריסה ב-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

אחרי פריסה מוצלחת, תוצג הודעת הצלחה עם כתובת ה-URL של שירות Cloud Run.

אפשר גם לאמת את הפריסה ממסוף Google Cloud בקטע Cloud Run → Services.

53f95a2aa7a169d6.png

6. הגדרת Gemini CLI

עד עכשיו, בנינו ופרסנו את שרת ה-MCP שלנו ב-Cloud Run.

עכשיו מגיע החלק הכיפי – חיבור ל-Gemini CLI והפיכת ההנחיות בשפה טבעית לפעולות אמיתיות בענן.

מתן הרשאה ל-Cloud Run Invoker

מכיוון ששירות Cloud Run שלנו הוא פרטי, נאמת את הזהות באמצעות אסימון זהות ונשייך את הרשאות ה-IAM הנכונות.

פרסנו את השירות עם --no-allow-unauthenticated ולכן צריך לתת הרשאה להפעיל אותו.

מגדירים את מזהה הפרויקט:

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

מקצים לעצמכם את התפקיד Cloud Run Invoker:

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

כך החשבון יכול להפעיל את שירות Cloud Run בצורה מאובטחת.

יצירת אסימון זהות

כדי לקבל גישה מאומתת ל-Cloud Run, צריך אסימון זהות.

יצירת תמונה:

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.

הגדרת שרת ה-MCP ב-Gemini CLI

פותחים את קובץ ההגדרות של 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"
   }
 }
}

אימות שרתי ה-MCP שהוגדרו ב-Gemini CLI

מריצים את הפקודה הבאה כדי להפעיל את Gemini CLI בטרמינל של Cloud Shell:

gemini

יוצג הפלט הבא

193224319056d340.png

ב-Gemini CLI, מריצים את הפקודה:

/mcp refresh
/mcp list

עכשיו אמור להופיע הכיתוב gcs-cloudrun-server רשום. למטה מוצגת דוגמה לצילום מסך:

726738c48290fc30.png

7. הפעלת פעולות ב-Google Storage באמצעות שפה טבעית

Create bucket

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

תופיע בקשה לאישור שלכם להפעיל את הכלי create_bucket משרת ה-MCP.

5ab2225295285077.png

לוחצים על 'אישור חד-פעמי', ואז הדלי ייווצר בהצלחה באזור הספציפי שביקשתם.

List buckets

כדי להציג את רשימת הדליים, מזינים את ההנחיה הבאה:

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, כי בדרך כלל אי אפשר לבטל את הפעולה הזו.

כדי לא לצבור חיובים לחשבון Google Cloud על המשאבים שבהם השתמשתם ב-Code Lab הזה:

  • במסוף Google Cloud, עוברים לדף Manage resources.
  • ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק.
  • לוחצים על מחיקה.

בתיבת הדו-שיח, כותבים את מזהה הפרויקט ולוחצים על 'השבתה' כדי למחוק את הפרויקט לצמיתות.

מחיקת הפרויקט תגרום להפסקת החיוב על כל המשאבים שנעשה בהם שימוש בפרויקט, כולל שירותי Cloud Run וקובצי אימג' בקונטיינר שאוחסנו ב-Artifact Registry.

לחלופין, אם רוצים לשמור את הפרויקט אבל להסיר את השירות שנפרס:

  1. נכנסים אל Cloud Run במסוף Google Cloud.
  2. בוחרים באפשרות gcs-mcp-server service (שירות gcs-mcp-server).
  3. לוחצים על 'מחיקה' כדי להסיר את השירות.

או מזינים את הפקודה gcloud הבאה בטרמינל של Cloud Shell.

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

9. סיכום

‫🎉 איזה כיף! הרגע יצרתם את תהליך העבודה הראשון שלכם בענן שמבוסס על AI.

הטמעת:

  • שרת MCP מותאם אישית שמבוסס על Python
  • יכולות של הפעלת כלים ב-Google Cloud Storage
  • יצירת קונטיינרים באמצעות Docker
  • פריסה מאובטחת ב-Cloud Run
  • אימות מבוסס-זהות
  • שילוב עם Gemini CLI

עכשיו אפשר להרחיב את הארכיטקטורה הזו כדי לתמוך בשירותים נוספים של Google Cloud, כמו BigQuery, ‏ Pub/Sub או Compute Engine.

הדפוס הזה מדגים איך מערכות AI יכולות לקיים אינטראקציה מאובטחת עם תשתית ענן באמצעות הפעלה מובנית של כלים.