Превратите естественный язык в реальные действия Google Cloud с помощью Gemini MCP на Cloud Run.

1. Введение

В этом практическом занятии вы узнаете, как создать собственный сервер MCP (Model Context Protocol) на Python, развернуть его в Google Cloud Run и подключить к нему Gemini CLI для выполнения реальных операций Google Cloud Storage с использованием естественного языка.

Архитектурный процесс: Gemini CLI → Cloud Run → 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-сервер на основе Python.
• Контейнеризация приложения
• Разверните его в Cloud Run.
• Защитите его с помощью IAM и токенов идентификации.
• Подключите его с помощью Gemini CLI.
• Выполняйте операции на наземной станции управления в режиме реального времени, используя естественный язык.

Что вы узнаете

• Что такое MCP (протокол контекста модели) и как он работает?
• Как создать возможности вызова инструментов с помощью 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 на странице выбора проекта выберите или создайте проект Google Cloud.
2. Убедитесь, что для вашего облачного проекта включена функция выставления счетов. Узнайте, как проверить, включена ли функция выставления счетов для проекта .
3. Вы будете использовать Cloud Shell — среду командной строки, работающую в Google Cloud и поставляемую с предустановленным bq. Нажмите «Активировать Cloud Shell» в верхней части консоли Google Cloud.

4. После подключения к Cloud Shell необходимо проверить, прошли ли вы аутентификацию и установлен ли идентификатор вашего проекта, используя следующую команду:

gcloud auth list

5. Выполните следующую команду в Cloud Shell, чтобы убедиться, что команда gcloud знает о вашем проекте.

gcloud config list project

6. Если ваш проект не задан, используйте следующую команду для его настройки:

gcloud config set project <YOUR_PROJECT_ID>

7. Включите необходимые API с помощью команды, указанной ниже. Это может занять несколько минут, поэтому, пожалуйста, наберитесь терпения.

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

Если появится запрос на авторизацию, нажмите «Авторизовать», чтобы продолжить.

После успешного выполнения команды вы должны увидеть сообщение, похожее на показанное ниже:

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.
• Работает по протоколу 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 : Для взаимодействия с GCS импортируется библиотека google.cloud.storage , а также exceptions для обработки ошибок.

Инициализация :

Мы настраиваем формат логирования для отладки и отслеживания идентификации сервера. Кроме того, мы настраиваем экземпляр FastMCP с именем MyEnhancedGCSMCPServer . Этот объект ( mcp ) будет использоваться для регистрации всех инструментов (функций), которые предоставляет сервер. Мы определяем следующие инструменты:

• list_gcs_buckets : Получает список всех хранилищ в соответствующем проекте Google Cloud.
• create_bucket : Создает новый бакет с определенным именем и местоположением.
• delete_bucket : Удаляет существующий бакет.
• list_objects : Отображает список всех файлов (блобов) внутри определенного сегмента.
• 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 позволит:

• Создайте образ контейнера
• Отправьте его в Реестр артефактов.
• Разверните его в Cloud Run.

Введите 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

⚠️ Примечание: В производственных средах всегда используйте флаг --no-allow-unauthenticated чтобы предотвратить несанкционированный доступ. Аутентификация Gemini CLI будет выполняться с помощью токена идентификации на следующем шаге.

После успешного развертывания вы увидите сообщение об успехе с URL-адресом службы Cloud Run.

Вы также можете проверить развертывание в консоли Google Cloud в разделе Cloud Run → Services .

6. Настройка Gemini CLI

На данный момент мы создали и развернули наш сервер MCP на платформе Cloud Run.

Теперь настала самая интересная часть — подключение к Gemini CLI и превращение ваших запросов на естественном языке в реальные действия в облаке.

Предоставить Cloud Run Invoker Permission

Поскольку наш сервис 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

Вы увидите результат, показанный ниже.

В командной строке Gemini выполните следующую команду:

/mcp refresh
/mcp list

Теперь вы должны увидеть, что ваш gcs-cloudrun-serve зарегистрирован. Пример скриншота показан ниже:

7. Вызов операций хранилища Google с помощью естественного языка.

⚠️ Не забудьте заменить имя my-ai-bucket в приведенном ниже запросе на глобально уникальное имя корзины, поскольку Google Cloud Storage разрешает только глобально уникальные имена. Если вы получите ошибку о том, что корзина уже существует, попробуйте другое имя.

Создать корзину

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

В результате вам будет предложено запросить разрешение на вызов инструмента create_bucket с сервера MCP.

Нажмите кнопку «Разрешить» один раз, и ваш бакет будет успешно создан в указанном вами регионе.

Список категорий

Чтобы отобразить список категорий, введите следующую команду:

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 перейдите на страницу «Управление ресурсами».
• В списке проектов выберите проект, который хотите удалить.
• Нажмите «Удалить».

В диалоговом окне введите идентификатор проекта , а затем нажмите «Завершить», чтобы навсегда удалить проект.

Удаление проекта прекращает выставление счетов за все ресурсы, используемые в рамках этого проекта, включая сервисы Cloud Run и образы контейнеров, хранящиеся в реестре артефактов.

В качестве альтернативы, если вы хотите сохранить проект, но удалить развернутый сервис:

1. Перейдите в раздел Cloud Run в консоли Google Cloud.
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.

Этот пример демонстрирует, как системы искусственного интеллекта могут безопасно взаимодействовать с облачной инфраструктурой посредством структурированного вызова инструментов.