1. 准备工作
Model Context Protocol (MCP) 是一种开放标准，可让 AI 模型和智能体安全地访问工具、数据库和企业上下文。自 2024 年推出以来，该协议已获得云服务提供商和 LLM 提供商的广泛采用。最新规范 MCP Spec 2026-07-28 (MCP 2.0) 定义了无状态架构、简化的传输和严格的结果类型，同时保持与早期版本的向后兼容性。
官方 MCP Python SDK (mcp>=2.0.0) 提供 MCPServer (mcp.server.mcpserver.MCPServer)，它取代了 MCP 2.0 中的 FastMCP，成为一个高性能、开发者友好的框架，用于打造符合最新 MCP 规范 (2026-07-28) 的可用于生产用途的 MCP 服务器，该服务器通过 SSL/TLS 使用可流式传输的 HTTP 和服务器发送的事件 (SSE) 传输。
在此 Codelab 中，您将使用 MCP 2.0 Python SDK 中的 MCPServer 和 uv 依赖项管理功能构建生产级 MCP 服务器。您将为 MCP 服务器配备四种 Google Cloud 工具（Vertex AI Gemini 调用、Google Cloud Storage 检查、Cloud Logging 审核写入和 Google Cloud 资源健康状况检查）。然后，您将对服务器进行容器化处理，并将其部署到两个 Google Cloud 运行时目标平台：Cloud Run 和 Google Kubernetes Engine (GKE) Autopilot，并在 Gemini Enterprise Agent Platform 中注册和使用 MCP 服务器。
您将执行的操作
- 使用 Python 3.12 及更高版本和
uv构建MCPServer，并符合 MCP 规范 2026-07-28。 - 使用多阶段 Docker 构建将 MCP 服务器容器化。
- 通过强制执行 IAM 身份验证和 SSL/TLS，将 MCP 服务器安全地部署到 Cloud Run。
- 使用 Workload Identity 和 Kubernetes Gateway API with TLS 安全地将 MCP 服务器部署到 GKE Autopilot。
- 使用 OIDC 不记名令牌标头向 Gemini Enterprise Agent Platform 注册安全的 MCP 服务器端点。
所需条件
- 启用了结算功能的 Google Cloud 项目。
- 已安装并配置 Google Cloud SDK (
gcloudCLI)。 - 已安装 Python 3.12 及更高版本和
uv软件包管理系统。 - 已安装
docker。 kubectl命令行工具已安装。
2. 设置 Google Cloud 环境
在创建资源之前，请对您的环境进行身份验证并启用必要的 Google Cloud API。
对 gcloud CLI 进行身份验证
登录您的 Google Cloud 账号：
gcloud auth login
设置有效的 Google Cloud 项目 ID：
export PROJECT_ID=$(gcloud config get-value project)
gcloud config set project ${PROJECT_ID}
启用 Google Cloud 服务
为 Cloud Run、GKE、Vertex AI、Artifact Registry、Cloud Build、Cloud Logging、Storage 和 Compute Engine 启用所有必需的 API：
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}"
验证 API 是否已成功启用：
Operation "operations/..." finished successfully.
对应用默认凭证进行身份验证
对您的环境进行身份验证，以便 Python 客户端库在开发期间可以在本地访问 Vertex AI 和 Cloud Storage：
gcloud auth application-default login
3. 使用 MCPServer 和 uv 构建 MCP 服务器
在此步骤中，您将使用 uv 初始化 Python 项目，并构建支持四项 Google Cloud 功能的 MCPServer。
使用 uv 初始化项目
创建服务器目录并初始化 uv：
mkdir -p mcp-server/src/mcp_server
cd mcp-server
uv init --lib
将代码复制到 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"]
使用 uv 同步依赖项：
uv sync
编写 MCPServer 代码
在 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,
)
在本地测试 MCP 服务器
第 1 步：启动 MCP 服务器
在主终端中，使用 uv 启动服务器：
uv run python -m src.mcp_server.server
您应该会看到启动日志，确认 Streamable HTTP 服务器正在监听端口 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)
让此终端保持打开和运行状态。
第 2 步：使用 curl 验证可流式传输的 HTTP 端点
在 MCP 2.0（MCP 规范 2026-07-28）中，有状态的 initialize 握手和 Mcp-Session-Id 标头被无状态请求所取代。您可以在 params._meta 中传递 MCP-Protocol-Version 和 Mcp-Method 标头以及客户端元数据，从而直接调用 tools/list。
打开第二个终端窗口，并发送可流式传输的 HTTP POST 请求：
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 将以 HTTP/1.1 200 OK 进行响应，并直接返回 JSON-RPC tools/list 结果：
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",...}]}}
第 3 步：运行 MCP 测试客户端
如需测试通过可流式传输的 HTTP 在本地服务器上发现和执行工具，请创建测试客户端脚本 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())
使用 uv 运行测试客户端脚本：
uv run python src/mcp_server/test_client.py
您应该会看到输出，显示连接、工具发现和工具执行均成功：
[*] 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}
验证完成后，在主终端中按 CTRL+C 停止本地服务器。
4. 将 MCP 服务器容器化
如需将 MCP 服务器部署到 Cloud Run 和 GKE Autopilot，请使用由 uv 提供支持的多阶段 Dockerfile 将服务器打包到最小容器映像中。
创建 Dockerfile
在 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"]
构建映像并将其推送到 Artifact Registry
创建 Artifact Registry 代码库：
gcloud artifacts repositories create mcp-servers \
--repository-format=docker \
--location=us-central1 \
--description="Docker repository for MCP Servers" \
--project="${PROJECT_ID}"
向您的用户账号和默认 Compute Engine 服务账号授予所需的 IAM 权限，以便 Cloud Build 可以暂存来源、写入日志并将映像推送到 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"
使用 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}"
完成后，您的容器映像会安全地存储在 Artifact Registry 中：
SUCCESS: Image published to us-central1-docker.pkg.dev/.../secure-mcp-server:latest
5. 使用 IAM 和 HTTPS 部署到 Cloud Run
Cloud Run 提供全托管式无服务器环境，支持自动终止 HTTPS / SSL 证书，并提供精细的 Cloud IAM 访问权限控制。
创建专用服务账号
为 Cloud Run 创建最低权限的 Google 服务账号：
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"
将服务部署到 Cloud Run
将容器部署到 Cloud Run，并强制执行 IAM 身份验证 (--no-allow-unauthenticated) 和默认的代管式 SSL：
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}"
检索 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}"
验证端点安全
尝试向 Streamable HTTP 端点发送未经授权的请求将返回 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
使用 gcloud 生成 OIDC ID 令牌，以验证授权通信：
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",...}]}}
测试实时工具执行：Cloud Storage 检查器
现在，远程 MCP 服务器已通过身份验证并开始响应，接下来测试针对 Google Cloud 基础架构执行 gcs_bucket_inspector 工具。
第 1 步：创建测试 Cloud Storage 存储桶
创建测试存储桶并上传示例文件：
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"
第 2 步：创建远程 GCS 工具测试客户端
创建名为 src/mcp_server/test_gcs_tool.py 的测试客户端脚本，以针对 Cloud Run 进行身份验证并调用 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())
第 3 步：运行远程 GCS 测试客户端
使用 uv 运行测试脚本：
uv run python src/mcp_server/test_gcs_tool.py
您应该会看到从 Cloud Run MCP 服务器返回的现行对象元数据：
[*] 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. 使用 Workload Identity 和 TLS 部署到 GKE Autopilot
对于 Kubernetes 工作负载，GKE Autopilot 会管理节点预配，而Workload Identity则会消除静态服务账号密钥。
预配 GKE Autopilot 集群
预配 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}"
设置 Workload Identity
创建 Google 服务账号 (GSA) 和 Kubernetes 服务账号 (KSA)，然后使用 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}"
应用 Kubernetes Deployment 和 Gateway API（使用 TLS）
第 1 步：预留静态 IP 并使用 nip.io 配置 Google 管理的 SSL 证书
预留一个全局外部 IP 地址，并利用通配符 DNS 服务 nip.io (MCP_DOMAIN)：
# 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}"
您无需等待 SSL 证书完成配置即可继续操作。事实上，在第 3 步中将 Google 管理的证书附加到网关负载均衡器之前，这些证书会一直处于 PROVISIONING 状态。立即继续执行后续步骤。
第 2 步：创建 Deployment 和 Service 清单
创建包含 Deployment 和内部 Service 定义的 deployment.yaml。由于 MCP 2.0 (MCP 规范 2026-07-28) 是无状态的，因此 Kubernetes Service 不需要客户端 IP 会话亲和性：
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
替换 MCP_DOMAIN 和 PROJECT_ID 并应用部署：
sed -e "s|MCP_DOMAIN|${MCP_DOMAIN}|g" \
-e "s|PROJECT_ID|${PROJECT_ID}|g" \
deployment.yaml | kubectl apply -f -
第 3 步：创建 Gateway API、HealthCheckPolicy 和 GCPBackendPolicy 清单
创建包含 Gateway、HTTPRoute、HealthCheckPolicy 和 GCPBackendPolicy 的 gateway.yaml：
HealthCheckPolicy：将 Google Cloud 负载均衡器配置为探测端口 8080 上的/healthz。GCPBackendPolicy（后端超时）：将timeoutSec: 300设置为与 MCP SDK v2 的 300 秒可流式传输的 HTTP / SSE 流读取超时时间保持一致（如果省略，GKE Gateway 默认为 30 秒）。
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
应用 Gateway、HTTPRoute、HealthCheckPolicy 和 GCPBackendPolicy：
sed "s/MCP_DOMAIN/${MCP_DOMAIN}/g" gateway.yaml | kubectl apply -f -
第 4 步：使用端口转发验证部署并测试 MCP 服务器
由于 Google 管理的 SSL 证书和外部 Google Cloud 应用负载均衡器需要 5 到 15 分钟才能完成预配并建立 DNS 路由，因此您可以使用 kubectl port-forward 立即测试正在运行的 GKE pod。
首先，检查您的 pod 和网关是否正在运行：
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
接下来，使用端口转发在本地测试实时 GKE 服务：
- 在终端中，将本地端口
8080转发到 GKE ClusterIP 服务：
kubectl port-forward svc/mcp-server-service 8080:80
- 在第二个终端中，创建一个名为
src/mcp_server/test_vertex_tool.py的测试客户端脚本，以使用 Workload Identity 在 GKE 上调用vertex_ai_generate_content工具：
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))
- 针对本地端口转发实例运行测试脚本：
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.
这确认了 GKE Pod 运行正常，Workload Identity 正在成功地针对 Vertex AI 进行身份验证（无需静态凭证），并且可流式传输的 HTTP 传输正在按预期运行！
第 5 步：验证公共 HTTPS 网关端点
使用以下命令检查证书预配状态：
gcloud compute ssl-certificates describe mcp-server-cert --global --format="value(managed.status)"
证书处于 ACTIVE 状态后，使用带有 --host 标志的 Python 测试客户端测试公共 HTTPS 端点：
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. 将 MCP 服务器与 Google Cloud Agent Platform 集成
现在，您的 MCPServer 已安全部署到 Cloud Run 和 GKE Autopilot 上的 HTTPS 端点，请使用 Google Cloud Agent Platform（Agent Registry）注册并发现您的 MCP 工具，以便企业智能体和 AI 模型可以动态发现并调用它们。
1. 向 Agent Platform 注册自定义 Cloud Run MCP 服务器并使用服务发现功能进行测试
第 1 步：提取工具规范 (toolspec.json)
如需在 Agent Registry 中注册外部或自定义 MCP 服务器，请在实时无状态 MCP 2.0 服务器上查询 tools/list，剥离工具级 _meta 字段，并将结果保存到单个流水线中的 toolspec.json：
# 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
第 2 步：在 Agent Registry 中注册服务
使用 gcloud agent-registry services create 将您的 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"
验证服务是否已成功注册：
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
第 3 步：创建并运行服务发现测试客户端
创建一个名为 src/mcp_server/test_agent_platform.py 的 Python 脚本，该脚本可从 Agent Registry mcp-servers 目录中动态解析端点，通过 Google Cloud IAM 进行身份验证，并使用 httpx2 和 MCP 2.0 Client 调用工具：
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())
执行测试脚本：
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. 使用 GKE 自动发现和注册 MCP 服务器
Google Kubernetes Engine (GKE) 可自动与 Agent Registry 集成。通过为 GKE Deployment 清单添加标签和注解，GKE 会自动对 MCP 服务器执行内省扫描，将工具注册到目录中，并将服务投影到消费者端，而无需手动生成或上传 toolspec.json。
第 1 步：GKE 自动发现功能的运作方式
在上一部分中应用 deployment.yaml 时，以下配置启用了自动发现：
registry.gke.io/functional-type: "MCP_SERVER"：通知 GKE 集群控制器将部署注册到 Google Cloud Agent Registry。modelcontextprotocol.info/urls：为控制器自省和消费者路由提供外部 HTTPS 网关端点 (https://${MCP_DOMAIN}/mcp)。modelcontextprotocol.info/capabilities：声明 HTTP 传输端点 (/mcp)。iam.gke.io/spiffe-identity-type: agent-identity：为智能体通信配置 Workload Identity。
第 2 步：在 Agent Registry 中验证 GKE 自动注册
验证 GKE 是否已自动发现您的 MCP 服务器并将其注册到区域级 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
第 3 步：使用服务发现测试 GKE MCP 服务器
使用服务发现脚本针对 us-central1 中自动注册的服务器名称测试 GKE MCP 部署：
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. 清理资源
为避免系统因本 Codelab 中使用的资源向您的 Google Cloud 账号收取费用，请删除 Cloud Run 服务、GKE 集群、服务账号和容器映像库。
删除 Cloud Run 服务
gcloud run services delete secure-mcp-server \
--platform=managed \
--region=us-central1 \
--quiet \
--project="${PROJECT_ID}"
删除 GKE Autopilot 集群
gcloud container clusters delete mcp-gke-cluster \
--location=us-central1 \
--quiet \
--project="${PROJECT_ID}"
删除服务账号和 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}"
删除 Artifact Registry 代码库
gcloud artifacts repositories delete mcp-servers \
--location=us-central1 \
--quiet \
--project="${PROJECT_ID}"
删除 Agent Registry 自定义服务
gcloud agent-registry services delete secure-mcp-server \
--location=global \
--quiet \
--project="${PROJECT_ID}"
删除 Cloud Storage 存储桶
gcloud storage rm --recursive "gs://${BUCKET_NAME}" --quiet
删除静态 IP 和 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}"
验证删除清理是否已完成：
Deleted service [secure-mcp-server]. Deleted cluster [mcp-gke-cluster]. Deleted repository [mcp-servers].
9. 恭喜
恭喜！您已成功使用 MCP Python SDK 中的 MCPServer 和 Google Cloud 上的 uv 构建、保护和部署 MCP 2.0 服务器！
所学内容
- 构建了通过无状态可流式 HTTP 支持 MCP 规范 (2026-07-28) 的
MCPServer。 - 构建了 4 个集成 Vertex AI Gemini、Cloud Storage、Cloud Logging 和 Resource Health 的 Google Cloud 生产工具。
- 在 Cloud Run 上部署安全容器，并强制执行 IAM 身份验证和 SSL/TLS。
- 在 GKE Autopilot 上配置了无密钥身份验证，并使用了 Workload Identity 和 Kubernetes Gateway API with TLS。
- 使用 OIDC 授权标头通过 Gemini Enterprise Agent Platform (Agent Registry) 注册和发现 MCP 服务器端点。
后续步骤
- 如需了解高级资源和提示定义，请参阅 Model Context Protocol 规范。
- 详细了解 MCP Python SDK。
- 深入了解 Google Cloud Run 安全性和 GKE Workload Identity。