۱. مقدمه
این آزمایشگاه کد، شما را در ساخت یک سرور MCP (پروتکل زمینه مدل) سفارشی با استفاده از پایتون، استقرار آن در Google Cloud Run و اتصال آن به Gemini CLI برای اجرای عملیات واقعی Google Cloud Storage با استفاده از زبان طبیعی، راهنمایی میکند.
جریان معماری: رابط خط فرمان Gemini → اجرای ابری → MCP

تصور کنید: ترمینال خود را باز میکنید و یک اعلان ساده را در یک عامل هوش مصنوعی تایپ میکنید، مانند مواردی که در زیر نشان داده شده است:
-
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 مبتنی بر پایتون
- کانتینرایز کردن برنامه
- آن را در Cloud Run مستقر کنید
- با استفاده از IAM و توکنهای هویت، آن را ایمن کنید
- آن را با Gemini CLI متصل کنید
- اجرای عملیات GCS به صورت زنده با استفاده از زبان طبیعی
آنچه یاد خواهید گرفت
- MCP (پروتکل زمینه مدل) چیست و چگونه کار میکند؟
- نحوه ایجاد قابلیتهای فراخوانی ابزار با استفاده از پایتون
- نحوه استقرار برنامههای کانتینر شده در Cloud Run
- نحوه ادغام رابط خط فرمان Gemini با سرورهای MCP خارجی
- چگونه سرویسهای Cloud Run را به صورت ایمن احراز هویت کنیم؟
- نحوه اجرای عملیات واقعی ذخیرهسازی ابری گوگل با استفاده از هوش مصنوعی
آنچه نیاز دارید
- مرورگر وب کروم
- یک حساب جیمیل
- یک پروژه گوگل کلود با قابلیت پرداخت صورتحساب
- رابط خط فرمان Gemini (به صورت پیشفرض با پوسته ابری گوگل نصب شده است)
- آشنایی اولیه با پایتون و گوگل کلود
این آزمایشگاه کد انتظار دارد که کاربر با دانش پایه پایتون آشنا باشد.
۲. قبل از شروع
ایجاد یک پروژه
- در کنسول گوگل کلود ، در صفحه انتخاب پروژه، یک پروژه گوگل کلود را انتخاب یا ایجاد کنید.
- مطمئن شوید که صورتحساب برای پروژه ابری شما فعال است. یاد بگیرید که چگونه بررسی کنید که آیا صورتحساب در یک پروژه فعال است یا خیر .
- شما از Cloud Shell ، یک محیط خط فرمان که در Google Cloud اجرا میشود و bq از قبل روی آن بارگذاری شده است، استفاده خواهید کرد. روی Activate Cloud Shell در بالای کنسول Google Cloud کلیک کنید.

- پس از اتصال به Cloud Shell، با استفاده از دستور زیر بررسی میکنید که آیا از قبل احراز هویت شدهاید و پروژه روی شناسه پروژه شما تنظیم شده است یا خیر:
gcloud auth list
- دستور زیر را در Cloud Shell اجرا کنید تا تأیید شود که دستور gcloud از پروژه شما اطلاع دارد.
gcloud config list project
- اگر پروژه شما تنظیم نشده است، از دستور زیر برای تنظیم آن استفاده کنید:
gcloud config set project <YOUR_PROJECT_ID>
- API های مورد نیاز را از طریق دستور زیر فعال کنید. این کار ممکن است چند دقیقه طول بکشد، پس لطفاً صبور باشید.
gcloud services enable \
run.googleapis.com \
artifactregistry.googleapis.com \
cloudbuild.googleapis.com
اگر از شما خواسته شد که مجوز دهید، برای ادامه روی تأیید کلیک کنید.

