Przekształcanie języka naturalnego w rzeczywiste działania w Google Cloud za pomocą Gemini i MCP w Cloud Run

1. Wprowadzenie

W tym ćwiczeniu pokazujemy, jak utworzyć niestandardowy serwer MCP (Model Context Protocol) przy użyciu Pythona, wdrożyć go w Google Cloud Run i połączyć z interfejsem wiersza poleceń Gemini, aby wykonywać rzeczywiste operacje Google Cloud Storage przy użyciu języka naturalnego.

Przepływ architektury: interfejs wiersza poleceń Gemini → Cloud Run → MCP

e149713a547f4157.png

Wyobraź sobie, że otwierasz terminal i wpisujesz w agencie AI prosty prompt, taki jak te poniżej:

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

W ciągu kilku sekund chmura nasłuchuje i wykonuje polecenie. Bez skomplikowanych poleceń. Koniec z niekończącym się przełączaniem kart. To po prostu język naturalny zamieniany na rzeczywiste działania w chmurze.

Co musisz zrobić

Utworzysz i wdrożysz niestandardowy serwer MCP, który łączy interfejs wiersza poleceń Gemini z Google Cloud Storage.

W ramach ćwiczenia:

  • Kompilowanie serwera MCP opartego na Pythonie
  • Konteneryzacja aplikacji
  • wdrożyć go w Cloud Run;
  • Zabezpieczanie go za pomocą IAM i tokenów tożsamości
  • Połącz go z interfejsem wiersza poleceń Gemini
  • Wykonywanie operacji GCS na żywo za pomocą języka naturalnego

Czego się nauczysz

  • Czym jest MCP (Model Context Protocol) i jak działa
  • Jak tworzyć funkcje wywoływania narzędzi za pomocą Pythona
  • Wdrażanie aplikacji skonteneryzowanych w Cloud Run
  • Jak interfejs wiersza poleceń Gemini integruje się z zewnętrznymi serwerami MCP
  • Jak bezpiecznie uwierzytelniać usługi Cloud Run
  • Wykonywanie rzeczywistych operacji w Google Cloud Storage za pomocą AI

Czego potrzebujesz

  • przeglądarki Chrome,
  • konto Gmail,
  • Projekt Google Cloud z włączonymi płatnościami
  • Interfejs wiersza poleceń Gemini (jest już zainstalowany w Google Cloud Shell)
  • Podstawowa znajomość Pythona i Google Cloud

To ćwiczenie wymaga podstawowej znajomości Pythona.

2. Zanim zaczniesz

Utwórz projekt

  1. W konsoli Google Cloud na stronie selektora projektów wybierz lub utwórz projekt Google Cloud.
  2. Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie są włączone płatności .
  3. Będziesz używać Cloud Shell, czyli środowiska wiersza poleceń działającego w Google Cloud, które jest wstępnie załadowane narzędziem bq. U góry konsoli Google Cloud kliknij Aktywuj Cloud Shell.

Obraz przycisku aktywowania Cloud Shell

  1. Po połączeniu z Cloud Shell sprawdź, czy uwierzytelnianie zostało już przeprowadzone, a projekt jest już ustawiony na Twój identyfikator projektu, używając tego polecenia:
gcloud auth list
  1. Aby potwierdzić, że polecenie gcloud zna Twój projekt, uruchom w Cloud Shell to polecenie:
gcloud config list project
  1. Jeśli projekt nie jest ustawiony, użyj tego polecenia, aby go ustawić:
gcloud config set project <YOUR_PROJECT_ID>
  1. Włącz wymagane interfejsy API za pomocą polecenia pokazanego poniżej. Może to potrwać kilka minut, więc zachowaj cierpliwość.
gcloud services enable \
  run.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com

Jeśli pojawi się pytanie o autoryzację, kliknij Autoryzuj, aby kontynuować.

5e681903144bdfbe.png

Po pomyślnym wykonaniu polecenia powinien wyświetlić się komunikat podobny do tego poniżej:

Operation "operations/..." finished successfully.

Jeśli pominiesz jakiś interfejs API, możesz go włączyć w trakcie wdrażania.

Informacje o poleceniach gcloud i ich użyciu znajdziesz w dokumentacji.

Przygotowywanie projektu w Pythonie

W tej sekcji utworzysz projekt w Pythonie, który będzie hostować serwer MCP, i skonfigurujesz jego zależności na potrzeby wdrożenia w Cloud Run.

Tworzenie katalogu projektu

Zacznij od utworzenia nowego folderu o nazwie mcp-on-cloudrun, w którym będziesz przechowywać kod źródłowy:

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

Tworzenie pliku requirements.txt

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

Dodaj do pliku te wiersze:

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

Zapisz plik.

3. Tworzenie serwera MCP

