Biến ngôn ngữ tự nhiên thành các thao tác thực tế trên Google Cloud bằng Gemini + MCP trên Cloud Run

1. Giới thiệu

Lớp học lập trình này hướng dẫn bạn cách xây dựng một máy chủ MCP (Giao thức bối cảnh mô hình) tuỳ chỉnh bằng Python, triển khai máy chủ đó lên Google Cloud Run và kết nối máy chủ đó với Gemini CLI để thực hiện các thao tác thực tế trên Google Cloud Storage bằng ngôn ngữ tự nhiên.

Luồng kiến trúc: Gemini CLI → Cloud Run → MCP

Hãy tưởng tượng rằng bạn mở thiết bị đầu cuối và nhập một câu lệnh đơn giản trong một AI Agent, chẳng hạn như những câu lệnh dưới đây:

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

Trong vài giây, đám mây sẽ lắng nghe và thực thi. Không có lệnh phức tạp. Không cần chuyển đổi thẻ liên tục. Chỉ cần ngôn ngữ đơn giản biến thành các hành động thực tế trên đám mây.

Bạn sẽ thực hiện

Bạn sẽ tạo và triển khai một máy chủ MCP tuỳ chỉnh kết nối Gemini CLI với Google Cloud Storage.

Bạn sẽ:

• Tạo một máy chủ MCP dựa trên Python
• Đóng gói ứng dụng vào vùng chứa
• Triển khai ứng dụng đó lên Cloud Run
• Bảo mật bằng IAM và mã thông báo nhận dạng
• Kết nối với Gemini CLI
• Thực hiện các thao tác GCS trực tiếp bằng ngôn ngữ tự nhiên

Kiến thức bạn sẽ học được

• MCP (Giao thức ngữ cảnh mô hình) là gì và cách thức hoạt động
• Cách tạo các chức năng gọi công cụ bằng Python
• Cách triển khai các ứng dụng trong vùng chứa lên Cloud Run
• Cách Gemini CLI tích hợp với các máy chủ MCP bên ngoài
• Cách xác thực an toàn các dịch vụ Cloud Run
• Cách thực hiện các thao tác thực tế trên Google Cloud Storage bằng AI

Bạn cần có

• Trình duyệt web Chrome
• Tài khoản Gmail
• Một Dự án trên Google Cloud đã bật tính năng thanh toán
• Gemini CLI (Được cài đặt sẵn trong Google Cloud Shell)
• Hiểu biết cơ bản về Python và Google Cloud

Lớp học lập trình này giả định rằng người dùng đã quen thuộc với kiến thức cơ bản về Python

2. Trước khi bắt đầu

Tạo dự án

1. Trong Google Cloud Console, trên trang chọn dự án, hãy chọn hoặc tạo một dự án trên Google Cloud.
2. Đảm bảo rằng bạn đã bật tính năng thanh toán cho dự án trên Cloud. Tìm hiểu cách kiểm tra xem tính năng thanh toán có được bật trong một dự án hay không .
3. Bạn sẽ sử dụng Cloud Shell, một môi trường dòng lệnh chạy trong Google Cloud và được tải sẵn bq. Nhấp vào Kích hoạt Cloud Shell ở đầu bảng điều khiển Cloud.

4. Sau khi kết nối với Cloud Shell, bạn có thể kiểm tra để đảm bảo rằng bạn đã được xác thực và dự án được đặt thành mã dự án của bạn bằng lệnh sau:

gcloud auth list

5. Chạy lệnh sau trong Cloud Shell để xác nhận rằng lệnh gcloud biết về dự án của bạn.

gcloud config list project

6. Nếu bạn chưa đặt dự án, hãy dùng lệnh sau để đặt:

gcloud config set project <YOUR_PROJECT_ID>

7. Bật các API bắt buộc thông qua lệnh bên dưới. Quá trình này có thể mất vài phút, vì vậy, vui lòng kiên nhẫn chờ đợi.

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

