تحويل اللغة الطبيعية إلى إجراءات حقيقية على Google Cloud باستخدام Gemini وMCP على Cloud Run

1. مقدمة

يوضّح لك هذا الدرس التطبيقي حول الترميز كيفية إنشاء خادم MCP (بروتوكول سياق النموذج) مخصّص باستخدام Python، ونشره على Google Cloud Run، وربطه بواجهة سطر الأوامر Gemini CLI لتنفيذ عمليات فعلية على Google Cloud Storage باستخدام اللغة الطبيعية.

المخطط المعماري: Gemini CLI → Cloud Run → MCP

e149713a547f4157.png

تخيَّل ما يلي: تفتح نافذة الأوامر وتكتب طلبًا بسيطًا في "وكيل الذكاء الاصطناعي"، مثل الطلبات الموضّحة أدناه:

  • 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
  • تأمينها باستخدام خدمة "إدارة الهوية وإمكانية الوصول" ورموز التعريف
  • ربطها بـ Gemini CLI
  • تنفيذ عمليات GCS المباشرة باستخدام اللغة الطبيعية

ما ستتعلمه

  • التعريف ببروتوكول MCP (Model Context Protocol) وطريقة عمله
  • كيفية إنشاء إمكانات استدعاء الأدوات باستخدام Python
  • كيفية نشر التطبيقات المحفوظة في حاويات إلى Cloud Run
  • كيفية دمج Gemini CLI مع خوادم MCP خارجية
  • كيفية مصادقة خدمات Cloud Run بأمان
  • كيفية تنفيذ عمليات Google Cloud Storage الحقيقية باستخدام الذكاء الاصطناعي

المتطلبات

  • متصفّح الويب Chrome
  • حساب Gmail
  • مشروع Google Cloud تم تفعيل الفوترة فيه
  • ‫Gemini CLI (تأتي مثبّتة مسبقًا مع Google Cloud Shell)
  • معرفة أساسية بلغة Python وGoogle Cloud

يفترض هذا الدرس التطبيقي حول الترميز أنّ المستخدم على دراية بالمعلومات الأساسية عن لغة Python.

2. قبل البدء

إنشاء مشروع

  1. في Google Cloud Console، ضمن صفحة اختيار المشروع، اختَر مشروعًا على Google Cloud أو أنشِئه.
  2. تأكَّد من تفعيل الفوترة لمشروعك على السحابة الإلكترونية. تعرَّف على كيفية التحقّق مما إذا كانت الفوترة مفعَّلة في مشروع .
  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. فعِّل واجهات برمجة التطبيقات المطلوبة من خلال الأمر الموضّح أدناه. قد تستغرق هذه العملية بضع دقائق، لذا يُرجى الانتظار.
gcloud services enable \
  run.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com

إذا طُلب منك التفويض، انقر على "تفويض" للمتابعة.

5e681903144bdfbe.png

عند تنفيذ الأمر بنجاح، من المفترض أن تظهر لك رسالة مشابهة للرسالة الموضّحة أدناه:

Operation "operations/..." finished successfully.

في حال عدم توفّر أي واجهة برمجة تطبيقات، يمكنك تفعيلها في أي وقت أثناء عملية التنفيذ.

راجِع المستندات لمعرفة أوامر 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، بدءًا من إدراج الحِزم وإنشائها إلى تحميل الكائنات الثنائية الكبيرة وتنزيلها وإدارتها

إنشاء ملف التطبيق الرئيسي

داخل دليل 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: يتم استيراد مكتبة 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":

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
  • نشرها على 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:

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 Server في 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 a bucket named my-ai-bucket in asia-south1 region

سيؤدي ذلك إلى مطالبتك بتقديم إذنك لاستخدام أداة create_bucket من خادم MCP.

5ab2225295285077.png

انقر على "السماح مرة واحدة"، وسيتم إنشاء الحزمة بنجاح في المنطقة المحدّدة التي طلبتها.

قوائم الحزم

لعرض قائمة بالحِزم، أدخِل الطلب التالي:

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 مقابل الموارد المستخدَمة في هذا الدرس التطبيقي حول الترميز، اتّبِع الخطوات التالية:

  • في Google Cloud Console، انتقِل إلى صفحة "إدارة الموارد".
  • في قائمة المشاريع، اختَر المشروع الذي تريد حذفه.
  • انقر على حذف.

في مربّع الحوار، أدخِل رقم تعريف المشروع، ثم انقر على "إيقاف" لحذف المشروع نهائيًا.

يؤدي حذف المشروع إلى إيقاف الفوترة لجميع الموارد المستخدَمة في هذا المشروع، بما في ذلك خدمات Cloud Run وصور الحاويات المخزّنة في Artifact Registry.

بدلاً من ذلك، إذا أردت الاحتفاظ بالمشروع ولكن إزالة الخدمة التي تم نشرها، اتّبِع الخطوات التالية:

  1. انتقِل إلى Cloud Run في Google Cloud Console.
  2. اختَر خدمة gcs-mcp-server.
  3. انقر على "حذف" لإزالة الخدمة.

أو أدخِل الأمر gcloud التالي في وحدة Cloud Shell الطرفية.

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

9- الخاتمة

🎉 تهانينا! لقد أنشأت للتو أول سير عمل مستند إلى السحابة الإلكترونية ومزوّد بالذكاء الاصطناعي.

نفّذت ما يلي:

  • خادم MCP مخصّص مستنِد إلى Python
  • إمكانات استدعاء الأدوات في Google Cloud Storage
  • استخدام الحاويات باستخدام Docker
  • النشر بأمان على Cloud Run
  • المصادقة المستندة إلى الهوية
  • التكامل مع Gemini CLI

يمكنك الآن توسيع هذه البنية الأساسية لتشمل خدمات إضافية من Google Cloud، مثل BigQuery أو Pub/Sub أو Compute Engine.

يوضّح هذا النمط كيف يمكن لأنظمة الذكاء الاصطناعي التفاعل بأمان مع البنية الأساسية السحابية من خلال استدعاء الأدوات المنظَّمة.