1. 事前準備
Model Context Protocol (MCP) 是一項開放標準，可讓 AI 模型和代理安全存取工具、資料庫和企業脈絡。自 2024 年推出以來，這項通訊協定已獲得雲端和 LLM 供應商廣泛採用。最新規格 MCP 規格 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)。
在本程式碼研究室中，您將使用 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，建構符合 MCP 規格 2026-07-28 的MCPServer，並包含 4 個 Google Cloud 工具。 - 使用多級式 Docker 建構作業，將 MCP 伺服器容器化。
- 將 MCP 伺服器部署至 Cloud Run，並強制執行 IAM 驗證和 SSL/TLS。
- 使用 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 測試用戶端
如要測試透過 Streamable 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}"
將必要的 IAM 權限授予使用者帳戶和預設 Compute Engine 服務帳戶，讓 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 Inspector
遠端 MCP 伺服器通過驗證並回應後，請測試對 Google Cloud 基礎架構執行 gcs_bucket_inspector 工具。
步驟 1：建立測試 Cloud Storage bucket
建立測試 bucket 並上傳範例檔案：
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}"
使用 TLS 部署 Kubernetes Deployment 和 Gateway API
步驟 1：預留靜態 IP，並使用 nip.io 佈建 Google 代管的 SSL 憑證
保留全域外部 IP 位址，並運用萬用字元 DNS 服務 nip.io (MCP_DOMAIN)，然後再套用 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}"
您不必等待 SSL 憑證佈建完成，即可繼續操作。事實上，Google 代管憑證會維持 PROVISIONING 狀態，直到在步驟 3 中附加至閘道負載平衡器為止。立即進行後續步驟。
步驟 2：建立部署作業和服務資訊清單
建立包含 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 和 Gateway 正在執行：
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後，請使用 Python 測試用戶端和 --host 旗標測試公開 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. 清理資源
如要避免系統向您的 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 bucket
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 伺服器！
涵蓋內容
- 建構
MCPServer，透過無狀態 Streamable HTTP 支援 MCP 規格 (2026-07-28)。 - 建構 4 項 Google Cloud 生產工具，整合 Vertex AI Gemini、Cloud Storage、Cloud Logging 和 Resource Health。
- 在 Cloud Run 上安全地部署容器，並強制執行 IAM 驗證和 SSL/TLS。
- 在 GKE Autopilot 上設定無密碼驗證，並搭配 Workload Identity 和 Kubernetes Gateway API (含 TLS)。
- 使用 OIDC 授權標頭，透過 Gemini Enterprise Agent Platform (Agent Registry) 註冊及探索 MCP 伺服器端點。
後續步驟
- 如需進階資源和提示定義，請參閱 Model Context Protocol 規格。
- 進一步瞭解 MCP Python SDK。
- 深入瞭解 Google Cloud Run 安全性和 GKE Workload Identity。