Transformer le langage naturel en actions Google Cloud réelles avec Gemini et MCP sur Cloud Run

1. Introduction

Cet atelier vous explique comment créer un serveur MCP (Model Context Protocol) personnalisé à l'aide de Python, le déployer sur Google Cloud Run et le connecter à Gemini CLI pour exécuter des opérations Google Cloud Storage réelles à l'aide du langage naturel.

Flux architectural : Gemini CLI → Cloud Run → MCP

e149713a547f4157.png

Imaginez que vous ouvrez votre terminal et que vous saisissez une invite simple dans un agent d'IA, comme celles présentées ci-dessous :

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

En quelques secondes, votre cloud écoute et exécute. Pas de commandes compliquées. Pas de changement d'onglet sans fin. Juste du langage simple qui se transforme en actions cloud réelles.

Objectifs de l'atelier

Vous allez créer et déployer un serveur MCP personnalisé qui connecte Gemini CLI à Google Cloud Storage.

Vous découvrirez comment :

  • Créer un serveur MCP basé sur Python
  • Conteneuriser l'application
  • La déployer sur Cloud Run
  • La sécuriser à l'aide d'IAM et de jetons d'identité
  • La connecter à Gemini CLI
  • Exécuter des opérations GCS en direct à l'aide du langage naturel

Points abordés

  • Qu'est-ce que le protocole MCP (Model Context Protocol) et comment il fonctionne
  • Comment créer des fonctionnalités d'appel d'outils à l'aide de Python
  • Comment déployer des applications conteneurisées sur Cloud Run
  • Comment Gemini CLI s'intègre aux serveurs MCP externes
  • Comment authentifier de manière sécurisée les services Cloud Run
  • Comment exécuter des opérations Google Cloud Storage réelles à l'aide de l'IA

Prérequis

  • Navigateur Web Chrome
  • Un compte Gmail
  • Un projet Google Cloud avec la facturation activée
  • Gemini CLI (préinstallé avec Google Cloud Shell)
  • Connaissances de base de Python et de Google Cloud

Cet atelier part du principe que l'utilisateur connaît les bases de Python.

2. Avant de commencer

Créer un projet

  1. Dans la console Google Cloud, sur la page du sélecteur de projet, sélectionnez ou créez un projet Google Cloud.
  2. Assurez-vous que la facturation est activée pour votre projet Cloud. Découvrez comment vérifier si la facturation est activée pour un projet .
  3. Vous allez utiliser Cloud Shell, un environnement de ligne de commande exécuté dans Google Cloud et fourni avec bq. Cliquez sur Activer Cloud Shell en haut de la console Google Cloud.

Image du bouton &quot;Activer Cloud Shell&quot;

  1. Une fois connecté à Cloud Shell, vérifiez que vous êtes déjà authentifié et que le projet est défini sur votre ID de projet à l'aide de la commande suivante :
gcloud auth list
  1. Exécutez la commande suivante dans Cloud Shell pour vérifier que la commande gcloud reconnaît votre projet.
gcloud config list project
  1. Si votre projet n'est pas défini, utilisez la commande suivante pour le définir :
gcloud config set project <YOUR_PROJECT_ID>
  1. Activez les API requises à l'aide de la commande indiquée ci-dessous. Cela peut prendre quelques minutes. Veuillez patienter.
gcloud services enable \
  run.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com

Si vous êtes invité à autoriser, cliquez sur Autoriser pour continuer.

5e681903144bdfbe.png

Si la commande s'exécute correctement, un message semblable à celui ci-dessous s'affiche :

Operation "operations/..." finished successfully.

Si une API est manquante, vous pouvez toujours l'activer au cours de l'implémentation.

Consultez la documentation pour connaître les commandes gcloud ainsi que leur utilisation.

Préparer votre projet Python

Dans cette section, vous allez créer le projet Python qui hébergera votre serveur MCP et configurer ses dépendances pour le déploiement sur Cloud Run.

Créer le répertoire du projet

Commencez par créer un dossier nommé mcp-on-cloudrun pour stocker votre code source :

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

Créer le fichier requirements.txt

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

Ajoutez le contenu suivant au fichier :

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

Enregistrez le fichier.

3. Créer le serveur MCP

Dans cette section, vous allez créer le serveur MCP qui expose les opérations Google Cloud Storage en tant qu'outils appelables.

