Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Compila, protege e implementa un servidor de MCP en Google Cloud

1. Antes de comenzar

El Protocolo de contexto del modelo (MCP) es un estándar abierto que permite que los modelos y agentes de IA accedan de forma segura a herramientas, bases de datos y contexto empresarial. Desde su introducción en 2024, el protocolo ha sido ampliamente adoptado por los proveedores de LLM y de servicios en la nube. La especificación más reciente MCP Spec 2026-07-28 (MCP 2.0) define una arquitectura sin estado, transportes simplificados y un estricto tipado de resultados, a la vez que sigue siendo retrocompatible con versiones anteriores.

El SDK de Python de MCP oficial (mcp>=2.0.0) proporciona MCPServer (mcp.server.mcpserver.MCPServer), que reemplaza a FastMCP en MCP 2.0 como el framework de alto rendimiento y fácil de usar para los desarrolladores que permite crear servidores de MCP listos para la producción que cumplen con la moderna especificación de MCP (28-07-2026) con transportes de HTTP y eventos enviados por el servidor (SSE) transmitibles a través de SSL/TLS.

En este codelab, crearás un servidor de MCP de nivel de producción con MCPServer del SDK de Python de MCP 2.0 y la administración de dependencias de uv. Equiparás tu servidor de MCP con cuatro herramientas de Google Cloud (invocación de Vertex AI Gemini, inspección de Google Cloud Storage, escritura de auditoría de Cloud Logging y verificación del estado de los recursos de Google Cloud). Luego, crearás un contenedor para el servidor y lo implementarás en dos destinos de ejecución de Google Cloud: Cloud Run y Autopilot de Google Kubernetes Engine (GKE), y registrarás y usarás el servidor de MCP en Gemini Enterprise Agent Platform.

Aspectos destacados de la especificación de MCP del 28/7/2026 y el SDK v2.0:

MCPServer reemplaza a FastMCP: En el SDK oficial de mcp>=2.0.0 para Python, MCPServer (from mcp.server.mcpserver import MCPServer) reemplaza a FastMCP y conserva la API familiar basada en decoradores (@mcp.tool, @mcp.custom_route).
Arquitectura sin estado: Elimina los acuerdos de inicio de conexión con estado (initialize/initialized). Cada solicitud incluye la versión del protocolo y las capacidades del cliente en los encabezados y _meta.
Extremo de detección: Implementa server/discover para la negociación de capacidades y versiones.
Transporte Transmitible: Transporte HTTP/SSE simplificado que elimina los encabezados Mcp-Session-Id por conexión.
Typing de resultados: Todos los resultados de la ejecución de la herramienta incluyen un resultType obligatorio ("complete" o "input_required").

Actividades

Compila un MCPServer con 4 herramientas de Google Cloud usando Python 3.12 o versiones posteriores y uv que cumpla con la especificación de MCP del 28/7/2026.
Crea un contenedor para el servidor de MCP con una compilación de Docker de varias etapas.
Protege e implementa el servidor de MCP en Cloud Run con autenticación de IAM y SSL/TLS aplicadas.
Protege e implementa el servidor de MCP en GKE Autopilot con Workload Identity y la API de Kubernetes Gateway con TLS.
Registra el extremo seguro del servidor de MCP en Gemini Enterprise Agent Platform con encabezados de token de portador de OIDC.

Requisitos

Un proyecto de Google Cloud con facturación habilitada.
El SDK de Google Cloud (CLI de gcloud) está instalado y configurado.
Python 3.12 o versiones posteriores y el administrador de paquetes uv instalados
Se instaló docker.
La herramienta de línea de comandos de kubectl instalada

2. Configura el entorno de Google Cloud

Antes de crear recursos, autentica tu entorno y habilita las APIs de Google Cloud necesarias.

Autentica la CLI de gcloud

Accede a tu cuenta de Google Cloud:

gcloud auth login

Establece el ID de tu proyecto activo de Google Cloud:

export PROJECT_ID=$(gcloud config get-value project)
gcloud config set project ${PROJECT_ID}

Habilita los servicios de Google Cloud

Habilita todas las APIs requeridas para Cloud Run, GKE, Vertex AI, Artifact Registry, Cloud Build, Cloud Logging, Storage y Compute Engine:

gcloud services enable \
    agentregistry.googleapis.com \
    run.googleapis.com \
    container.googleapis.com \
    artifactregistry.googleapis.com \
    aiplatform.googleapis.com \
    logging.googleapis.com \
    storage.googleapis.com \
    compute.googleapis.com \
    iam.googleapis.com \
    cloudbuild.googleapis.com \
    --project="${PROJECT_ID}"

Verifica que las APIs se hayan habilitado correctamente:

Operation "operations/..." finished successfully.

Autentica las credenciales predeterminadas de la aplicación

Autentica tu entorno para que las bibliotecas cliente de Python puedan acceder a Vertex AI y Cloud Storage de forma local durante el desarrollo:

gcloud auth application-default login

3. Compila el servidor de MCP con MCPServer y uv

En este paso, inicializarás un proyecto de Python con uv y compilarás un MCPServer que admita cuatro capacidades de Google Cloud.

Inicializa el proyecto con uv

Crea el directorio del servidor y, luego, inicializa uv:

mkdir -p mcp-server/src/mcp_server
cd mcp-server
uv init --lib

Copia el código en el archivo pyproject.toml:

[project]
name = "secure-mcp-gcp-server"
version = "0.1.0"
description = "MCP server with Google Cloud tools supporting MCP Spec 2026-07-28 over Streamable HTTP"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "mcp>=2.0.0",
    "google-genai>=1.0.0",
    "google-cloud-storage>=2.14.0",
    "google-cloud-logging>=3.11.0",
    "google-cloud-resource-manager>=1.12.0",
    "uvicorn>=0.30.0",
    "httpx2>=0.1.0",
    "pydantic>=2.7.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/mcp_server"]

Sincroniza las dependencias con uv:

uv sync

Escribe el código de MCPServer

Crea el archivo de implementación del servidor en src/mcp_server/server.py:

import logging
import os
from typing import Any

from google import genai
from google.cloud import logging as cloud_logging
from google.cloud import storage
from mcp.server.mcpserver import MCPServer
from starlette.requests import Request
from starlette.responses import PlainTextResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-gcp-server")