W tej sekcji utworzysz serwer MCP, który udostępnia operacje Google Cloud Storage jako narzędzia, które można wywoływać.

Ten serwer:

  • Rejestrowanie narzędzi MCP
  • Łączenie z Google Cloud Storage
  • Uruchamianie przez HTTP
  • możliwość wdrożenia w Cloud Run,

Teraz utwórzmy podstawową logikę MCP w main.py.

Poniżej znajdziesz pełny kod, który definiuje wiele narzędzi do zarządzania Google Cloud Storage – od wyświetlania i tworzenia zasobników po przesyłanie, pobieranie i zarządzanie obiektami blob.

Tworzenie głównego pliku aplikacji

W katalogu mcp-on-cloudrun utwórz nowy plik o nazwie main.py:

touch main.py

Otwórz plik za pomocą edytora Cloud Shell:

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

Dodaj do pliku main.py to źródło:

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,
      )
  )

Po dodaniu kodu zapisz plik.

Struktura projektu powinna teraz wyglądać tak:

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

Krótkie wyjaśnienie kodu:

Importowanie i konfiguracja:

Kod zaczyna się od zaimportowania niezbędnych bibliotek.

  • Biblioteki standardowe: asyncio do wykonywania asynchronicznego, logging do wyświetlania komunikatów o stanie i os do zmiennych środowiskowych.
  • FastMCP podstawowa platforma używana do tworzenia serwera Model Context Protocol.
  • Google Cloud Storage: importowana jest biblioteka google.cloud.storage do interakcji z GCS oraz biblioteka exceptions do obsługi błędów.

Inicjowanie:

Konfigurujemy format logowania, aby ułatwić debugowanie i śledzenie tożsamości serwera. Dodatkowo konfigurujemy instancję FastMCP o nazwie MyEnhancedGCSMCPServer. Ten obiekt (mcp) będzie używany do rejestrowania wszystkich narzędzi (funkcji) udostępnianych przez serwer. Definiujemy te narzędzia:

  • list_gcs_buckets: pobiera listę wszystkich zasobników pamięci w powiązanym projekcie w chmurze Google.
  • create_bucket: Tworzy nowy zasobnik o określonej nazwie i lokalizacji.
  • delete_bucket: usuwa istniejący zasobnik.
  • list_objects: wyświetla listę wszystkich plików (blobów) w określonym zasobniku.
  • delete_blob: usuwa z zasobnika jeden konkretny plik.
  • get_bucket_metadata: zwraca szczegóły techniczne dotyczące zasobnika (lokalizacja, klasa pamięci masowej, obsługa wersji, czas utworzenia).
  • get_blob_metadata: zwraca szczegóły techniczne dotyczące konkretnego pliku (rozmiar, typ treści, skrót MD5, ostatnia aktualizacja).

Punkt wejścia:

Określa port. Jeśli nie zostanie ustawiony, domyślna wartość to 8080. Następnie używa asyncio.run(), aby asynchronicznie uruchomić serwer za pomocą mcp.run_async. Na koniec konfiguruje serwer do działania przez HTTP (host 0.0.0.0), dzięki czemu jest on dostępny dla przychodzących żądań sieciowych.

4. Konteneryzacja serwera MCP

W tej sekcji utworzysz plik Dockerfile, aby serwer MCP można było wdrożyć w Cloud Run.

Cloud Run wymaga skonteneryzowanej aplikacji. Określisz, jak aplikacja ma być zbudowana i uruchomiona.

Utwórz plik Dockerfile

Utwórz nowy plik o nazwie Dockerfile:

touch Dockerfile

Otwórz go w edytorze Cloud Shell:

cloudshell edit ~/gcs-mcp-server/Dockerfile

Dodawanie konfiguracji Dockera

Wklej do pliku Dockerfile tę treść:

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"]

Po dodaniu treści zapisz plik. Struktura projektu powinna teraz wyglądać tak:

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

5. Wdrażanie w Cloud Run

Teraz możesz wdrożyć serwer MCP bezpośrednio ze źródła.

Uruchom to polecenie w Cloud Shell:

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

Gdy pojawi się odpowiedni komunikat,

  • Zezwalać na nieuwierzytelnione wywołania? → Nie

Cloud Build:

  • Kompilowanie obrazu kontenera
  • Przesyłanie go do Artifact Registry
  • wdrożyć go w Cloud Run;

Aby potwierdzić, że repozytorium Artifact Registry może zostać utworzone, wpisz 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

Po pomyślnym wdrożeniu zobaczysz komunikat o powodzeniu z adresem URL usługi Cloud Run.

Wdrożenie możesz też sprawdzić w konsoli Google Cloud w sekcji Cloud Run → Services.

53f95a2aa7a169d6.png

6. Konfiguracja interfejsu wiersza poleceń Gemini

Do tej pory skompilowaliśmy i wdrożyliśmy serwer MCP w Cloud Run.

