สร้าง รักษาความปลอดภัย และติดตั้งใช้งานเซิร์ฟเวอร์ MCP ใน Google Cloud

1. ก่อนเริ่มต้น

Model Context Protocol (MCP) เป็นมาตรฐานแบบเปิดที่ช่วยให้โมเดลและ Agent AI เข้าถึงเครื่องมือ ฐานข้อมูล และบริบทขององค์กรได้อย่างปลอดภัย ตั้งแต่เปิดตัวในปี 2024 โปรโตคอลนี้ได้รับการยอมรับอย่างกว้างขวางจากผู้ให้บริการระบบคลาวด์และ LLM ข้อกำหนดล่าสุด MCP Spec 2026-07-28 (MCP 2.0) กำหนดสถาปัตยกรรมแบบไม่เก็บสถานะ การรับส่งที่เรียบง่าย และการพิมพ์ผลลัพธ์ที่เข้มงวด ในขณะที่ยังคงความเข้ากันได้แบบย้อนหลังกับเวอร์ชันก่อนหน้า

MCP Python SDK อย่างเป็นทางการ (mcp>=2.0.0) มี MCPServer (mcp.server.mcpserver.MCPServer) ซึ่งจะแทนที่ FastMCP ใน MCP 2.0 ในฐานะเฟรมเวิร์กที่มีประสิทธิภาพสูงและเป็นมิตรกับนักพัฒนาซอฟต์แวร์สำหรับการสร้างเซิร์ฟเวอร์ MCP ที่พร้อมใช้งานจริงตามข้อกำหนด MCP (2026-07-28) สมัยใหม่ที่มีการรับส่ง HTTP แบบสตรีมและเหตุการณ์ที่เซิร์ฟเวอร์ส่ง (SSE) ผ่าน SSL/TLS

ใน Codelab นี้ คุณจะได้สร้างเซิร์ฟเวอร์ MCP ระดับเวอร์ชันที่ใช้งานจริงโดยใช้ MCPServer จาก MCP 2.0 Python SDK และการจัดการการขึ้นต่อกันของ uv คุณจะติดตั้งเครื่องมือ Google Cloud 4 อย่าง (การเรียกใช้ Vertex AI Gemini, การตรวจสอบ Google Cloud Storage, การเขียนการตรวจสอบ Cloud Logging และการตรวจสอบสถานะทรัพยากร Google Cloud) ให้กับเซิร์ฟเวอร์ MCP จากนั้นคุณจะสร้างคอนเทนเนอร์ของเซิร์ฟเวอร์และติดตั้งใช้งานในเป้าหมายรันไทม์ 2 รายการของ Google Cloud ได้แก่ Cloud Run และ Google Kubernetes Engine (GKE) Autopilot รวมถึงลงทะเบียนและใช้เซิร์ฟเวอร์ MCP ใน Gemini Enterprise Agent Platform

สิ่งที่คุณจะทำ

  • สร้าง MCPServer ด้วยเครื่องมือ 4 อย่างของ Google Cloud โดยใช้ Python 3.12 ขึ้นไปและ uv ตาม MCP Spec 2026-07-28
  • สร้างคอนเทนเนอร์เซิร์ฟเวอร์ MCP โดยใช้การสร้าง Docker แบบหลายขั้นตอน
  • รักษาความปลอดภัยและติดตั้งใช้งานเซิร์ฟเวอร์ MCP ใน Cloud Run ด้วยการตรวจสอบสิทธิ์ IAM และ SSL/TLS ที่บังคับใช้
  • รักษาความปลอดภัยและทำให้เซิร์ฟเวอร์ MCP ใช้งานได้ใน GKE Autopilot โดยใช้ Workload Identity และ Kubernetes Gateway API ที่มี TLS
  • ลงทะเบียนปลายทางเซิร์ฟเวอร์ MCP ที่ปลอดภัยกับ Gemini Enterprise Agent Platform โดยใช้ส่วนหัวของโทเค็นผู้ถือ OIDC

สิ่งที่คุณต้องมี

  • โปรเจ็กต์ Google Cloud ที่เปิดใช้การเรียกเก็บเงิน
  • ติดตั้งและกำหนดค่า Google Cloud SDK (gcloud CLI) แล้ว
  • ติดตั้ง Python 3.12 ขึ้นไปและเครื่องมือจัดการแพ็กเกจ uv
  • ติดตั้ง docker แล้ว
  • kubectl ติดตั้งเครื่องมือบรรทัดคำสั่งแล้ว

