Trasforma il linguaggio naturale in azioni Google Cloud reali con Gemini + MCP su Cloud Run

1. Introduzione

Questo codelab ti guida nella creazione di un server MCP (Model Context Protocol) personalizzato utilizzando Python, nel deployment in Google Cloud Run e nella connessione con Gemini CLI per eseguire operazioni reali di Google Cloud Storage utilizzando il linguaggio naturale.

Flusso architetturale: Gemini CLI → Cloud Run → MCP

e149713a547f4157.png

Immagina di aprire il terminale e digitare un semplice prompt in un agente AI, come quelli mostrati di seguito:

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

In pochi secondi, il cloud ascolta ed esegue. Nessun comando complicato. Nessun cambio di scheda infinito. Solo linguaggio semplice che si trasforma in azioni cloud reali.

Attività previste

Creerai ed eseguirai il deployment di un server MCP personalizzato che connette Gemini CLI a Google Cloud Storage.

Imparerai a:

  • Creare un server MCP basato su Python
  • Containerizzare l'applicazione
  • Eseguire il deployment in Cloud Run
  • Proteggerlo utilizzando IAM e token di identità
  • Connetterlo a Gemini CLI
  • Eseguire operazioni GCS live utilizzando il linguaggio naturale

Cosa imparerai a fare

  • Che cos'è MCP (Model Context Protocol) e come funziona
  • Come creare funzionalità di chiamata di strumenti utilizzando Python
  • Come eseguire il deployment di applicazioni containerizzate in Cloud Run
  • Come Gemini CLI si integra con i server MCP esterni
  • Come autenticare in modo sicuro i servizi Cloud Run
  • Come eseguire operazioni reali di Google Cloud Storage utilizzando l'AI

Che cosa ti serve

  • Browser web Chrome
  • Un account Gmail
  • Un progetto Google Cloud con la fatturazione abilitata
  • Gemini CLI (è preinstallata con Google Cloud Shell)
  • Conoscenza di base di Python e Google Cloud

Questo codelab presuppone che l'utente abbia una conoscenza di base di Python

2. Prima di iniziare

Crea un progetto

  1. Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud.
  2. Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata per un progetto .
  3. Utilizzerai Cloud Shell, un ambiente a riga di comando in esecuzione in Google Cloud che viene fornito con bq precaricato. Fai clic su Attiva Cloud Shell nella parte superiore della console Google Cloud.

Immagine del pulsante Attiva Cloud Shell

  1. Una volta eseguita la connessione a Cloud Shell, verifica che il tuo account sia già autenticato e che il progetto sia impostato sul tuo ID progetto utilizzando il seguente comando:
gcloud auth list
  1. Esegui questo comando in Cloud Shell per verificare che il comando gcloud conosca il tuo progetto.
gcloud config list project
  1. Se il progetto non è impostato, utilizza il seguente comando per impostarlo:
gcloud config set project <YOUR_PROJECT_ID>
  1. Abilita le API richieste tramite il comando mostrato di seguito. Questa operazione potrebbe richiedere alcuni minuti, quindi pazienta.
gcloud services enable \
  run.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com

Se ti viene chiesto di autorizzare, fai clic su Autorizza per continuare.

5e681903144bdfbe.png

Al termine dell'esecuzione del comando, dovresti visualizzare un messaggio simile a quello riportato di seguito:

Operation "operations/..." finished successfully.

Se manca un'API, puoi sempre abilitarla durante il corso dell'implementazione.

Consulta la documentazione per i comandi e l'utilizzo di gcloud.

Prepara il progetto Python

In questa sezione creerai il progetto Python che ospiterà il server MCP e configurerai le relative dipendenze per il deployment in Cloud Run.

Crea la directory del progetto

Inizia creando una nuova cartella denominata mcp-on-cloudrun per archiviare il codice sorgente:

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

Crea requirements.txt

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

Aggiungi i seguenti contenuti al file:

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

Salva il file.

3. Crea il server MCP

In questa sezione creerai il server MCP che espone le operazioni di Google Cloud Storage come strumenti chiamabili.

Questo server:

  • Registra gli strumenti MCP
  • Si connette a Google Cloud Storage
  • Viene eseguito tramite HTTP
  • Può essere sottoposto a deployment in Cloud Run