در صورت اجرای موفقیتآمیز دستور، باید پیامی مشابه آنچه در زیر نشان داده شده است را مشاهده کنید:
Operation "operations/..." finished successfully.
اگر هر API از قلم افتاده باشد، میتوانید همیشه آن را در طول پیادهسازی فعال کنید.
برای دستورات و نحوهی استفاده از gcloud به مستندات مراجعه کنید.
پروژه پایتون خود را آماده کنید
در این بخش، پروژه پایتونی ایجاد خواهید کرد که میزبان سرور 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
فایل را ذخیره کنید.
۳. سرور MCP را ایجاد کنید
در این بخش، سرور MCP را ایجاد خواهید کرد که عملیات ذخیرهسازی ابری گوگل را به عنوان ابزارهای قابل فراخوانی در معرض نمایش قرار میدهد.
این سرور:
- ابزارهای MCP را ثبت کنید
- اتصال به فضای ذخیرهسازی ابری گوگل
- اجرای HTTP
- قابل استقرار در Cloud Run باشد
حالا بیایید منطق اصلی MCP خود را درون main.py ایجاد کنیم.
در زیر کد کاملی وجود دارد که ابزارهای متعددی را برای مدیریت فضای ذخیرهسازی ابری گوگل تعریف میکند - از فهرست کردن و ایجاد باکتها گرفته تا آپلود، دانلود و مدیریت blobها
ایجاد فایل اصلی برنامه
درون دایرکتوری 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، به همراهexceptionsبرای مدیریت خطا، وارد شده است.
مقداردهی اولیه :
ما قالب ثبت وقایع را برای کمک به اشکالزدایی و ردیابی هویت سرور پیکربندی میکنیم. علاوه بر این، نمونهای از FastMCP به نام MyEnhancedGCSMCPServer را پیکربندی میکنیم. این شیء ( mcp ) برای ثبت تمام ابزارها (توابع) که سرور در معرض نمایش قرار میدهد، استفاده خواهد شد. ابزارهای زیر را تعریف میکنیم:
-
list_gcs_buckets: فهرستی از تمام سطلهای ذخیرهسازی در پروژه مرتبط با Google Cloud را بازیابی میکند. -
create_bucket: یک سطل جدید با نام و مکان مشخص ایجاد میکند. -
delete_bucket: یک سطل موجود را حذف میکند. -
list_objects: تمام فایلها (blobها) درون یک سطل خاص را لیست میکند. -
delete_blob: یک فایل خاص را از یک سطل حذف میکند. -
get_bucket_metadata: جزئیات فنی مربوط به یک باکت (مکان، کلاس ذخیرهسازی، وضعیت نسخهبندی، زمان ایجاد) را برمیگرداند. -
get_blob_metadata: جزئیات فنی مربوط به یک فایل خاص (اندازه، نوع محتوا، هش MD5، آخرین بهروزرسانی) را برمیگرداند.
نقطه ورود:
این دستور پورت را پیکربندی میکند و در صورت عدم تنظیم، آن را به صورت پیشفرض روی 8080 تنظیم میکند. سپس از asyncio.run() برای شروع ناهمگام سرور با mcp.run_async استفاده میکند. در نهایت، سرور را برای اجرا روی HTTP ( host 0.0.0.0 ) پیکربندی میکند و آن را برای درخواستهای ورودی شبکه قابل دسترسی میسازد.
۴. کانتینر کردن سرور MCP
در این بخش، شما یک Dockerfile ایجاد خواهید کرد تا سرور MCP شما بتواند روی Cloud Run مستقر شود.
Cloud Run به یک برنامه کانتینر شده نیاز دارد. شما نحوه ساخت و شروع برنامه خود را تعریف خواهید کرد.
ایجاد فایل داکر
یک فایل جدید با نام Dockerfile ایجاد کنید:
touch Dockerfile
آن را در ویرایشگر Cloud Shell باز کنید:
cloudshell edit ~/gcs-mcp-server/Dockerfile
اضافه کردن پیکربندی داکر
محتوای زیر را در 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
۵. استقرار در Cloud Run
اکنون سرور MCP خود را مستقیماً از منبع مستقر کنید.
دستور زیر را در Cloud Shell اجرا کنید:
gcloud run deploy gcs-mcp-server \
--no-allow-unauthenticated \
--region=us-central1 \
--source=. \
--labels=session=buildersdayblr
وقتی از شما خواسته شد که
- آیا اجازه فراخوانیهای احراز هویت نشده داده میشود؟ → خیر
ساخت ابری:
- ساخت تصویر کانتینر
- آن را به رجیستری مصنوعات فشار دهید
- آن را در Cloud Run مستقر کنید
برای تأیید ایجاد مخزن Artifact Registry، حرف Y را وارد کنید.
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
پس از استقرار موفقیتآمیز، یک پیام موفقیتآمیز به همراه آدرس اینترنتی سرویس Cloud Run مشاهده خواهید کرد.
همچنین میتوانید استقرار را از کنسول Google Cloud خود در مسیر Cloud Run → Services تأیید کنید .