2. ตั้งค่าสภาพแวดล้อม Google Cloud

ก่อนสร้างทรัพยากร ให้ตรวจสอบสิทธิ์สภาพแวดล้อมและเปิดใช้ Google Cloud API ที่จำเป็น

ตรวจสอบสิทธิ์ gcloud CLI

เข้าสู่ระบบบัญชี Google Cloud โดยทำดังนี้

gcloud auth login

ตั้งค่ารหัสโปรเจ็กต์ที่อยู่ในระบบคลาวด์ของ Google ที่ใช้งานอยู่

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

เปิดใช้บริการ Google Cloud

เปิดใช้ API ที่จำเป็นทั้งหมดสำหรับ Cloud Run, GKE, Vertex AI, Artifact Registry, Cloud Build, Cloud Logging, Storage และ 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}"

ตรวจสอบว่าเปิดใช้ API สำเร็จแล้วโดยทำดังนี้

Operation "operations/..." finished successfully.

ตรวจสอบสิทธิ์ข้อมูลรับรองเริ่มต้นของแอปพลิเคชัน

ตรวจสอบสิทธิ์สภาพแวดล้อมเพื่อให้ไลบรารีของไคลเอ็นต์ Python เข้าถึง Vertex AI และ Cloud Storage ได้ในเครื่องระหว่างการพัฒนา

gcloud auth application-default login

3. สร้างเซิร์ฟเวอร์ MCP ด้วย MCPServer และ uv

ในขั้นตอนนี้ คุณจะเริ่มต้นโปรเจ็กต์ Python โดยใช้ uv และสร้าง MCPServer ที่รองรับความสามารถ 4 อย่างของ Google Cloud

เริ่มต้นโปรเจ็กต์ด้วย 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"]

ซิงค์ทรัพยากร Dependency โดยใช้ 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

คุณควรเห็นบันทึกการเริ่มต้นที่ยืนยันว่าเซิร์ฟเวอร์ HTTP ของ Streamable กำลังรับฟังในพอร์ต 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: ยืนยันปลายทาง HTTP ที่สตรีมได้โดยใช้ curl

ใน MCP 2.0 (ข้อกำหนด MCP 2026-07-28) คำขอแบบไม่เก็บสถานะจะแทนที่การแฮนด์เชค initialize แบบเก็บสถานะและส่วนหัว Mcp-Session-Id คุณเรียกใช้ tools/list ได้โดยตรงด้วยการส่งส่วนหัว MCP-Protocol-Version และ Mcp-Method พร้อมกับข้อมูลเมตาของไคลเอ็นต์ใน params._meta

เปิดหน้าต่างเทอร์มินัลที่ 2 แล้วส่งคำขอ 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 ให้แพ็กเกจเซิร์ฟเวอร์เป็นอิมเมจคอนเทนเนอร์ขนาดเล็กโดยใช้ Dockerfile แบบหลายขั้นตอนที่ขับเคลื่อนโดย uv

สร้าง 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. ติดตั้งใช้งานใน Cloud Run ด้วย IAM และ HTTPS

Cloud Run มีสภาพแวดล้อมแบบ Serverless ที่มีการจัดการเต็มรูปแบบพร้อมการสิ้นสุดใบรับรอง HTTPS / SSL โดยอัตโนมัติและการควบคุมการเข้าถึง Cloud IAM แบบละเอียด

สร้างบัญชีบริการเฉพาะ

สร้างบัญชีบริการ Google ที่มีสิทธิ์น้อยที่สุดสำหรับ 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"

ติดตั้งใช้งานบริการใน 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 URL โดยทำดังนี้

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

ยืนยันความปลอดภัยของอุปกรณ์ปลายทาง

การพยายามส่งคำขอที่ไม่ได้รับอนุญาตไปยังปลายทาง HTTP ของ Streamable จะแสดงผล 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

สร้างโทเค็นรหัส OIDC โดยใช้ gcloud เพื่อยืนยันการสื่อสารที่ได้รับอนุญาต

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 ระยะไกลได้รับการตรวจสอบสิทธิ์และตอบกลับแล้ว ให้ทดสอบการเรียกใช้เครื่องมือ gcs_bucket_inspector กับโครงสร้างพื้นฐานของ Google Cloud