Ora creiamo la logica MCP principale all'interno di main.py.

Di seguito è riportato il codice completo che definisce più strumenti per la gestione di Google Cloud Storage, dall'elenco e dalla creazione di bucket al caricamento, al download e alla gestione dei blob.

Crea il file dell'applicazione principale

Nella directory mcp-on-cloudrun, crea un nuovo file denominato main.py:

touch main.py

Apri il file utilizzando l'editor di Cloud Shell:

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

Aggiungi la seguente origine ai contenuti del file 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,
      )
  )

Salva il file dopo aver aggiunto il codice.

La struttura del progetto dovrebbe ora essere simile alla seguente:

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

Vediamo brevemente il codice:

Importazioni e configurazione:

Il codice inizia importando le librerie necessarie.

  • Librerie standard: asyncio per l'esecuzione asincrona, logging per l'output dei messaggi di stato e os per le variabili di ambiente.
  • FastMCP: il framework principale utilizzato per creare il server Model Context Protocol.
  • Google Cloud Storage: la libreria google.cloud.storage viene importata per interagire con GCS, insieme a exceptions per la gestione degli errori.

Inizializzazione:

Configuriamo il formato di logging per facilitare il debug e il monitoraggio dell'identità del server. Inoltre, configuriamo un'istanza di FastMCP denominata MyEnhancedGCSMCPServer. Questo oggetto (mcp) verrà utilizzato per registrare tutti gli strumenti (funzioni) esposti dal server. Definiamo i seguenti strumenti:

  • list_gcs_buckets: recupera un elenco di tutti i bucket di archiviazione nel progetto Google Cloud associato.
  • create_bucket: crea un nuovo bucket con un nome e una località specifici.
  • delete_bucket: elimina un bucket esistente.
  • list_objects: elenca tutti i file (blob) all'interno di un bucket specifico.
  • delete_blob: elimina un singolo file specifico da un bucket.
  • get_bucket_metadata: restituisce i dettagli tecnici di un bucket (località, classe di archiviazione, stato del controllo delle versioni, ora di creazione).
  • get_blob_metadata: restituisce i dettagli tecnici di un file specifico (dimensioni, tipo di contenuto, hash MD5, ultimo aggiornamento).

Punto di ingresso:

Questa configurazione imposta la porta, che per impostazione predefinita è 8080 se non è impostata. Quindi utilizza asyncio.run() per avviare il server in modo asincrono con mcp.run_async. Infine, configura il server per l'esecuzione tramite HTTP (host 0.0.0.0), rendendolo accessibile per le richieste di rete in entrata.

4. Containerizza il server MCP

In questa sezione creerai un Dockerfile in modo che il server MCP possa essere sottoposto a deployment in Cloud Run.

Cloud Run richiede un'applicazione containerizzata. Definirai come viene creata e avviata l'applicazione.

Crea il Dockerfile

Crea un nuovo file denominato Dockerfile:

touch Dockerfile

Aprilo nell'editor di Cloud Shell:

cloudshell edit ~/gcs-mcp-server/Dockerfile

Aggiungi la configurazione Docker

Incolla i seguenti contenuti nel 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"]

Salva il file dopo aver aggiunto i contenuti. La struttura del progetto dovrebbe ora essere simile alla seguente:

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

5. Deployment in Cloud Run

Ora esegui il deployment del server MCP direttamente dall'origine.

Esegui il comando riportato di seguito in Cloud Shell:

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

Quando ti viene chiesto di

  • Consenti chiamate non autenticate? → No

Cloud Build:

  • Crea l'immagine container
  • Esegue il push in Artifact Registry
  • Esegue il deployment in Cloud Run

Inserisci Y per confermare che il repository Artifact Registry può essere creato.

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

Al termine del deployment, vedrai un messaggio di riuscita con l'URL del servizio Cloud Run.

Puoi anche verificare il deployment dalla console Google Cloud in Cloud Run → Services.

53f95a2aa7a169d6.png

6. Configurazione di Gemini CLI

Finora abbiamo creato ed eseguito il deployment del server MCP in Cloud Run.