۶. پیکربندی رابط خط فرمان Gemini
تاکنون، ما سرور MCP خود را روی Cloud Run ساخته و مستقر کردهایم.
حالا وقت بخش جالب ماجراست - اتصال آن به Gemini CLI و تبدیل دستورات زبان طبیعی شما به اقدامات ابری واقعی.
اعطای مجوز اجرای ابری 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
شما از این توکن در پیکربندی رابط خط فرمان (CLI) جمینی (Gemini) استفاده خواهید کرد.
پیکربندی سرور 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
خروجی زیر را مشاهده خواهید کرد

داخل رابط خط فرمان Gemini، دستور زیر را اجرا کنید:
/mcp refresh
/mcp list
اکنون باید gcs-cloudrun-serve خود را که ثبت شده است، ببینید. یک نمونه تصویر در زیر نشان داده شده است:

۷. فراخوانی عملیات ذخیرهسازی گوگل از طریق زبان طبیعی
ایجاد سطل
Create a bucket named my-ai-bucket in asia-south1 region
این باعث میشود که از شما اجازه دسترسی برای فراخوانی ابزار create_bucket از سرور MCP درخواست شود.

یک بار روی Allow کلیک کنید، و سپس سطل شما با موفقیت در منطقه خاصی که درخواست کردهاید ایجاد خواهد شد.
لیست سطلها
برای فهرست کردن سطلها، دستور زیر را وارد کنید:
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>
۸. تمیز کردن
قبل از تصمیم به حذف پروژه Google Cloud، ابتدا کل این بخش را بخوانید، زیرا این اقدام معمولاً قابل برگشت نیست.
برای جلوگیری از تحمیل هزینه به حساب Google Cloud خود برای منابع استفاده شده در این آزمایشگاه کد، این مراحل را دنبال کنید:
- در کنسول گوگل کلود، به صفحه مدیریت منابع بروید.
- در لیست پروژهها، پروژهای را که میخواهید حذف کنید، انتخاب کنید.
- روی حذف کلیک کنید.
در کادر محاورهای، شناسه پروژه را تایپ کنید و سپس برای حذف دائمی پروژه، روی خاموش کردن کلیک کنید.
حذف پروژه، پرداخت هزینه برای تمام منابع مورد استفاده در آن پروژه، از جمله سرویسهای Cloud Run و تصاویر کانتینر ذخیره شده در Artifact Registry را متوقف میکند.
از طرف دیگر، اگر میخواهید پروژه را نگه دارید اما سرویس مستقر شده را حذف کنید:
- در کنسول ابری گوگل به Cloud Run بروید.
- سرویس gcs-mcp-server را انتخاب کنید.
- برای حذف سرویس، روی حذف کلیک کنید..
یا دستور gcloud زیر را در ترمینال Cloud Shell اجرا کنید.
gcloud run services delete gcs-mcp-server --region=us-central1
۹. نتیجهگیری
🎉 تبریک! شما اولین گردش کار ابری مبتنی بر هوش مصنوعی خود را ساختید!
شما پیادهسازی کردید:
- یک سرور MCP سفارشی مبتنی بر پایتون
- قابلیتهای فراخوانی ابزار برای فضای ذخیرهسازی ابری گوگل
- کانتینرسازی با استفاده از داکر
- استقرار امن در Cloud Run
- احراز هویت مبتنی بر هویت
- ادغام با رابط خط فرمان Gemini
اکنون میتوانید این معماری را برای پشتیبانی از سرویسهای ابری اضافی گوگل مانند BigQuery، Pub/Sub یا Compute Engine گسترش دهید.
این الگو نشان میدهد که چگونه سیستمهای هوش مصنوعی میتوانند از طریق فراخوانی ساختاریافتهی ابزار، به طور ایمن با زیرساخت ابری تعامل داشته باشند.