# Initialize MCPServer conforming to MCP Spec 2026-07-28
mcp = MCPServer(
    "Google Cloud Production Tools",
    instructions="MCP Server conforming to MCP Spec 2026-07-28 for Vertex AI, Cloud Storage, Audit Logging, and Health Inspection.",
)


@mcp.custom_route("/healthz", methods=["GET"])
async def health_check(request: Request) -> PlainTextResponse:
    """Kubernetes readiness and liveness probe health check endpoint."""
    return PlainTextResponse("OK")


@mcp.tool(description="Generate content or answer questions using Vertex AI Gemini model.")
def vertex_ai_generate_content(
    prompt: str,
    model_name: str = "gemini-2.5-flash",
    project_id: str | None = None,
    location: str = "us-central1",
) -> str:
    """Invokes Vertex AI Gemini API using official google-genai SDK."""
    target_project = project_id or os.getenv("GCP_PROJECT") or os.getenv("GOOGLE_CLOUD_PROJECT")
    if not target_project:
        return "Error: GCP project ID not configured."

    try:
        client = genai.Client(vertexai=True, project=target_project, location=location)
        response = client.models.generate_content(
            model=model_name,
            contents=prompt,
        )
        return response.text or "No text returned from Gemini."
    except Exception as e:
        logger.error("Vertex AI Tool Error: %s", e)
        return f"Error executing Vertex AI tool: {e!s}"


@mcp.tool(description="List objects and inspect metadata for a specified Google Cloud Storage bucket.")
def gcs_bucket_inspector(
    bucket_name: str,
    max_results: int = 10,
    prefix: str | None = None,
) -> dict[str, Any]:
    """Inspects GCS bucket content and metadata conforming to MCP Spec 2026-07-28 resultType schema."""
    try:
        client = storage.Client()
        bucket = client.bucket(bucket_name)
        blobs = list(client.list_blobs(bucket, max_results=max_results, prefix=prefix))
        items = [{"name": b.name, "size_bytes": b.size, "updated": str(b.updated)} for b in blobs]
        return {
            "resultType": "complete",
            "bucket_name": bucket_name,
            "object_count_sample": len(items),
            "objects": items,
        }
    except Exception as e:
        logger.error("GCS Inspector Error: %s", e)
        return {"resultType": "complete", "error": f"Failed to inspect GCS bucket: {e!s}"}