Nếu được nhắc uỷ quyền, hãy nhấp vào Uỷ quyền để tiếp tục.

Khi thực thi lệnh thành công, bạn sẽ thấy một thông báo tương tự như thông báo dưới đây:

Operation "operations/..." finished successfully.

Nếu bỏ lỡ API nào, bạn luôn có thể bật API đó trong quá trình triển khai.

Tham khảo tài liệu để biết các lệnh và cách sử dụng gcloud.

Chuẩn bị dự án Python

Trong phần này, bạn sẽ tạo dự án Python lưu trữ máy chủ MCP và định cấu hình các phần phụ thuộc của dự án để triển khai đến Cloud Run.

Tạo thư mục dự án

Bắt đầu bằng cách tạo một thư mục mới có tên là mcp-on-cloudrun để lưu trữ mã nguồn:

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

Tạo requirements.txt

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

Thêm nội dung sau vào tệp:

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

Lưu tệp.

3. Tạo máy chủ MCP

Trong phần này, bạn sẽ tạo máy chủ MCP để hiển thị các thao tác Google Cloud Storage dưới dạng các công cụ có thể gọi.

Máy chủ này sẽ:

• Đăng ký công cụ MCP
• Kết nối với Google Cloud Storage
• Chạy qua HTTP
• Có thể triển khai lên Cloud Run

Giờ hãy tạo logic MCP cốt lõi bên trong main.py.

Dưới đây là mã hoàn chỉnh xác định nhiều công cụ để quản lý Google Cloud Storage – từ việc liệt kê và tạo bộ chứa cho đến việc tải lên, tải xuống và quản lý các blob

Tạo tệp ứng dụng chính

Trong thư mục mcp-on-cloudrun, hãy tạo một tệp mới có tên là main.py:

touch main.py

Mở tệp bằng Cloud Shell Editor:

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

Thêm nguồn sau vào nội dung tệp 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,
      )
  )

Lưu tệp sau khi thêm mã.

Cấu trúc dự án của bạn hiện sẽ có dạng như sau:

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

Hãy cùng tìm hiểu sơ bộ về đoạn mã này:

Nhập và thiết lập:

Mã này bắt đầu bằng cách nhập các thư viện cần thiết.

• Thư viện chuẩn: asyncio để thực thi không đồng bộ, logging để xuất thông báo trạng thái và os cho các biến môi trường.
• FastMCP: Khung cốt lõi được dùng để tạo máy chủ Giao thức ngữ cảnh mô hình.
• Google Cloud Storage: Thư viện google.cloud.storage được nhập để tương tác với GCS, cùng với exceptions để xử lý lỗi.

Khởi chạy:

Chúng tôi định cấu hình định dạng nhật ký để giúp gỡ lỗi và theo dõi danh tính của máy chủ. Ngoài ra, chúng ta sẽ định cấu hình một phiên bản của FastMCP có tên là MyEnhancedGCSMCPServer. Đối tượng này (mcp) sẽ được dùng để đăng ký tất cả các công cụ (hàm) mà máy chủ cung cấp. Chúng tôi xác định các công cụ sau:

• list_gcs_buckets: Truy xuất danh sách tất cả các bộ chứa lưu trữ trong dự án trên đám mây của Google được liên kết.
• create_bucket: Tạo một bộ chứa mới có tên và vị trí cụ thể.
• delete_bucket: Xoá một nhóm hiện có.
• list_objects: Liệt kê tất cả các tệp (BLOB) trong một bộ chứa cụ thể.
• delete_blob: Xoá một tệp cụ thể khỏi một nhóm.
• get_bucket_metadata: Trả về thông tin kỹ thuật về một bộ chứa (Vị trí, Lớp lưu trữ, Trạng thái kiểm soát phiên bản, Thời gian tạo).
• get_blob_metadata: Trả về thông tin chi tiết về kỹ thuật của một tệp cụ thể (Kích thước, Loại nội dung, Hàm băm MD5, Lần cập nhật gần đây nhất).

Điểm truy cập:

Thao tác này định cấu hình cổng, mặc định là 8080 nếu không được đặt. Sau đó, nó dùng asyncio.run() để khởi động máy chủ không đồng bộ bằng mcp.run_async. Cuối cùng, nó định cấu hình máy chủ để chạy qua HTTP (host 0.0.0.0), giúp máy chủ có thể truy cập được đối với các yêu cầu mạng đến.

4. Tạo vùng chứa cho Máy chủ MCP

Trong phần này, bạn sẽ tạo một Dockerfile để có thể triển khai máy chủ MCP lên Cloud Run.

Cloud Run yêu cầu một ứng dụng nằm trong vùng chứa. Bạn sẽ xác định cách ứng dụng được tạo và khởi động.

Tạo Dockerfile

Tạo một tệp mới có tên là Dockerfile:

touch Dockerfile

Mở tệp đó trong Cloud Shell Editor:

cloudshell edit ~/gcs-mcp-server/Dockerfile

Thêm cấu hình Docker

Dán nội dung sau vào 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"]

Lưu tệp sau khi thêm nội dung. Cấu trúc dự án của bạn hiện sẽ có dạng như sau:

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

5. Triển khai lên Cloud Run

Giờ đây, hãy triển khai máy chủ MCP ngay từ nguồn.

Chạy lệnh sau trong Cloud Shell:

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

Khi được nhắc

• Cho phép lệnh gọi chưa được xác thực? → Không

Cloud Build sẽ:

• Tạo hình ảnh vùng chứa
• Đẩy đến Artifact Registry
• Triển khai ứng dụng đó lên Cloud Run

Nhập Y để xác nhận rằng bạn có thể tạo kho lưu trữ 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

⚠️ Lưu ý: Luôn sử dụng cờ --no-allow-unauthenticated cho môi trường phát hành công khai để ngăn chặn truy cập trái phép.Bạn sẽ xác thực Gemini CLI bằng Mã thông báo nhận dạng ở bước tiếp theo.

Sau khi triển khai thành công, bạn sẽ thấy một thông báo thành công kèm theo URL dịch vụ Cloud Run.

Bạn cũng có thể xác minh việc triển khai trên Bảng điều khiển Google Cloud trong phần Cloud Run → Services.

6. Cấu hình Gemini CLI

Đến nay, chúng ta đã tạo và triển khai máy chủ MCP trên Cloud Run.

Giờ là lúc bạn có thể kết nối với Gemini CLI và biến câu lệnh bằng ngôn ngữ tự nhiên thành các hành động thực tế trên đám mây.

Cấp quyền gọi Cloud Run

Vì dịch vụ Cloud Run của chúng tôi là dịch vụ riêng tư, nên chúng tôi sẽ xác thực bằng mã nhận dạng và chỉ định các quyền IAM phù hợp.

Chúng tôi triển khai dịch vụ này bằng --no-allow-unauthenticated, do đó, bạn phải cấp quyền để gọi dịch vụ này.

Đặt mã dự án:

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

Cấp cho chính bạn vai trò Cloud Run Invoker (Đối tượng gọi Cloud Run):

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

Điều này cho phép tài khoản của bạn gọi dịch vụ Cloud Run một cách an toàn.

Tạo mã thông báo nhận dạng

Cloud Run yêu cầu Mã nhận dạng để truy cập có xác thực.

Tạo một:

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

Xác minh:

echo $PROJECT_NUMBER
echo $ID_TOKEN

Bạn sẽ sử dụng mã thông báo này trong cấu hình Gemini CLI.

Định cấu hình MCP Server trong Gemini CLI

Mở tệp cài đặt Gemini CLI:

cloudshell edit ~/.gemini/settings.json

Thêm cấu hình sau:

{
 "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"
   }
 }
}

Xác thực các máy chủ MCP được định cấu hình trong Gemini CLI

Chạy Gemini CLI trong cửa sổ dòng lệnh Cloud Shell bằng lệnh sau:

gemini