Ora è il momento della parte divertente: connetterlo a Gemini CLI e trasformare i prompt in linguaggio naturale in azioni cloud reali.

Concedi l'autorizzazione Cloud Run Invoker

Poiché il nostro servizio Cloud Run è privato, eseguiremo l'autenticazione utilizzando un token di identità e assegneremo le autorizzazioni IAM corrette.

Abbiamo eseguito il deployment del servizio con --no-allow-unauthenticated, pertanto devi concedere l'autorizzazione per chiamarlo.

Imposta l'ID progetto:

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

Concediti il ruolo Cloud Run Invoker:

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

In questo modo, il tuo account può chiamare in modo sicuro il servizio Cloud Run.

Genera un token di identità

Cloud Run richiede un token di identità per l'accesso autenticato.

Generane uno:

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

Verificalo:

echo $PROJECT_NUMBER
echo $ID_TOKEN

Utilizzerai questo token all'interno della configurazione di Gemini CLI.

Configura il server MCP in Gemini CLI

Apri il file di configurazione di Gemini CLI:

cloudshell edit ~/.gemini/settings.json

Aggiungi la seguente configurazione:

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

Convalida i server MCP configurati in Gemini CLI

Avvia Gemini CLI nel terminale Cloud Shell tramite il seguente comando:

gemini

Verrà visualizzato l'output riportato di seguito

193224319056d340.png

In Gemini CLI, esegui:

/mcp refresh
/mcp list

Ora dovresti vedere il server gcs-cloudrun-server registrato. Di seguito è riportato uno screenshot di esempio:

726738c48290fc30.png

7. Richiama le operazioni di Google Storage tramite il linguaggio naturale

Crea bucket

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

Ti verrà chiesto di concedere l'autorizzazione per chiamare lo strumento create_bucket dal server MCP.

5ab2225295285077.png

Fai clic su Consenti una volta, quindi il bucket verrà creato correttamente nella regione specifica che hai richiesto.

Elenca bucket

Per elencare i bucket, inserisci il prompt riportato di seguito:

List all my GCS buckets

Elimina bucket

Per eliminare un bucket, inserisci il prompt riportato di seguito (sostituisci <your_bucket_name> con il nome del bucket):

Delete the bucket <your_bucket_name>

Recupera i metadati del bucket

Per recuperare i metadati di un bucket, inserisci il prompt riportato di seguito (sostituisci <your_bucket_name> con il nome del bucket):

Give me metadata of the <your_bucket_name>

8. Libera spazio

Leggi l'intera sezione prima di decidere di eliminare il progetto Google Cloud, poiché questa azione in genere non è reversibile.

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo codelab, segui questi passaggi:

  • Nella console Google Cloud, vai alla pagina Gestisci risorse.
  • Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare.
  • Fai clic su Elimina.

Nella finestra di dialogo, digita l'ID progetto e fai clic su Arresta per eliminare definitivamente il progetto.

L'eliminazione del progetto interrompe la fatturazione per tutte le risorse utilizzate al suo interno, inclusi i servizi Cloud Run e le immagini container archiviate in Artifact Registry.

In alternativa, se vuoi conservare il progetto ma rimuovere il servizio di cui è stato eseguito il deployment:

  1. Vai a Cloud Run nella console Google Cloud.
  2. Seleziona il servizio gcs-mcp-server.
  3. Fai clic su Elimina per rimuovere il servizio.

oppure esegui il seguente comando gcloud nel terminale Cloud Shell.

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

9. Conclusione

🎉 Congratulazioni! Hai appena creato il tuo primo workflow cloud basato sull'AI.

Hai implementato:

  • Un server MCP personalizzato basato su Python
  • Funzionalità di chiamata di strumenti per Google Cloud Storage
  • Containerizzazione con Docker
  • Deployment sicuro in Cloud Run
  • Autenticazione basata sull'identità
  • Integrazione con Gemini CLI

Ora puoi estendere questa architettura per supportare altri servizi Google Cloud come BigQuery, Pub/Sub o Compute Engine.

Questo pattern mostra come i sistemi AI possono interagire in modo sicuro con l'infrastruttura cloud tramite la chiamata strutturata di strumenti.