@mcp.tool(description="Write structured operational or security audit log records to Google Cloud Logging.")
def cloud_logging_audit_writer(
    log_name: str,
    message: str,
    severity: str = "INFO",
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Sends structured audit entry to Cloud Logging."""
    try:
        client = cloud_logging.Client()
        logger_instance = client.logger(log_name)
        payload = {"message": message, "metadata": metadata or {}, "source": "mcp-server-gcp"}
        logger_instance.log_struct(payload, severity=severity.upper())
        return {
            "resultType": "complete",
            "status": "success",
            "log_name": log_name,
            "recorded_message": message,
        }
    except Exception as e:
        logger.error("Cloud Logging Error: %s", e)
        return {"resultType": "complete", "error": f"Failed to record audit log: {e!s}"}


@mcp.tool(description="Check health and operational state of Google Cloud project resources.")
def gcp_resource_health_checker(project_id: str | None = None) -> dict[str, Any]:
    """Returns project resource summary and status."""
    target_project = project_id or os.getenv("GOOGLE_CLOUD_PROJECT") or "unknown-project"
    return {
        "resultType": "complete",
        "status": "HEALTHY",
        "project_id": target_project,
        "mcp_spec_version": "2026-07-28",
        "_meta": {
            "io.modelcontextprotocol/protocolVersion": "2026-07-28",
            "io.modelcontextprotocol/serverInfo": {"name": "mcp-gcp-server", "version": "0.1.0"},
        },
        "transports_enabled": ["Streamable HTTP"],
        "ssl_tls_enabled": True,
    }


if __name__ == "__main__":
    port = int(os.getenv("PORT", "8080"))
    logger.info("Starting MCPServer on port %d (Streamable HTTP Transport, Spec 2026-07-28)...", port)
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=port,
        stateless_http=True,
        json_response=True,
    )

Prueba el servidor de MCP de forma local

Paso 1: Inicia el servidor de MCP

En tu terminal principal, inicia el servidor con uv:

uv run python -m src.mcp_server.server

Deberías ver registros de inicio que confirmen que el servidor HTTP transmisible está escuchando en el puerto 8080:

INFO:mcp-gcp-server:Starting MCPServer on port 8080 (Streamable HTTP Transport, Spec 2026-07-28)...
INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:mcp.server.streamable_http_manager:StreamableHTTP session manager started
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)

Mantén esta terminal abierta y en ejecución.

Paso 2: Verifica el extremo HTTP de transmisión con curl

En MCP 2.0 (MCP Spec 2026-07-28), el handshake initialize con estado y el encabezado Mcp-Session-Id se reemplazan por solicitudes sin estado. Puedes invocar tools/list directamente pasando los encabezados MCP-Protocol-Version y Mcp-Method junto con los metadatos del cliente en params._meta.

Abre una segunda ventana de la terminal y envía una solicitud HTTP POST transmisible:

cd mcp-server
curl -i -X POST http://localhost:8080/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -H "MCP-Protocol-Version: 2026-07-28" \
  -H "Mcp-Method: tools/list" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": {
          "name": "curl-test",
          "version": "1.0.0"
        },
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }'

MCPServer responderá con HTTP/1.1 200 OK y devolverá el resultado de tools/list de JSON-RPC directamente:

HTTP/1.1 200 OK
date: Wed, 12 Aug 2026 14:55:00 GMT
server: uvicorn

content-type: application/json

{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"vertex_ai_generate_content",...},{"name":"gcs_bucket_inspector",...},{"name":"cloud_logging_audit_writer",...},{"name":"gcp_resource_health_checker",...}]}}

Paso 3: Ejecuta el cliente de prueba de MCP

Para probar el descubrimiento y la ejecución de herramientas en tu servidor local a través de HTTP transmitible, crea una secuencia de comandos de cliente de prueba src/mcp_server/test_client.py:

import asyncio
from mcp import Client
from mcp.types import TextContent


async def test_mcp_server() -> None:
    """Connects to the local MCP v2.0 server, lists tools, and invokes health check."""
    server_url = "http://localhost:8080/mcp"
    print(f"[*] Connecting to local MCPServer at {server_url} (Streamable HTTP)...")

    async with Client(server_url) as client:
        print(
            f"[+] Connected (protocol: {client.protocol_version}, "
            f"server: {client.server_info.name if client.server_info else 'unknown'})."
        )

        # List available tools
        tools_response = await client.list_tools()
        print("\n[*] Discovered MCP Tools:")
        for tool in tools_response.tools:
            print(f"  - {tool.name}: {tool.description}")

        # Invoke gcp_resource_health_checker tool
        print("\n[*] Invoking tool: gcp_resource_health_checker...")
        health_result = await client.call_tool("gcp_resource_health_checker", {})
        print("[+] Result:")
        for content in health_result.content:
            if isinstance(content, TextContent):
                print(content.text)


if __name__ == "__main__":
    asyncio.run(test_mcp_server())

Ejecuta la secuencia de comandos del cliente de prueba con uv:

uv run python src/mcp_server/test_client.py

Deberías ver un resultado que demuestre la conexión, el descubrimiento y la ejecución de la herramienta:

[*] Connecting to local MCPServer at http://localhost:8080/mcp (Streamable HTTP)...
[+] Connected (protocol: 2026-07-28, server: Google Cloud Production Tools).

[*] Discovered MCP Tools:
  - vertex_ai_generate_content: Generate content or answer questions using Vertex AI Gemini model.
  - gcs_bucket_inspector: List objects and inspect metadata for a specified Google Cloud Storage bucket.
  - cloud_logging_audit_writer: Write structured operational or security audit log records to Google Cloud Logging.
  - gcp_resource_health_checker: Check health and operational state of Google Cloud project resources.

[*] Invoking tool: gcp_resource_health_checker...
[+] Result:
{"resultType": "complete", "status": "HEALTHY", "project_id": "my-gcp-project", "mcp_spec_version": "2026-07-28", "_meta": {"io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/serverInfo": {"name": "mcp-gcp-server", "version": "0.1.0"}}, "transports_enabled": ["Streamable HTTP"], "ssl_tls_enabled": true}

Una vez que se realice la verificación, presiona CTRL+C en tu terminal principal para detener el servidor local.

4. Crea un contenedor para el servidor de MCP

Para implementar el servidor de MCP en Cloud Run y GKE Autopilot, empaqueta el servidor en una imagen de contenedor mínima con un Dockerfile de varias etapas potenciado por uv.

Crea un Dockerfile

En mcp-server/Dockerfile:

FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS builder

WORKDIR /app
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy

COPY pyproject.toml uv.lock* /app/
RUN uv sync --no-install-project --no-dev

COPY README.md /app/
COPY src /app/src
RUN uv sync --no-dev

FROM python:3.12-slim-bookworm

WORKDIR /app
COPY --from=builder /app /app

ENV PATH="/app/.venv/bin:$PATH"
ENV PORT=8080
ENV PYTHONUNBUFFERED=1

EXPOSE 8080
CMD ["python", "-m", "src.mcp_server.server"]

Compila y envía la imagen a Artifact Registry

Crea un repositorio de Artifact Registry:

gcloud artifacts repositories create mcp-servers \
    --repository-format=docker \
    --location=us-central1 \
    --description="Docker repository for MCP Servers" \
    --project="${PROJECT_ID}"

Otorga los permisos de IAM necesarios a tu cuenta de usuario y a la cuenta de servicio predeterminada de Compute Engine para que Cloud Build pueda organizar fuentes, escribir registros y enviar imágenes a Artifact Registry:

export USER_EMAIL=$(gcloud config get-value account)

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="user:${USER_EMAIL}" \
    --role="roles/cloudbuild.builds.editor"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="user:${USER_EMAIL}" \
    --role="roles/storage.admin"

export DEFAULT_SA=$(gcloud iam service-accounts list \
    --filter="email:compute@developer.gserviceaccount.com" \
    --format="value(email)" \
    --project="${PROJECT_ID}")

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="serviceAccount:${DEFAULT_SA}" \
    --role="roles/storage.objectViewer"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="serviceAccount:${DEFAULT_SA}" \
    --role="roles/artifactregistry.writer"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="serviceAccount:${DEFAULT_SA}" \
    --role="roles/logging.logWriter"

Envía la compilación de la imagen con Cloud Build:

export IMAGE_URI="us-central1-docker.pkg.dev/${PROJECT_ID}/mcp-servers/secure-mcp-server:latest"

gcloud builds submit . --tag="${IMAGE_URI}" --project="${PROJECT_ID}"

Una vez que se complete el proceso, la imagen del contenedor se almacenará de forma segura en Artifact Registry:

SUCCESS: Image published to us-central1-docker.pkg.dev/.../secure-mcp-server:latest

5. Implementa en Cloud Run con IAM y HTTPS

Cloud Run proporciona un entorno sin servidores completamente administrado con finalización automática de certificados HTTPS / SSL y control de acceso detallado de Cloud IAM.

Crea una cuenta de servicio dedicada

Crea una cuenta de servicio de Google con privilegios mínimos para Cloud Run:

gcloud iam service-accounts create mcp-server-cr-sa \
    --display-name="MCP Server Cloud Run SA" \
    --project="${PROJECT_ID}"

export SA_EMAIL="mcp-server-cr-sa@${PROJECT_ID}.iam.gserviceaccount.com"

# Grant Vertex AI, Logging, and GCS permissions
gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/aiplatform.user"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/logging.logWriter"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
    --member="serviceAccount:${SA_EMAIL}" \
    --role="roles/storage.objectViewer"

Implementa el servicio en Cloud Run

Implementa el contenedor en Cloud Run con la autenticación de IAM aplicada (--no-allow-unauthenticated) y el SSL administrado predeterminado:

gcloud run deploy secure-mcp-server \
    --image="${IMAGE_URI}" \
    --platform=managed \
    --region=us-central1 \
    --service-account="${SA_EMAIL}" \
    --set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
    --no-allow-unauthenticated \
    --ingress=all \
    --project="${PROJECT_ID}"

Recupera la URL HTTPS:

export CLOUD_RUN_URL=$(gcloud run services describe secure-mcp-server --platform=managed --region=us-central1 --format='value(status.url)' --project="${PROJECT_ID}")
echo "Cloud Run HTTPS Endpoint: ${CLOUD_RUN_URL}"

Verifica la seguridad de extremos

Si intentas realizar una solicitud no autorizada al extremo HTTP de Streamable, se mostrará 403 Forbidden:

curl -i -X POST "${CLOUD_RUN_URL}/mcp" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -H "MCP-Protocol-Version: 2026-07-28" \
  -H "Mcp-Method: tools/list" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": {
          "name": "curl-test",
          "version": "1.0.0"
        },
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }'

HTTP/2 403
content-type: text/html; charset=UTF-8
date: Mon, 11 Aug 2026 14:00:00 GMT

Genera un token de ID de OIDC con gcloud para verificar la comunicación autorizada:

export ID_TOKEN=$(gcloud auth print-identity-token --audiences="${CLOUD_RUN_URL}")

curl -i -X POST "${CLOUD_RUN_URL}/mcp" \
  -H "Authorization: Bearer ${ID_TOKEN}" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -H "MCP-Protocol-Version: 2026-07-28" \
  -H "Mcp-Method: tools/list" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": {
          "name": "curl-test",
          "version": "1.0.0"
        },
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }'

HTTP/2 200
content-type: application/json

{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"vertex_ai_generate_content",...},{"name":"gcs_bucket_inspector",...},{"name":"cloud_logging_audit_writer",...},{"name":"gcp_resource_health_checker",...}]}}

Prueba de ejecución de la herramienta en vivo: Inspector de Cloud Storage

Ahora que el servidor de MCP remoto está autenticado y responde, prueba ejecutar la herramienta gcs_bucket_inspector en la infraestructura de Google Cloud.

Paso 1: Crea un bucket de Cloud Storage de prueba

Crea un bucket de prueba y sube un archivo de muestra:

export BUCKET_NAME="${PROJECT_ID}-mcp-demo"

# Create Cloud Storage bucket
gcloud storage buckets create "gs://${BUCKET_NAME}" \
    --location=us-central1 \
    --project="${PROJECT_ID}"

# Upload sample file
echo "Hello from Secure MCP on Google Cloud!" > sample.txt
gcloud storage cp sample.txt "gs://${BUCKET_NAME}/sample.txt"

Paso 2: Crea el cliente de prueba de la herramienta de GCS remota

Crea una secuencia de comandos de cliente de prueba llamada src/mcp_server/test_gcs_tool.py para autenticarte en Cloud Run y, luego, invoca la herramienta gcs_bucket_inspector:

import asyncio
import os
import subprocess
import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client
from mcp.types import TextContent


async def test_gcs_tool() -> None:
    """Authenticates to Cloud Run via OIDC and invokes the gcs_bucket_inspector tool."""
    cloud_run_url = os.getenv("CLOUD_RUN_URL")
    bucket_name = os.getenv("BUCKET_NAME")

    if not cloud_run_url or not bucket_name:
        print("[!] Please set both CLOUD_RUN_URL and BUCKET_NAME environment variables.")
        return

    # Generate Google OIDC ID token for Cloud Run authentication
    id_token = subprocess.check_output(
        ["gcloud", "auth", "print-identity-token", f"--audiences={cloud_run_url.rstrip('/')}"],
        text=True,
    ).strip()

    headers = {"Authorization": f"Bearer {id_token}"}
    server_url = f"{cloud_run_url.rstrip('/')}/mcp"

    print(f"[*] Connecting to remote Cloud Run MCP server at {server_url} (Streamable HTTP)...")
    async with httpx2.AsyncClient(
        headers=headers,
        timeout=httpx2.Timeout(30.0, read=300.0),
    ) as http_client:
        transport = streamable_http_client(server_url, http_client=http_client)
        async with Client(transport) as client:
            print(
                f"[+] Authenticated and connected (protocol: {client.protocol_version})."
            )

            # List tools
            tools_response = await client.list_tools()
            print(f"[*] Verified {len(tools_response.tools)} available tools on Cloud Run.")

            # Invoke gcs_bucket_inspector tool
            print(f"\n[*] Invoking tool: gcs_bucket_inspector on '{bucket_name}'...")
            result = await client.call_tool(
                "gcs_bucket_inspector", {"bucket_name": bucket_name}
            )

            print("[+] Response from Cloud Run MCP Server:")
            for content in result.content:
                if isinstance(content, TextContent):
                    print(content.text)


if __name__ == "__main__":
    asyncio.run(test_gcs_tool())

Paso 3: Ejecuta el cliente de prueba de GCS remoto

Ejecuta la secuencia de comandos de prueba con uv:

uv run python src/mcp_server/test_gcs_tool.py

Deberías ver los metadatos del objeto activo que devuelve el servidor de MCP de Cloud Run:

[*] Connecting to remote Cloud Run MCP server at https://secure-mcp-server-...-uc.a.run.app/mcp (Streamable HTTP)...
[+] Authenticated and connected (protocol: 2026-07-28).
[*] Verified 4 available tools on Cloud Run.

[*] Invoking tool: gcs_bucket_inspector on 'my-project-mcp-demo'...
[+] Response from Cloud Run MCP Server:
{"resultType":"complete","bucket_name":"my-project-mcp-demo","object_count_sample":1,"objects":[{"name":"sample.txt","size_bytes":39,"updated":"..."}]}

6. Implementa en GKE Autopilot con Workload Identity y TLS

En el caso de las cargas de trabajo de Kubernetes, GKE Autopilot administra el aprovisionamiento de nodos, mientras que Workload Identity elimina las claves estáticas de las cuentas de servicio.

Aprovisiona un clúster de GKE Autopilot

Aprovisiona un clúster de GKE Autopilot:

gcloud container clusters create-auto mcp-gke-cluster \
    --location=us-central1 \
    --project="${PROJECT_ID}"

gcloud container clusters get-credentials mcp-gke-cluster \
    --location=us-central1 \
    --project="${PROJECT_ID}"

Configura Workload Identity

Crea una cuenta de servicio de Google (GSA) y una cuenta de servicio de Kubernetes (KSA), y luego vincúlalas con Workload Identity:

# 1. Create GSA
gcloud iam service-accounts create mcp-gke-sa \
    --display-name="GKE MCP Service Account" \
    --project="${PROJECT_ID}"

export GSA_EMAIL="mcp-gke-sa@${PROJECT_ID}.iam.gserviceaccount.com"

# 2. Grant IAM Roles to GSA
gcloud projects add-iam-policy-binding "${PROJECT_ID}" --member="serviceAccount:${GSA_EMAIL}" --role="roles/aiplatform.user"
gcloud projects add-iam-policy-binding "${PROJECT_ID}" --member="serviceAccount:${GSA_EMAIL}" --role="roles/logging.logWriter"
gcloud projects add-iam-policy-binding "${PROJECT_ID}" --member="serviceAccount:${GSA_EMAIL}" --role="roles/storage.objectViewer"

# 3. Create KSA
kubectl create serviceaccount mcp-server-ksa --namespace default

# 4. Annotate KSA
kubectl annotate serviceaccount mcp-server-ksa \
    --namespace default \
    iam.gke.io/gcp-service-account="${GSA_EMAIL}"

# 5. Bind KSA to GSA
gcloud iam service-accounts add-iam-policy-binding "${GSA_EMAIL}" \
    --role="roles/iam.workloadIdentityUser" \
    --member="serviceAccount:${PROJECT_ID}.svc.id.goog[default/mcp-server-ksa]" \
    --project="${PROJECT_ID}"

Aplica la API de Kubernetes Gateway y Deployment con TLS

Paso 1: Reserva una IP estática y aprovisiona un certificado SSL administrado por Google con nip.io

Reserva una dirección IP externa global y aprovecha el servicio DNS de comodín nip.io (..nip.io) para establecer un nombre de dominio público válido (MCP_DOMAIN) antes de aplicar los manifiestos de Kubernetes:

# Reserve global static IP address for GKE Gateway Load Balancer
gcloud compute addresses create mcp-server-ip \
    --global \
    --project="${PROJECT_ID}"

export MCP_IP=$(gcloud compute addresses describe mcp-server-ip --global --format="value(address)" --project="${PROJECT_ID}")
export MCP_DOMAIN="mcp.${MCP_IP}.nip.io"

echo "Reserved Static IP: ${MCP_IP}"
echo "Configured nip.io Domain: ${MCP_DOMAIN}"

# Provision Google-managed SSL certificate
gcloud compute ssl-certificates create mcp-server-cert \
    --domains="${MCP_DOMAIN}" \
    --global \
    --project="${PROJECT_ID}"

No es necesario que esperes a que finalice el aprovisionamiento del certificado SSL para continuar. De hecho, los certificados administrados por Google permanecen en estado PROVISIONING hasta que se adjuntan al balanceador de cargas de puerta de enlace en el paso 3. Continúa de inmediato con los próximos pasos.

Paso 2: Crea el manifiesto de Deployment y Service

Crea deployment.yaml que contenga las definiciones de Deployment y Service internos. Dado que MCP 2.0 (MCP Spec 2026-07-28) no tiene estado, el Service de Kubernetes no requiere afinidad de sesión de IP de cliente:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server-deployment
  namespace: default
  labels:
    app: mcp-server
    # GKE takes this label and registers the deployment as an MCP server to Agent Registry
    registry.gke.io/functional-type: "MCP_SERVER"
  annotations:
    # Endpoint URL where the GKE controller can access this MCP server
    modelcontextprotocol.info/urls: |
      - https://MCP_DOMAIN/mcp
    # Defines structural capabilities for the MCP server card
    modelcontextprotocol.info/capabilities: |
      card:
        endpoint: "/mcp"
        protocol: "HTTP"
spec:
  replicas: 2
  selector:
    matchLabels:
      app: mcp-server
  template:
    metadata:
      labels:
        app: mcp-server
      annotations:
        # Workload Identity annotation for identity and access management
        iam.gke.io/spiffe-identity-type: agent-identity
    spec:
      serviceAccountName: mcp-server-ksa
      containers:
      - name: mcp-server
        image: us-central1-docker.pkg.dev/PROJECT_ID/mcp-servers/secure-mcp-server:latest
        ports:
        - containerPort: 8080
          name: http
        env:
        - name: PORT
          value: "8080"
        - name: GOOGLE_CLOUD_PROJECT
          value: "PROJECT_ID"
        resources:
          requests:
            cpu: "250m"
            memory: "512Mi"
          limits:
            cpu: "1000m"
            memory: "1Gi"
        readinessProbe:
          httpGet:
            path: /healthz
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 15
---
apiVersion: v1
kind: Service
metadata:
  name: mcp-server-service
  namespace: default
  labels:
    app: mcp-server
spec:
  type: ClusterIP
  ports:
  - port: 80
    targetPort: 8080
    name: http
  selector:
    app: mcp-server

Sustituye MCP_DOMAIN y PROJECT_ID, y aplica la implementación:

sed -e "s|MCP_DOMAIN|${MCP_DOMAIN}|g" \
    -e "s|PROJECT_ID|${PROJECT_ID}|g" \
    deployment.yaml | kubectl apply -f -

Paso 3: Crea el manifiesto de Gateway API, HealthCheckPolicy y GCPBackendPolicy

Crea gateway.yaml que contenga la puerta de enlace, la HTTPRoute, un HealthCheckPolicy y un GCPBackendPolicy:

HealthCheckPolicy: Configura el balanceador de cargas de Google Cloud para que sondee /healthz en el puerto 8080.
GCPBackendPolicy (Tiempo de espera del backend): Establece timeoutSec: 300 para que se alinee con el tiempo de espera de lectura de 300 segundos del SDK de MCP v2 para transmisiones HTTP / SSE transmitibles (la puerta de enlace de GKE tiene un valor predeterminado de 30 segundos si se omite).

apiVersion: gateway.networking.k8s.io/v1
kind: Gateway
metadata:
  name: mcp-gateway
  namespace: default
spec:
  gatewayClassName: gke-l7-global-external-managed
  listeners:
  - name: https
    protocol: HTTPS
    port: 443
    tls:
      mode: Terminate
      options:
        networking.gke.io/pre-shared-certs: mcp-server-cert
  addresses:
  - type: NamedAddress
    value: mcp-server-ip
---
apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
  name: mcp-http-route
  namespace: default
spec:
  parentRefs:
  - name: mcp-gateway
  hostnames:
  - "MCP_DOMAIN"
  rules:
  - matches:
    - path:
        type: PathPrefix
        value: /
    backendRefs:
    - name: mcp-server-service
      port: 80
---
apiVersion: networking.gke.io/v1
kind: HealthCheckPolicy
metadata:
  name: mcp-health-check-policy
  namespace: default
spec:
  default:
    checkIntervalSec: 15
    timeoutSec: 5
    healthyThreshold: 1
    unhealthyThreshold: 2
    config:
      type: HTTP
      httpHealthCheck:
        port: 8080
        requestPath: /healthz
  targetRef:
    group: ""
    kind: Service
    name: mcp-server-service
---
apiVersion: networking.gke.io/v1
kind: GCPBackendPolicy
metadata:
  name: mcp-backend-policy
  namespace: default
spec:
  default:
    # Aligns with MCP SDK v2's 300s read timeout for Streamable HTTP / SSE streams
    # (GKE Gateway defaults to 30s if omitted)
    timeoutSec: 300
  targetRef:
    group: ""
    kind: Service
    name: mcp-server-service

Aplica Gateway, HTTPRoute, HealthCheckPolicy y GCPBackendPolicy:

sed "s/MCP_DOMAIN/${MCP_DOMAIN}/g" gateway.yaml | kubectl apply -f -

Paso 4: Verifica la implementación y prueba el servidor de MCP con la redirección de puertos

Debido a que los certificados SSL administrados por Google y los balanceadores de cargas de aplicaciones externos de Google Cloud tardan entre 5 y 15 minutos en aprovisionarse y establecer el enrutamiento de DNS, puedes probar de inmediato los Pods de GKE en ejecución con kubectl port-forward.

Primero, verifica que tus Pods y la puerta de enlace se estén ejecutando:

kubectl get pods -l app=mcp-server
kubectl get gateway mcp-gateway

NAME                                     READY   STATUS    RESTARTS   AGE
mcp-server-deployment-7b8f9495c5-x2n8q   1/1     Running   0          45s
mcp-server-deployment-7b8f9495c5-z4k9p   1/1     Running   0          45s

NAME          CLASS                             ADDRESS          PROGRAMMED   AGE
mcp-gateway   gke-l7-global-external-managed   34.120.x.x       True         2m

A continuación, prueba el servicio de GKE activo de forma local con la redirección de puertos:

En tu terminal, reenvía el puerto local 8080 al servicio ClusterIP de GKE:

kubectl port-forward svc/mcp-server-service 8080:80

En una segunda terminal, crea una secuencia de comandos de cliente de prueba llamada src/mcp_server/test_vertex_tool.py para invocar la herramienta vertex_ai_generate_content en GKE con Workload Identity:

import argparse
import asyncio
import os
from mcp import Client
from mcp.types import TextContent


async def test_vertex_tool(server_url: str) -> None:
    """Connects to the MCP v2.0 server and invokes the vertex_ai_generate_content tool."""
    print(f"[*] Connecting to MCPServer at {server_url} (Streamable HTTP)...")

    async with Client(server_url) as client:
        print(
            f"[+] Connected (protocol: {client.protocol_version}, "
            f"server: {client.server_info.name if client.server_info else 'unknown'})."
        )

        # List available tools
        tools_response = await client.list_tools()
        print(f"[*] Discovered {len(tools_response.tools)} MCP Tools:")
        for tool in tools_response.tools:
            print(f"  - {tool.name}")

        # Invoke vertex_ai_generate_content tool
        prompt = "Explain in 2 sentences why Model Context Protocol (MCP) Streamable HTTP is great for cloud deployments."
        print(f"\n[*] Invoking tool: vertex_ai_generate_content with prompt: '{prompt}'...")

        result = await client.call_tool(
            "vertex_ai_generate_content",
            {
                "prompt": prompt,
                "model_name": "gemini-2.5-flash",
            },
        )

        print("\n[+] Response from Vertex AI Gemini:")
        for content in result.content:
            if isinstance(content, TextContent):
                print(content.text)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Test Vertex AI MCP Tool on MCPServer.")
    parser.add_argument(
        "--host",
        default=os.getenv("MCP_URL", "http://localhost:8080/mcp"),
        help="MCP Server host or URL (default: http://localhost:8080/mcp or $MCP_URL)",
    )
    args = parser.parse_args()

    # Normalize URL format
    url = args.host
    if not url.startswith("http://") and not url.startswith("https://"):
        url = f"https://{url}"
    if not url.endswith("/mcp"):
        url = f"{url.rstrip('/')}/mcp"

    asyncio.run(test_vertex_tool(url))

Ejecuta la secuencia de comandos de prueba en la instancia local con reenvío de puertos:

uv run python src/mcp_server/test_vertex_tool.py --host="http://localhost:8080/mcp"

[*] Connecting to MCPServer at http://localhost:8080/mcp (Streamable HTTP)...
[+] Connected (protocol: 2026-07-28, server: Google Cloud Production Tools).
[*] Discovered 4 MCP Tools:
  - vertex_ai_generate_content
  - gcs_bucket_inspector
  - cloud_logging_audit_writer
  - gcp_resource_health_checker

[*] Invoking tool: vertex_ai_generate_content with prompt: 'Explain in 2 sentences why Model Context Protocol (MCP) Streamable HTTP is great for cloud deployments.'...

[+] Response from Vertex AI Gemini:
MCP Streamable HTTP enables lightweight, stateless HTTP interactions that scale effortlessly across cloud-native platforms like GKE and Cloud Run. It simplifies infrastructure management by using standard HTTP/HTTPS protocols while preserving rich bidirectional streaming for AI agents.

Esto confirma que los Pods de GKE están en buen estado, que Workload Identity se autentica correctamente en Vertex AI sin credenciales estáticas y que el transporte HTTP transmitible funciona según lo esperado.

Paso 5: Verifica el extremo de la puerta de enlace HTTPS pública

Verifica el estado de aprovisionamiento del certificado con el siguiente comando:

gcloud compute ssl-certificates describe mcp-server-cert --global --format="value(managed.status)"

Una vez que el certificado esté ACTIVE, prueba el extremo HTTPS público con el cliente de prueba de Python y la marca --host:

uv run python src/mcp_server/test_vertex_tool.py --host="https://${MCP_DOMAIN}/mcp"

[*] Connecting to MCPServer at https://mcp.34.120.x.x.nip.io/mcp (Streamable HTTP)...
[+] Connected (protocol: 2026-07-28, server: Google Cloud Production Tools).
[*] Discovered 4 MCP Tools:
  - vertex_ai_generate_content
  - gcs_bucket_inspector
  - cloud_logging_audit_writer
  - gcp_resource_health_checker

[*] Invoking tool: vertex_ai_generate_content with prompt: 'Explain in 2 sentences why Model Context Protocol (MCP) Streamable HTTP is great for cloud deployments.'...

[+] Response from Vertex AI Gemini:
MCP Streamable HTTP enables lightweight, stateless HTTP interactions that scale effortlessly across cloud-native platforms like GKE and Cloud Run. It simplifies infrastructure management by using standard HTTP/HTTPS protocols while preserving rich bidirectional streaming for AI agents.

7. Integra el servidor de MCP con Google Cloud Agent Platform

Ahora que tu MCPServer se implementó de forma segura en extremos HTTPS en Cloud Run y GKE Autopilot, registra y descubre tus herramientas de MCP con Google Cloud Agent Platform (Agent Registry) para que los agentes empresariales y los modelos de IA puedan descubrirlos e invocarlos de forma dinámica.

1. Registra el servidor de MCP personalizado de Cloud Run en Agent Platform y realiza pruebas con Service Discovery

Paso 1: Extrae las especificaciones de la herramienta (`toolspec.json`).

Para registrar un servidor de MCP externo o personalizado en Agent Registry, consulta tools/list en tu servidor de MCP 2.0 sin estado activo, quita los campos _meta a nivel de la herramienta y guarda el resultado en toolspec.json en una sola canalización:

# 1. Generate identity token
export ID_TOKEN=$(gcloud auth print-identity-token --audiences="${CLOUD_RUN_URL}")

# 2. Query, parse, sanitize and save in one single pipeline
curl -s -X POST "${CLOUD_RUN_URL}/mcp" \
  -H "Authorization: Bearer ${ID_TOKEN}" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -H "MCP-Protocol-Version: 2026-07-28" \
  -H "Mcp-Method: tools/list" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {
      "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientInfo": {
          "name": "cli",
          "version": "1.0.0"
        },
        "io.modelcontextprotocol/clientCapabilities": {}
      }
    }
  }' \
  | sed -n 's/^data: //p; /^{/p' \
  | jq '.result | del(.tools[]._meta)' > toolspec.json

Paso 2: Registra el servicio en Agent Registry

Usa gcloud agent-registry services create para catalogar tu servidor de MCP:

export SERVER_NAME="secure-mcp-server"
export DISPLAY_NAME="Google Cloud MCPServer"
export REGION="global"

gcloud agent-registry services create "${SERVER_NAME}" \
  --project="${PROJECT_ID}" \
  --location="${REGION}" \
  --display-name="${DISPLAY_NAME}" \
  --mcp-server-spec-type="tool-spec" \
  --mcp-server-spec-content=toolspec.json \
  --interfaces="url=${CLOUD_RUN_URL}/mcp,protocolBinding=jsonrpc"

Verifica que el servicio se haya registrado correctamente:

gcloud agent-registry services describe "${SERVER_NAME}" --location="${REGION}"

name: projects/PROJECT_ID/locations/global/services/secure-mcp-server
displayName: Google Cloud MCPServer
interfaces:
- protocolBinding: JSONRPC
  url: https://secure-mcp-server-xxxx.a.run.app/mcp
mcpServerSpec:
  type: TOOL_SPEC

Paso 3: Crea y ejecuta el cliente de prueba de Service Discovery

Crea una secuencia de comandos de Python llamada src/mcp_server/test_agent_platform.py que resuelva de forma dinámica el extremo del catálogo mcp-servers de Agent Registry, se autentique con Google Cloud IAM y, luego, invoque una herramienta con httpx2 y el Client de MCP 2.0:

import asyncio
import os
import subprocess
import httpx2  # MCP SDK 2.x utilizes httpx2 instead of httpx
from mcp import Client
from mcp.client.streamable_http import streamable_http_client


async def main() -> None:
    server_name = os.getenv("SERVER_NAME", "secure-mcp-server")
    location = os.getenv("REGION", "global")

    # 1. Discover endpoint URL from Google Cloud Agent Registry (mcp-servers catalog)
    print(f"[*] Discovering '{server_name}' in '{location}' from Agent Registry...")
    url = subprocess.check_output(
        [
            "gcloud",
            "agent-registry",
            "mcp-servers",
            "list",
            f"--location={location}",
            f"--filter=displayName='{server_name}' OR mcpServerId ~ ':{server_name}$'",
            "--format=value(interfaces[0].url)",
            "--limit=1",
        ],
        text=True,
    ).strip()

    if not url:
        raise RuntimeError(
            f"No endpoint URL found for '{server_name}' in location '{location}'. "
            "Verify the server is registered and has an interface URL configured."
        )
    print(f"[+] Discovered Endpoint: {url}")

    # 2. Generate IAM identity token if connecting to Cloud Run
    headers: dict[str, str] = {}
    if "run.app" in url:
        audience = url.split("/mcp")[0]
        token = subprocess.check_output(
            ["gcloud", "auth", "print-identity-token", f"--audiences={audience}"],
            text=True,
        ).strip()
        headers["Authorization"] = f"Bearer {token}"

    # 3. Configure a custom httpx2 Client with MCP-safe timeouts
    # We set read to 300s to ensure the long-lived SSE/GET stream stays open
    async with httpx2.AsyncClient(
        headers=headers,
        timeout=httpx2.Timeout(30.0, read=300.0),
    ) as http_client:

        # 4. Initialize Streamable HTTP Transport (yielding a 2-tuple in MCP v2.x)
        transport = streamable_http_client(url, http_client=http_client)

        # 5. Connect using the clean high-level Client interface
        async with Client(transport) as client:
            print("[+] Session active. Connected successfully.")

            # Verify tools registered in the catalog (Attributes are snake_case in MCP v2)
            tools_response = await client.list_tools()
            print(f"[+] Discovered {len(tools_response.tools)} tools:")
            for tool in tools_response.tools:
                print(f"    - {tool.name}")

            # Test executing the health check tool
            print("\n[*] Invoking tool: gcp_resource_health_checker...")
            result = await client.call_tool("gcp_resource_health_checker", {})
            print(f"[+] Tool Output from Agent Platform:\n{result.content[0].text}")


if __name__ == "__main__":
    asyncio.run(main())

Ejecuta la secuencia de comandos de prueba:

uv run python src/mcp_server/test_agent_platform.py

[*] Discovering 'secure-mcp-server' in 'global' from Agent Registry...
[+] Discovered Endpoint: https://secure-mcp-server-xxxx.a.run.app/mcp
[+] Session active. Connected successfully.
[+] Discovered 4 tools:
    - vertex_ai_generate_content
    - gcs_bucket_inspector
    - cloud_logging_audit_writer
    - gcp_resource_health_checker

[*] Invoking tool: gcp_resource_health_checker...
[+] Tool Output from Agent Platform:
{"resultType": "complete", "status": "HEALTHY", "mcp_spec_version": "2026-07-28"}

2. Registro y descubrimiento automáticos del servidor de MCP con GKE

Google Kubernetes Engine (GKE) proporciona integración automática con Agent Registry. Cuando etiquetas y anotas tu manifiesto de implementación de GKE, GKE realiza automáticamente un análisis de introspección en el servidor de MCP, registra las herramientas en el catálogo y proyecta el servicio en el lado del consumidor sin necesidad de generar o subir manualmente un toolspec.json.

Paso 1: Cómo funciona el descubrimiento automático de GKE

Cuando se aplicó tu deployment.yaml en la sección anterior, las siguientes configuraciones habilitaron el descubrimiento automático:

registry.gke.io/functional-type: "MCP_SERVER": Informa al controlador del clúster de GKE que registre la implementación en el Agent Registry de Google Cloud.
modelcontextprotocol.info/urls: Proporciona el extremo de la puerta de enlace HTTPS externa (https://${MCP_DOMAIN}/mcp) para la introspección del controlador y el enrutamiento del consumidor.
modelcontextprotocol.info/capabilities: Declara el extremo de transporte HTTP (/mcp).
iam.gke.io/spiffe-identity-type: agent-identity: Configura Workload Identity para la comunicación con agentes.

Paso 2: Verifica el registro automático de GKE en el registro de agentes

Verifica que GKE haya descubierto y registrado automáticamente tu servidor de MCP en el catálogo regional mcp-servers:

# List the automatically registered GKE MCP server in us-central1
gcloud agent-registry mcp-servers list \
  --location=us-central1 \
  --format="table(displayName, tools.len():label=TOOLS)"

DISPLAY_NAME           TOOLS
mcp-server-deployment  4

Paso 3: Prueba el servidor de MCP de GKE con el descubrimiento de servicios

Prueba tu implementación de MCP de GKE con la secuencia de comandos de detección de servicios en el nombre del servidor registrado automáticamente en us-central1:

SERVER_NAME="mcp-server-deployment" \
REGION=us-central1 \
uv run python src/mcp_server/test_agent_platform.py

[*] Discovering 'mcp-server-deployment' in 'us-central1' from Agent Registry...
[+] Discovered Endpoint: https://mcp.34.120.x.x.nip.io/mcp
[+] Session active. Connected successfully.
[+] Discovered 4 tools:
    - vertex_ai_generate_content
    - gcs_bucket_inspector
    - cloud_logging_audit_writer
    - gcp_resource_health_checker

[*] Invoking tool: gcp_resource_health_checker...
[+] Tool Output from Agent Platform:
{"resultType": "complete", "status": "HEALTHY", "mcp_spec_version": "2026-07-28"}

8. Limpia los recursos

Para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos que usaste en este codelab, borra el servicio de Cloud Run, el clúster de GKE, las cuentas de servicio y el repositorio de contenedores.

Borra el servicio de Cloud Run

gcloud run services delete secure-mcp-server \
    --platform=managed \
    --region=us-central1 \
    --quiet \
    --project="${PROJECT_ID}"

Borra el clúster de GKE Autopilot

gcloud container clusters delete mcp-gke-cluster \
    --location=us-central1 \
    --quiet \
    --project="${PROJECT_ID}"

Borra cuentas de servicio y vinculaciones de IAM

gcloud iam service-accounts delete "mcp-server-cr-sa@${PROJECT_ID}.iam.gserviceaccount.com" --quiet --project="${PROJECT_ID}"
gcloud iam service-accounts delete "mcp-gke-sa@${PROJECT_ID}.iam.gserviceaccount.com" --quiet --project="${PROJECT_ID}"

Borra el repositorio de Artifact Registry

gcloud artifacts repositories delete mcp-servers \
    --location=us-central1 \
    --quiet \
    --project="${PROJECT_ID}"

Borra el servicio personalizado de Agent Registry

gcloud agent-registry services delete secure-mcp-server \
    --location=global \
    --quiet \
    --project="${PROJECT_ID}"

Borra el bucket de Cloud Storage

gcloud storage rm --recursive "gs://${BUCKET_NAME}" --quiet

Borra la IP estática y el certificado SSL

gcloud compute ssl-certificates delete mcp-server-cert --global --quiet --project="${PROJECT_ID}"
gcloud compute addresses delete mcp-server-ip --global --quiet --project="${PROJECT_ID}"

Verifica que se haya completado la limpieza de la eliminación:

Deleted service [secure-mcp-server].
Deleted cluster [mcp-gke-cluster].
Deleted repository [mcp-servers].

9. Felicitaciones

¡Felicitaciones! Compilaste, protegiste e implementaste correctamente un servidor de MCP 2.0 con MCPServer del SDK de Python del MCP y uv en Google Cloud.

Temas que se abordaron

Se construyó un MCPServer que admite la especificación de MCP (28-07-2026) a través de HTTP sin estado y transmitible.
Se crearon 4 herramientas de producción de Google Cloud que integran Vertex AI Gemini, Cloud Storage, Cloud Logging y Resource Health.
Implementaciones de contenedores protegidas en Cloud Run con autenticación de IAM y SSL/TLS aplicadas.
Se configuró la autenticación sin secretos en GKE Autopilot con Workload Identity y la API de Kubernetes Gateway con TLS.
Extremos de servidores de MCP registrados y descubiertos con Gemini Enterprise Agent Platform (Agent Registry) que usan encabezados de autorización de OIDC

Próximos pasos

Explora la Especificación del Protocolo de contexto del modelo para obtener recursos avanzados y definiciones de instrucciones.
Obtén más información sobre el SDK de Python de MCP.
Analiza en detalle la seguridad de Google Cloud Run y Workload Identity de GKE.