Ce serveur :

  • Enregistrera les outils MCP
  • Se connectera à Google Cloud Storage
  • S'exécutera sur HTTP
  • Pourra être déployé sur Cloud Run

Nous allons maintenant créer notre logique MCP de base dans main.py.

Vous trouverez ci-dessous le code complet qui définit plusieurs outils pour gérer Google Cloud Storage, de la liste et de la création de buckets à l'importation, au téléchargement et à la gestion des objets blob.

Créer le fichier d'application principal

Dans le répertoire mcp-on-cloudrun, créez un fichier nommé main.py :

touch main.py

Ouvrez le fichier à l'aide de l'éditeur Cloud Shell :

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

Ajoutez la source suivante au contenu du fichier 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,
      )
  )

Enregistrez le fichier après avoir ajouté le code.

La structure de votre projet devrait maintenant se présenter comme suit :

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

Examinons brièvement le code :

Importations et configuration :

Le code commence par importer les bibliothèques nécessaires.

  • Bibliothèques standards : asyncio pour l'exécution asynchrone, logging pour la sortie des messages d'état et os pour les variables d'environnement.
  • FastMCP : framework de base utilisé pour créer le serveur Model Context Protocol.
  • Google Cloud Storage : la bibliothèque google.cloud.storage est importée pour interagir avec GCS, ainsi que exceptions pour la gestion des erreurs.

Initialisation:

Nous configurons le format de journalisation pour faciliter le débogage et le suivi de l'identité du serveur. Nous configurons également une instance de FastMCP nommée MyEnhancedGCSMCPServer. Cet objet (mcp) sera utilisé pour enregistrer tous les outils (fonctions) exposés par le serveur. Nous définissons les outils suivants :

  • list_gcs_buckets : récupère la liste de tous les buckets de stockage du projet Google Cloud associé.
  • create_bucket : crée un bucket avec un nom et un emplacement spécifiques.
  • delete_bucket : supprime un bucket existant.
  • list_objects : liste tous les fichiers (objets blob) d'un bucket spécifique.
  • delete_blob : supprime un fichier spécifique d'un bucket.
  • get_bucket_metadata : renvoie des détails techniques sur un bucket (emplacement, classe de stockage, état de la gestion des versions, heure de création).
  • get_blob_metadata : renvoie des détails techniques sur un fichier spécifique (taille, type de contenu, hachage MD5, dernière mise à jour).

Point d'entrée :

Cette configuration définit le port, qui est défini par défaut sur 8080 s'il n'est pas défini. Elle utilise ensuite asyncio.run() pour démarrer le serveur de manière asynchrone avec mcp.run_async. Enfin, elle configure le serveur pour qu'il s'exécute sur HTTP (host 0.0.0.0), ce qui le rend accessible aux requêtes réseau entrantes.

4. Conteneuriser le serveur MCP

Dans cette section, vous allez créer un Dockerfile afin que votre serveur MCP puisse être déployé sur Cloud Run.

Cloud Run nécessite une application conteneurisée. Vous allez définir comment votre application est créée et démarrée.

Créer le Dockerfile

Créez un fichier nommé Dockerfile :

touch Dockerfile

Ouvrez-le dans l'éditeur Cloud Shell :

cloudshell edit ~/gcs-mcp-server/Dockerfile

Ajouter la configuration Docker

Collez le contenu suivant dans le 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"]

Enregistrez le fichier après avoir ajouté le contenu. La structure de votre projet devrait maintenant se présenter comme suit :

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

5. Déployer sur Cloud Run

Déployez maintenant votre serveur MCP directement à partir de la source.

Exécutez la commande suivante dans Cloud Shell :

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

Lorsque vous êtes invité à

  • Autoriser les appels non authentifiés ? → Non

Cloud Build :

  • Créera l'image de conteneur
  • La transférera vers Artifact Registry
  • La déploiera sur Cloud Run

Saisissez Y pour confirmer que le dépôt Artifact Registry peut être créé.

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

Une fois le déploiement réussi, un message de réussite s'affiche avec l'URL du service Cloud Run.

Vous pouvez également vérifier le déploiement à partir de votre console Google Cloud sous Cloud Run → Services.

53f95a2aa7a169d6.png

6. Configuration de Gemini CLI

Jusqu'à présent, nous avons créé et déployé notre serveur MCP sur Cloud Run.

Il est maintenant temps de passer à la partie amusante : le connecter à Gemini CLI et transformer vos invites en langage naturel en actions cloud réelles.