Bạn sẽ thấy kết quả dưới đây

Trong Gemini CLI, hãy chạy:

/mcp refresh
/mcp list

Lúc này, bạn sẽ thấy gcs-cloudrun-serve đã đăng ký. Dưới đây là ảnh chụp màn hình mẫu:

7. Gọi các thao tác trên Bộ nhớ Google bằng ngôn ngữ tự nhiên

⚠️ Nhớ thay thế tên my-ai-bucket trong lời nhắc bên dưới bằng một tên bộ chứa duy nhất trên toàn cầu vì Google Cloud Storage chỉ cho phép tên duy nhất trên toàn cầu. Nếu bạn gặp lỗi cho biết rằng nhóm đã tồn tại, hãy thử một tên khác.

Tạo bộ chứa

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

Thao tác này sẽ nhắc bạn yêu cầu quyền gọi công cụ create_bucket từ Máy chủ MCP.

Nhấp vào Cho phép một lần, sau đó, bạn sẽ tạo thành công vùng chứa trong khu vực cụ thể mà bạn yêu cầu.

Liệt kê các vùng chứa

Để liệt kê các nhóm, hãy nhập câu lệnh bên dưới:

List all my GCS buckets

Xoá bộ chứa

Để xoá một nhóm, hãy nhập câu lệnh bên dưới (thay thế <your_bucket_name> bằng tên nhóm của bạn):

Delete the bucket <your_bucket_name>

Lấy siêu dữ liệu của bộ chứa

Để lấy siêu dữ liệu của một nhóm, hãy nhập câu lệnh bên dưới (thay <your_bucket_name> bằng tên nhóm của bạn):

Give me metadata of the <your_bucket_name>

8. Dọn dẹp

Trước tiên, hãy đọc toàn bộ phần này trước khi quyết định xoá dự án trên Google Cloud, vì thông thường bạn không thể đảo ngược thao tác này.

Để tránh phát sinh phí cho tài khoản Google Cloud của bạn đối với các tài nguyên được dùng trong lớp học lập trình này, hãy làm theo các bước sau:

• Trong Google Cloud Console, hãy chuyển đến trang Quản lý tài nguyên.
• Trong danh sách dự án, hãy chọn dự án mà bạn muốn xoá.
• Nhấp vào Xoá.

Trong hộp thoại, hãy nhập Mã dự án, rồi nhấp vào Tắt để xoá vĩnh viễn dự án.

Khi bạn xoá dự án, hoạt động thanh toán cho tất cả tài nguyên được dùng trong dự án đó sẽ dừng lại, bao gồm cả các dịch vụ Cloud Run và hình ảnh vùng chứa được lưu trữ trong Artifact Registry.

Hoặc nếu bạn muốn giữ lại dự án nhưng xoá dịch vụ đã triển khai:

1. Chuyển đến Cloud Run trong Google Cloud Console.
2. Chọn dịch vụ gcs-mcp-server.
3. Nhấp vào Xoá để xoá dịch vụ.

hoặc đưa ra lệnh gcloud sau đây trong thiết bị đầu cuối Cloud Shell.

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

9. Kết luận

🎉 Xin chúc mừng! Bạn vừa tạo quy trình làm việc trên đám mây đầu tiên dựa trên AI!

Bạn đã triển khai:

• Máy chủ MCP tuỳ chỉnh dựa trên Python
• Khả năng gọi công cụ cho Google Cloud Storage
• Tạo vùng chứa bằng Docker
• Triển khai bảo mật trên Cloud Run
• Xác thực dựa trên danh tính
• Tích hợp với Gemini CLI

Giờ đây, bạn có thể mở rộng kiến trúc này để hỗ trợ các dịch vụ bổ sung của Google Cloud, chẳng hạn như BigQuery, Pub/Sub hoặc Compute Engine.

Mẫu này minh hoạ cách các hệ thống AI có thể tương tác an toàn với cơ sở hạ tầng đám mây thông qua lệnh gọi công cụ có cấu trúc.