利用 Cloud Run 上的 Gemini 和 MCP 将自然语言转换为实际的 Google Cloud 操作

1. 简介

本 Codelab 将引导您使用 Python 构建自定义 MCP（模型上下文协议）服务器，将其部署到 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

几秒钟内，云端会听取并执行指令。无需复杂的命令。无需在标签页之间来回切换。只需使用简单的语言，即可将自然语言转换为实际的云操作。

实践内容

您将构建并部署一个自定义 MCP 服务器，用于将 Gemini CLI 与 Google Cloud Storage 相连接。

您将学习以下内容：

构建基于 Python 的 MCP 服务器
将应用容器化
将其部署到 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. 准备工作

创建项目

在 Google Cloud Console 的项目选择器页面上，选择或创建一个 Google Cloud 项目。
确保您的 Cloud 项目已启用结算功能。了解如何检查项目是否已启用结算功能。
您将使用 Cloud Shell，这是一个在 Google Cloud 中运行的命令行环境，它预加载了 bq。点击 Google Cloud 控制台顶部的“激活 Cloud Shell”。

“激活 Cloud Shell”按钮图片

连接到 Cloud Shell 后，您可以使用以下命令检查自己是否已通过身份验证，以及项目是否已设置为您的项目 ID：

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 命令和用法，请参阅文档。

准备 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 服务器

在本部分中，您将创建 MCP 服务器，该服务器会将 Google Cloud Storage 操作公开为可调用的工具。

此服务器将：

注册 MCP 工具
连接到 Google Cloud Storage
通过 HTTP 运行
可部署到 Cloud Run

现在，我们来创建 main.py 内的核心 MCP 逻辑。

以下是完整代码，其中定义了用于管理 Google Cloud Storage 的多种工具，包括列出和创建存储分区，以及上传、下载和管理 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：导入 google.cloud.storage 库以与 GCS 交互，并导入 exceptions 以进行错误处理。

初始化：

我们配置了日志记录格式，以帮助调试和跟踪服务器的身份。此外，我们还配置了一个名为 MyEnhancedGCSMCPServer 的 FastMCP 实例。此对象 (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) 运行，从而使其可供传入的网络请求访问。

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

当系统提示

允许未经身份验证的调用？→ 否

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

成功部署后，您会看到一条成功消息，其中包含 Cloud Run 服务网址。

您还可以通过 Google Cloud 控制台中的 Cloud Run → Services来验证部署。

6. Gemini CLI 配置

到目前为止，我们已在 Cloud Run 上构建并部署了 MCP 服务器。

现在到了有趣的部分 - 将其与 Gemini CLI 连接，并将自然语言提示转换为实际的云操作。

授予 Cloud Run Invoker 权限

由于我们的 Cloud Run 服务是私有的，因此我们将使用身份令牌进行身份验证，并分配正确的 IAM 权限。

我们使用 --no-allow-unauthenticated 部署了该服务，因此您必须授予调用该服务的权限。

设置项目 ID：

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 配置中使用此令牌。

在 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-serve 已注册。示例屏幕截图如下所示：

7. 通过自然语言调用 Google 存储空间操作

创建存储桶

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

系统会提示您，请求您授予从 MCP 服务器调用 create_bucket 工具的权限。

点击“允许一次”，然后系统会在您请求的特定区域中成功创建存储分区。

列出存储桶

如需列出存储分区，请输入以下提示：

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 中的容器映像。

或者，如果您想保留项目但移除已部署的服务，请执行以下操作：

在 Google Cloud 控制台中，前往 Cloud Run。
选择 gcs-mcp-server 服务。
点击“删除”以移除相应服务。

或者在 Cloud Shell 终端中运行以下 gcloud 命令。

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

9. 总结

🎉 恭喜！您刚刚构建了第一个 AI 赋能的云工作流！

您已实现：

基于 Python 的自定义 MCP 服务器
Google Cloud Storage 的工具调用功能
使用 Docker 进行容器化
安全地部署到 Cloud Run
基于身份的身份验证
与 Gemini CLI 集成

现在，您可以扩展此架构，以支持其他 Google Cloud 服务，例如 BigQuery、Pub/Sub 或 Compute Engine。

此模式演示了 AI 系统如何通过结构化工具调用与云基础架构安全地交互。