Teraz czas na przyjemniejszą część – połączenie go z interfejsem wiersza poleceń Gemini i przekształcenie promptów w języku naturalnym w rzeczywiste działania w chmurze.

Udzielanie uprawnień wywołującemu Cloud Run

Ponieważ nasza usługa Cloud Run jest prywatna, uwierzytelnimy się za pomocą tokena tożsamości i przypiszemy odpowiednie uprawnienia IAM.

Usługa została wdrożona za pomocą --no-allow-unauthenticated, dlatego musisz przyznać uprawnienia do jej wywoływania.

Ustaw identyfikator projektu:

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

Przyznaj sobie rolę Wywołujący Cloud Run:

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

Dzięki temu Twoje konto może bezpiecznie wywoływać usługę Cloud Run.

Generowanie tokena tożsamości

Cloud Run wymaga tokena tożsamości do uwierzytelnionego dostępu.

Wygeneruj:

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

Sprawdź:

echo $PROJECT_NUMBER
echo $ID_TOKEN

Użyjesz tego tokena w konfiguracji interfejsu wiersza poleceń Gemini.

Konfigurowanie serwera MCP w interfejsie wiersza poleceń Gemini

Otwórz plik ustawień interfejsu wiersza poleceń Gemini:

cloudshell edit ~/.gemini/settings.json

Dodaj tę konfigurację:

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

Sprawdzanie serwerów MCP skonfigurowanych w interfejsie wiersza poleceń Gemini

Uruchom interfejs wiersza poleceń Gemini w terminalu Cloud Shell za pomocą tego polecenia:

gemini

Pojawią się te dane wyjściowe:

193224319056d340.png

W interfejsie wiersza poleceń Gemini uruchom:

/mcp refresh
/mcp list

Powinien pojawić się komunikat o zarejestrowaniu gcs-cloudrun-server. Przykładowy zrzut ekranu znajdziesz poniżej:

726738c48290fc30.png

7. Wywoływanie operacji Google Storage za pomocą języka naturalnego

Utwórz zasobnik

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

Wyświetli się prośba o zezwolenie na wywołanie narzędzia create_bucket z serwera MCP.

5ab2225295285077.png

Kliknij Zezwól raz, a zasobnik zostanie utworzony w wybranym regionie.

Wyświetlanie listy zasobników

Aby wyświetlić listę zasobników, wpisz ten prompt:

List all my GCS buckets

Usuń zasobnik

Aby usunąć zasobnik, wpisz poniższy prompt (zastąp <your_bucket_name> nazwą zasobnika):

Delete the bucket <your_bucket_name>

Pobieranie metadanych zasobnika

Aby uzyskać metadane zasobnika, wpisz poniższy prompt (zastąp <your_bucket_name> nazwą zasobnika):

Give me metadata of the <your_bucket_name>

8. Czyszczenie danych

Zanim zdecydujesz się usunąć projekt Google Cloud, przeczytaj całą tę sekcję, ponieważ tego działania zwykle nie można cofnąć.

Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby użyte w tym ćwiczeniu w Codelabs, wykonaj te czynności:

  • W konsoli Google Cloud otwórz stronę Zarządzanie zasobami.
  • Z listy projektów wybierz projekt do usunięcia.
  • Kliknij Usuń.

W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby trwale usunąć projekt.

Usunięcie projektu powoduje zaprzestanie naliczania opłat za wszystkie zasoby wykorzystywane w ramach tego projektu, w tym za usługi Cloud Run i obrazy kontenerów przechowywane w Artifact Registry.

Jeśli chcesz zachować projekt, ale usunąć wdrożoną usługę:

  1. W konsoli Google Cloud otwórz Cloud Run.
  2. Wybierz usługę gcs-mcp-server.
  3. Aby usunąć usługę, kliknij Usuń.

Możesz też wpisać to polecenie gcloud w terminalu Cloud Shell.

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

9. Podsumowanie

🎉 Gratulacje! Właśnie udało Ci się utworzyć pierwszy przepływ pracy w chmurze oparty na AI.

Wdrożono:

  • Niestandardowy serwer MCP oparty na Pythonie
  • Funkcje wywoływania narzędzi w Google Cloud Storage
  • Konteneryzacja za pomocą Dockera
  • Bezpieczne wdrażanie w Cloud Run
  • Uwierzytelnianie na podstawie tożsamości
  • Integracja z interfejsem wiersza poleceń Gemini

Możesz teraz rozszerzyć tę architekturę, aby obsługiwała dodatkowe usługi Google Cloud, takie jak BigQuery, Pub/Sub czy Compute Engine.

Ten wzorzec pokazuje, jak systemy AI mogą bezpiecznie wchodzić w interakcje z infrastrukturą w chmurze za pomocą strukturalnego wywoływania narzędzi.