ขั้นตอนที่ 1: สร้าง Bucket ของ Cloud Storage สำหรับทดสอบ

สร้าง 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

คุณควรเห็นข้อมูลเมตาของออบเจ็กต์แบบเรียลไทม์ที่ส่งคืนจากเซิร์ฟเวอร์ MCP ของ 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. ทําให้ใช้งานได้ใน GKE Autopilot ด้วย Workload Identity และ TLS

สำหรับภาระงาน 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) จากนั้นลิงก์บัญชีทั้ง 2 โดยใช้ 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 และ Gateway API กับ TLS

ขั้นตอนที่ 1: จอง IP แบบคงที่และจัดสรรใบรับรอง SSL ที่จัดการโดย Google ด้วย nip.io

จองที่อยู่ IP ภายนอกแบบสากลและใช้ประโยชน์จากบริการ DNS แบบไวลด์การ์ด nip.io (..nip.io) เพื่อสร้างชื่อโดเมนสาธารณะที่ถูกต้อง (MCP_DOMAIN) ก่อนที่จะใช้ไฟล์ Manifest ของ 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 จนกว่าจะแนบกับ Gateway Load Balancer ในขั้นตอนที่ 3 ดำเนินการตามขั้นตอนถัดไปทันที

ขั้นตอนที่ 2: สร้างไฟล์ Manifest ของการติดตั้งใช้งานและบริการ

สร้าง deployment.yaml ที่มีคำจำกัดความของการติดตั้งใช้งานและบริการภายใน เนื่องจาก MCP 2.0 (ข้อกำหนด MCP 2026-07-28) เป็นแบบไม่เก็บสถานะ Service ของ Kubernetes จึงไม่จำเป็นต้องมีเซสชันแอฟฟินิตี้ 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: สร้างไฟล์ Manifest ของ Gateway API, HealthCheckPolicy และ GCPBackendPolicy

สร้าง gateway.yaml ที่มี Gateway, HTTPRoute, HealthCheckPolicy และ GCPBackendPolicy

  • HealthCheckPolicy: กำหนดค่าตัวจัดสรรภาระงานของ Google Cloud ให้ตรวจสอบ /healthz ในพอร์ต 8080
  • GCPBackendPolicy (การหมดเวลาของแบ็กเอนด์): ตั้งค่า timeoutSec: 300 ให้สอดคล้องกับการหมดเวลาในการอ่าน 300 วินาทีของ MCP SDK v2 สำหรับสตรีม 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 โดยใช้การส่งต่อพอร์ต

เนื่องจากใบรับรอง SSL ที่ Google จัดการและตัวจัดสรรภาระงานของแอปพลิเคชัน Google Cloud ภายนอกใช้เวลาในการจัดสรรและสร้างการกำหนดเส้นทาง DNS 5-15 นาที คุณจึงทดสอบพ็อด GKE ที่ทำงานได้ทันทีโดยใช้ kubectl port-forward

ก่อนอื่น ให้ตรวจสอบว่าพ็อดและเกตเวย์ทำงานอยู่ โดยทำดังนี้

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 ที่ใช้งานจริงในเครื่องโดยใช้การส่งต่อพอร์ต

  1. ในเทอร์มินัล ให้ส่งต่อพอร์ต 8080 ในเครื่องไปยังบริการ ClusterIP ของ GKE
kubectl port-forward svc/mcp-server-service 8080:80
  1. ในเทอร์มินัลที่ 2 ให้สร้างสคริปต์ไคลเอ็นต์ทดสอบชื่อ src/mcp_server/test_vertex_tool.py เพื่อเรียกใช้เครื่องมือ vertex_ai_generate_content ใน GKE โดยใช้ 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))
  1. เรียกใช้สคริปต์การทดสอบกับอินสแตนซ์ที่ส่งต่อพอร์ตในเครื่อง
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 ทำงานได้ดี, Workload Identity ตรวจสอบสิทธิ์กับ Vertex AI ได้สำเร็จโดยไม่ต้องใช้ข้อมูลเข้าสู่ระบบแบบคงที่ และการรับส่ง HTTP แบบสตรีมทำงานได้ตามที่คาดไว้

ขั้นตอนที่ 5: ยืนยันปลายทางเกตเวย์ HTTPS สาธารณะ

ตรวจสอบสถานะการจัดสรรใบรับรองด้วยคำสั่งต่อไปนี้

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

