Mit Gemini + MCP in Cloud Run natürliche Sprache in echte Google Cloud-Aktionen umwandeln

1. Einführung

In diesem Codelab erfahren Sie, wie Sie mit Python einen benutzerdefinierten MCP-Server (Model Context Protocol) erstellen, ihn in Google Cloud Run bereitstellen und ihn mit der Gemini CLI verbinden, um mit natürlicher Sprache echte Google Cloud Storage-Vorgänge auszuführen.

Architekturablauf:Gemini CLI → Cloud Run → MCP

Stellen Sie sich Folgendes vor: Sie öffnen Ihr Terminal und geben einen einfachen Prompt in einen KI‑Agenten ein, wie die unten gezeigten:

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

Innerhalb von Sekunden hört Ihre Cloud zu und führt den Befehl aus. Keine komplizierten Befehle. Kein endloses Wechseln zwischen Tabs Einfache Sprache wird in echte Cloud-Aktionen umgewandelt.

Aufgaben

Sie erstellen und stellen einen benutzerdefinierten MCP-Server bereit, der die Gemini CLI mit Google Cloud Storage verbindet.

Sie werden Folgendes tun:

• Python-basierten MCP-Server erstellen
• Anwendung containerisieren
• In Cloud Run bereitstellen
• Mit IAM und Identitätstokens sichern
• Mit der Gemini CLI verbinden
• GCS-Live-Vorgänge mit natürlicher Sprache ausführen

Lerninhalte

• Was das MCP (Model Context Protocol) ist und wie es funktioniert
• Toolaufruffunktionen mit Python erstellen
• Containeranwendungen in Cloud Run bereitstellen
• Integration der Gemini CLI in externe MCP-Server
• Cloud Run-Dienste sicher authentifizieren
• Echte Google Cloud Storage-Vorgänge mit KI ausführen

Voraussetzungen

• Chrome-Webbrowser
• Ein Gmail-Konto
• Ein Google Cloud-Projekt mit aktivierter Abrechnung
• Gemini CLI (in Google Cloud Shell vorinstalliert)
• Grundkenntnisse in Python und Google Cloud

In diesem Codelab wird davon ausgegangen, dass der Nutzer über grundlegende Python-Kenntnisse verfügt.

2. Hinweis

Projekt erstellen

1. Wählen Sie in der Google Cloud Console auf der Seite zur Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
2. Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist .
3. Sie verwenden Cloud Shell, eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und in der bq vorinstalliert ist. Klicken Sie oben in der Google Cloud Console auf „Cloud Shell aktivieren“.

4. Sobald die Verbindung mit der Cloud Shell hergestellt ist, prüfen Sie mit dem folgenden Befehl, ob Sie bereits authentifiziert sind und für das Projekt schon Ihre Projekt-ID eingestellt ist:

gcloud auth list

5. Führen Sie den folgenden Befehl in Cloud Shell aus, um zu bestätigen, dass der gcloud-Befehl Ihr Projekt kennt.

gcloud config list project

6. Wenn Ihr Projekt nicht festgelegt ist, verwenden Sie den folgenden Befehl, um es festzulegen:

gcloud config set project <YOUR_PROJECT_ID>

7. Aktivieren Sie die erforderlichen APIs mit dem unten gezeigten Befehl. Dies kann einige Minuten dauern.

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

Wenn Sie zur Autorisierung aufgefordert werden, klicken Sie auf „Autorisieren“, um fortzufahren.

Bei erfolgreicher Ausführung des Befehls sollte eine Meldung wie die unten gezeigte angezeigt werden:

Operation "operations/..." finished successfully.

Wenn eine API fehlt, können Sie sie jederzeit während der Implementierung aktivieren.

Informationen zu gcloud-Befehlen und deren Verwendung finden Sie in der Dokumentation.

Python-Projekt vorbereiten

In diesem Abschnitt erstellen Sie das Python-Projekt, in dem Ihr MCP-Server gehostet wird, und konfigurieren seine Abhängigkeiten für die Bereitstellung in Cloud Run.

Projektverzeichnis erstellen

Erstellen Sie zuerst einen neuen Ordner mit dem Namen mcp-on-cloudrun, in dem Sie Ihren Quellcode speichern:

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

requirements.txt erstellen

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

Fügen Sie der Datei den folgenden Inhalt hinzu.

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

Speichern Sie die Datei.

3. MCP-Server erstellen

In diesem Abschnitt erstellen Sie den MCP-Server, der Google Cloud Storage-Vorgänge als aufrufbare Tools bereitstellt.

Dieser Server:

• MCP-Tools registrieren
• Mit Google Cloud Storage verbinden
• Über HTTP ausführen
• In Cloud Run bereitstellbar sein

Erstellen wir nun die MCP-Kernlogik in main.py.

Unten finden Sie den vollständigen Code, der mehrere Tools zum Verwalten von Google Cloud Storage definiert – vom Auflisten und Erstellen von Buckets bis zum Hochladen, Herunterladen und Verwalten von Blobs.

Hauptanwendungsdatei erstellen