Accorder l'autorisation de demandeur Cloud Run

Étant donné que notre service Cloud Run est privé, nous allons nous authentifier à l'aide d'un jeton d'identité et attribuer les autorisations IAM appropriées.

Nous avons déployé le service avec --no-allow-unauthenticated. Vous devez donc accorder l'autorisation de l'appeler.

Définissez votre ID de projet :

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

Accordez-vous le rôle Demandeur Cloud Run :

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

Cela permet à votre compte d'appeler de manière sécurisée le service Cloud Run.

Générer un jeton d'identité

Cloud Run nécessite un jeton d'identité pour l'accès authentifié.

Générez-en un :

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

Vérifiez-le :

echo $PROJECT_NUMBER
echo $ID_TOKEN

Vous utiliserez ce jeton dans la configuration de Gemini CLI.

Configurer le serveur MCP dans Gemini CLI

Ouvrez le fichier de paramètres de Gemini CLI :

cloudshell edit ~/.gemini/settings.json

Ajoutez la configuration suivante :

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

Valider les serveurs MCP configurés dans Gemini CLI

Lancez Gemini CLI dans le terminal Cloud Shell à l'aide de la commande suivante :

gemini

La sortie ci-dessous s'affiche :

193224319056d340.png

Dans Gemini CLI, exécutez :

/mcp refresh
/mcp list

Votre gcs-cloudrun-server devrait maintenant être enregistré. Vous trouverez ci-dessous un exemple de capture d'écran :

726738c48290fc30.png

7. Appeler des opérations Google Storage via le langage naturel

Créer un bucket

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

Vous serez invité à demander l'autorisation d'appeler l'outil create_bucket à partir du serveur MCP.

5ab2225295285077.png

Cliquez sur Autoriser une fois, puis votre bucket sera créé dans la région spécifique que vous avez demandée.

Répertorier les buckets

Pour répertorier les buckets, saisissez l'invite ci-dessous :

List all my GCS buckets

Supprimer un bucket

Pour supprimer un bucket, saisissez l'invite ci-dessous (remplacez <your_bucket_name> par le nom de votre bucket) :

Delete the bucket <your_bucket_name>

Obtenir les métadonnées du bucket

Pour obtenir les métadonnées d'un bucket, saisissez l'invite ci-dessous (remplacez <your_bucket_name> par le nom de votre bucket) :

Give me metadata of the <your_bucket_name>

8. Effectuer un nettoyage

Lisez d'abord l'intégralité de cette section avant de décider de supprimer le projet Google Cloud, car cette action n'est généralement pas réversible.

Pour éviter que les ressources utilisées dans cet atelier soient facturées sur votre compte Google Cloud, procédez comme suit :

  • Dans Google Cloud Console, accédez à la page "Gérer les ressources".
  • Dans la liste des projets, sélectionnez celui que vous souhaitez supprimer.
  • Cliquez sur Supprimer.

Dans la boîte de dialogue, saisissez l'ID du projet, puis cliquez sur Arrêter pour supprimer définitivement le projet.

La suppression du projet arrête la facturation de toutes les ressources utilisées dans ce projet, y compris les services Cloud Run et les images de conteneurs stockées dans Artifact Registry.

Si vous souhaitez conserver le projet, mais supprimer le service déployé :

  1. Accédez à Cloud Run dans la console Google Cloud.
  2. Sélectionnez le service gcs-mcp-server.
  3. Cliquez sur Supprimer pour supprimer le service.

Vous pouvez également exécuter la commande gcloud suivante dans le terminal Cloud Shell.

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

9. Conclusion

🎉 Félicitations ! Vous venez de créer votre premier workflow cloud basé sur l'IA.

Vous avez implémenté :

  • Un serveur MCP personnalisé basé sur Python
  • Des fonctionnalités d'appel d'outils pour Google Cloud Storage
  • La conteneurisation à l'aide de Docker
  • Un déploiement sécurisé sur Cloud Run
  • L'authentification basée sur l'identité
  • L'intégration à Gemini CLI

Vous pouvez désormais étendre cette architecture pour prendre en charge d'autres services Google Cloud tels que BigQuery, Pub/Sub ou Compute Engine.

Ce modèle montre comment les systèmes d'IA peuvent interagir de manière sécurisée avec l'infrastructure cloud via l'appel d'outils structurés.