เมื่อใบรับรองACTIVEแล้ว ให้ทดสอบปลายทาง HTTPS สาธารณะโดยใช้ไคลเอ็นต์ทดสอบ Python ด้วยแฟล็ก --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. ผสานรวมเซิร์ฟเวอร์ MCP กับแพลตฟอร์มเอเจนต์ของ Google Cloud

ตอนนี้คุณได้ติดตั้งใช้งาน MCPServer อย่างปลอดภัยกับปลายทาง HTTPS ใน Cloud Run และ GKE Autopilot แล้ว ให้ลงทะเบียนและค้นพบเครื่องมือ MCP ด้วย Google Cloud Agent Platform (Agent Registry) เพื่อให้ Agent ขององค์กรและโมเดล AI ค้นพบและเรียกใช้เครื่องมือเหล่านั้นแบบไดนามิกได้

1. ลงทะเบียนเซิร์ฟเวอร์ MCP ของ Cloud Run ที่กำหนดเองกับแพลตฟอร์ม Agent และทดสอบโดยใช้การค้นพบบริการ

ขั้นตอนที่ 1: ดึงข้อมูลจำเพาะของเครื่องมือ (toolspec.json)

หากต้องการลงทะเบียนเซิร์ฟเวอร์ MCP ภายนอกหรือที่กำหนดเองในรีจิสทรี Agent ให้ค้นหา tools/list ในเซิร์ฟเวอร์ MCP 2.0 แบบไม่เก็บสถานะที่ใช้งานจริง ลบฟิลด์ _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: ลงทะเบียนบริการในรีจิสทรีของเอเจนต์

ใช้ 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: สร้างและเรียกใช้ไคลเอ็นต์ทดสอบการค้นพบบริการ

สร้างสคริปต์ Python ชื่อ src/mcp_server/test_agent_platform.py ที่จะแก้ไขปลายทางแบบไดนามิกจากแคตตาล็อกรีจิสทรีของเอเจนต์ 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. การค้นหาและการลงทะเบียนเซิร์ฟเวอร์ MCP อัตโนมัติด้วย GKE

Google Kubernetes Engine (GKE) มีการผสานรวมกับรีจิสทรีของ Agent โดยอัตโนมัติ การติดป้ายกำกับและใส่คำอธิบายประกอบในไฟล์ Manifest ของการติดตั้งใช้งาน GKE จะทำให้ 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: ยืนยันการลงทะเบียนอัตโนมัติของ 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 โดยใช้การค้นพบบริการ

ทดสอบการติดตั้งใช้งาน GKE MCP โดยใช้สคริปต์การค้นพบบริการกับชื่อเซิร์ฟเวอร์ที่ลงทะเบียนอัตโนมัติใน 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. ล้างข้อมูลทรัพยากร

หากต้องการหลีกเลี่ยงการเรียกเก็บเงินจากบัญชี Google Cloud สำหรับทรัพยากรที่ใช้ใน Codelab นี้ ให้ลบบริการ 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}"

ลบบริการที่กำหนดเองของรีจิสทรีตัวแทน

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

ลบ Bucket ของ 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 2.0 โดยใช้ MCPServer จาก MCP Python SDK และ uv ใน Google Cloud เรียบร้อยแล้ว

สิ่งที่คุณได้พูดถึง

  • สร้าง MCPServer ที่รองรับ MCP Spec (2026-07-28) ผ่าน HTTP แบบสตรีมที่ไม่มีสถานะ
  • สร้างเครื่องมือ Google Cloud 4 รายการที่พร้อมใช้งานจริงโดยผสานรวม Vertex AI Gemini, Cloud Storage, Cloud Logging และสถานะทรัพยากร
  • การติดตั้งใช้งานคอนเทนเนอร์ที่ปลอดภัยใน Cloud Run พร้อมการตรวจสอบสิทธิ์ IAM และ SSL/TLS ที่บังคับใช้
  • กำหนดค่าการตรวจสอบสิทธิ์แบบไม่ใช้ความลับใน GKE Autopilot ด้วย Workload Identity และ Kubernetes Gateway API พร้อม TLS
  • ลงทะเบียนและค้นหาปลายทางเซิร์ฟเวอร์ MCP ด้วย Gemini Enterprise Agent Platform (Agent Registry) โดยใช้ส่วนหัวการให้สิทธิ์ OIDC

ขั้นตอนถัดไป