Erstellen Sie im Verzeichnis mcp-on-cloudrun eine neue Datei mit dem Namen main.py:

touch main.py

Öffnen Sie die Datei mit dem Cloud Shell-Editor:

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

Fügen Sie dem Inhalt der Datei main.py die folgende Quelle hinzu:

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

Speichern Sie die Datei, nachdem Sie den Code hinzugefügt haben.

Die Projektstruktur sollte nun so aussehen:

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

Hier eine kurze Erläuterung des Codes:

Importe und Einrichtung:

Der Code beginnt mit dem Importieren der erforderlichen Bibliotheken.

• Standardbibliotheken: asyncio für die asynchrone Ausführung, logging für die Ausgabe von Statusmeldungen und os für Umgebungsvariablen.
• FastMCP: Das Kern-Framework, das zum Erstellen des Model Context Protocol-Servers verwendet wird.
• Google Cloud Storage: Die google.cloud.storage-Bibliothek wird importiert, um mit GCS zu interagieren, zusammen mit exceptions für die Fehlerbehandlung.

Initialisierung:

Wir konfigurieren das Logging-Format, um die Identität des Servers zu debuggen und nachzuverfolgen. Außerdem konfigurieren wir eine Instanz von FastMCP mit dem Namen MyEnhancedGCSMCPServer. Dieses Objekt (mcp) wird verwendet, um alle Tools (Funktionen) zu registrieren, die der Server bereitstellt. Wir definieren die folgenden Tools:

• list_gcs_buckets: Ruft eine Liste aller Storage-Buckets im zugehörigen Google Cloud-Projekt ab.
• create_bucket: Erstellt einen neuen Bucket mit einem bestimmten Namen und Speicherort.
• delete_bucket: Löscht einen vorhandenen Bucket.
• list_objects: Listet alle Dateien (Blobs) in einem bestimmten Bucket auf.
• delete_blob: Löscht eine einzelne bestimmte Datei aus einem Bucket.
• get_bucket_metadata: Gibt technische Details zu einem Bucket zurück (Speicherort, Speicherklasse, Versionierungsstatus, Erstellungszeit).
• get_blob_metadata: Gibt technische Details zu einer bestimmten Datei zurück (Größe, Inhaltstyp, MD5-Hash, zuletzt aktualisiert).

Einstiegspunkt:

Damit wird der Port konfiguriert. Wenn kein Wert festgelegt ist, wird standardmäßig 8080 verwendet. Anschließend wird asyncio.run() verwendet, um den Server asynchron mit mcp.run_async zu starten. Schließlich wird der Server für die Ausführung über HTTP (host 0.0.0.0) konfiguriert, sodass er für eingehende Netzwerkanfragen zugänglich ist.

4. MCP-Server containerisieren

In diesem Abschnitt erstellen Sie ein Dockerfile, damit Ihr MCP-Server in Cloud Run bereitgestellt werden kann.

Für Cloud Run ist eine Containeranwendung erforderlich. Sie legen fest, wie Ihre Anwendung erstellt und gestartet wird.

Dockerfile erstellen

Erstellen Sie eine neue Datei mit dem Namen Dockerfile:

touch Dockerfile

Öffnen Sie die Datei im Cloud Shell-Editor:

cloudshell edit ~/gcs-mcp-server/Dockerfile

Docker-Konfiguration hinzufügen

Fügen Sie Folgendes in das Dockerfile ein:

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

Speichern Sie die Datei, nachdem Sie den Inhalt hinzugefügt haben. Die Projektstruktur sollte nun so aussehen:

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

5. In Cloud Run bereitstellen

Stellen Sie Ihren MCP-Server jetzt direkt aus der Quelle bereit.

Führen Sie in Cloud Shell den folgenden Befehl aus:

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

Wenn Sie dazu aufgefordert werden

• Nicht authentifizierte Aufrufe zulassen? → Nein

Cloud Build führt folgende Aktionen aus:

• Container-Image erstellen
• Per Push in Artifact Registry übertragen
• In Cloud Run bereitstellen

Geben Sie Y ein, um zu bestätigen, dass das Artifact Registry-Repository erstellt werden kann.

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

⚠️ Hinweis: Verwenden Sie in Produktionsumgebungen immer das Flag --no-allow-unauthenticated, um unbefugten Zugriff zu verhindern.Im nächsten Schritt authentifizieren Sie die Gemini CLI mit einem Identitätstoken.

Nach erfolgreicher Bereitstellung wird eine Erfolgsmeldung mit der Cloud Run-Dienst-URL angezeigt.

Sie können die Bereitstellung auch in der Google Cloud Console unter Cloud Run → Services überprüfen.

6. Gemini CLI konfigurieren

Bisher haben wir unseren MCP-Server in Cloud Run erstellt und bereitgestellt.

Jetzt kommt der spannende Teil – die Verbindung mit der Gemini CLI und die Umwandlung Ihrer Prompts in natürlicher Sprache in echte Cloud-Aktionen.

Cloud Run-Aufruferberechtigung erteilen

Da unser Cloud Run-Dienst privat ist, authentifizieren wir uns mit einem Identitätstoken und weisen die richtigen IAM-Berechtigungen zu.

Wir haben den Dienst mit --no-allow-unauthenticated bereitgestellt. Sie müssen daher die Berechtigung zum Aufrufen des Dienstes erteilen.

Legen Sie Ihre Projekt-ID fest:

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

Weisen Sie sich selbst die Rolle Cloud Run Invoker zu:

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

So kann Ihr Konto den Cloud Run-Dienst sicher aufrufen.

Identitätstoken generieren

Für den authentifizierten Zugriff auf Cloud Run ist ein Identitätstoken erforderlich.

Einen erstellen:

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

So überprüfen Sie es:

echo $PROJECT_NUMBER
echo $ID_TOKEN

Sie verwenden dieses Token in der Gemini CLI-Konfiguration.

MCP-Server in der Gemini CLI konfigurieren

Öffnen Sie die Einstellungsdatei für die Gemini CLI:

cloudshell edit ~/.gemini/settings.json

Fügen Sie die folgende Konfiguration hinzu:

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

In der Gemini CLI konfigurierte MCP-Server validieren

Starten Sie die Gemini CLI im Cloud Shell-Terminal mit dem folgenden Befehl:

gemini

Die folgende Ausgabe wird angezeigt:

Führen Sie in der Gemini CLI Folgendes aus:

/mcp refresh
/mcp list

Sie sollten jetzt sehen, dass Ihre gcs-cloudrun-serve registriert wurde. Hier ein Beispiel-Screenshot:

7. Google Storage-Vorgänge über natürliche Sprache aufrufen

⚠️ Denken Sie daran, den Namen my-ai-bucket im Prompt unten durch einen global eindeutigen Bucket-Namen zu ersetzen, da Google Cloud Storage nur global eindeutige Namen zulässt. Wenn Sie eine Fehlermeldung erhalten, dass der Bucket bereits vorhanden ist, versuchen Sie es mit einem anderen Namen.

Bucket erstellen

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

Sie werden dann aufgefordert, die Berechtigung zum Aufrufen des Tools „create_bucket“ vom MCP-Server zu erteilen.

Klicken Sie auf „Einmal zulassen“. Ihr Bucket wird dann in der von Ihnen angeforderten Region erstellt.

Buckets auflisten

Geben Sie den folgenden Prompt ein, um die Buckets aufzulisten:

List all my GCS buckets

Bucket löschen

Geben Sie zum Löschen eines Buckets den folgenden Prompt ein und ersetzen Sie <your_bucket_name> durch den Namen Ihres Buckets:

Delete the bucket <your_bucket_name>

Metadaten des Buckets abrufen

Geben Sie den folgenden Prompt ein, um die Metadaten eines Buckets abzurufen. Ersetzen Sie dabei <your_bucket_name> durch den Namen Ihres Buckets:

Give me metadata of the <your_bucket_name>

8. Bereinigen

Lesen Sie diesen Abschnitt vollständig, bevor Sie sich für das Löschen des Google Cloud-Projekts entscheiden, da diese Aktion in der Regel nicht rückgängig gemacht werden kann.

So vermeiden Sie, dass Ihrem Google Cloud-Konto die in diesem Codelab verwendeten Ressourcen in Rechnung gestellt werden:

• Wechseln Sie in der Google Cloud Console zur Seite „Ressourcen verwalten“.
• Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten.
• Klicken Sie auf Löschen.

Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf „Beenden“, um das Projekt endgültig zu löschen.

Wenn Sie das Projekt löschen, wird die Abrechnung für alle in diesem Projekt verwendeten Ressourcen beendet, einschließlich Cloud Run-Diensten und in Artifact Registry gespeicherten Container-Images.

Wenn Sie das Projekt beibehalten, aber den bereitgestellten Dienst entfernen möchten, gehen Sie so vor:

1. Rufen Sie in der Google Cloud Console Cloud Run auf.
2. Wählen Sie den gcs-mcp-server-Dienst aus.
3. Klicken Sie auf „Löschen“, um den Dienst zu entfernen.

oder geben Sie den folgenden gcloud-Befehl im Cloud Shell-Terminal ein.

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

9. Fazit

🎉 Das wars! Sie haben Ihren ersten KI-gestützten Cloud-Workflow erstellt.

Sie haben Folgendes implementiert:

• Ein benutzerdefinierter Python-basierter MCP-Server
• Funktionen zum Aufrufen von Tools für Google Cloud Storage
• Containerisierung mit Docker
• Sichere Bereitstellung in Cloud Run
• Identitätsbasierte Authentifizierung
• Integration mit der Gemini CLI

Sie können diese Architektur jetzt auf zusätzliche Google Cloud-Dienste wie BigQuery, Pub/Sub oder Compute Engine ausweiten.

In diesem Muster wird gezeigt, wie KI-Systeme über strukturierte Tool-Aufrufe sicher mit der Cloud-Infrastruktur interagieren können.