این صفحه به‌وسیله ‏Cloud Translation API‏ ترجمه شده است.

ساخت، ایمن‌سازی و استقرار یک سرور MCP در فضای ابری گوگل

۱. قبل از شروع

پروتکل زمینه مدل (MCP) یک استاندارد باز است که به مدل‌ها و عامل‌های هوش مصنوعی امکان دسترسی ایمن به ابزارها، پایگاه‌های داده و زمینه سازمانی را می‌دهد. از زمان معرفی آن در سال 2024، این پروتکل به طور گسترده توسط ارائه‌دهندگان Cloud و LLM پذیرفته شده است. جدیدترین مشخصات MCP Spec 2026-07-28 (MCP 2.0) یک معماری بدون وضعیت، انتقال‌های ساده‌شده و تایپ دقیق نتایج را تعریف می‌کند، در حالی که با نسخه‌های قبلی سازگار است.

کیت توسعه نرم‌افزار رسمی MCP Python ( mcp>=2.0.0 ) MCPServer ( mcp.server.mcpserver.MCPServer ) را ارائه می‌دهد که جایگزین FastMCP در MCP 2.0 شده است و به عنوان چارچوبی با کارایی بالا و مناسب برای توسعه‌دهندگان، سرورهای MCP آماده برای تولید را مطابق با مشخصات مدرن MCP (2026-07-28) با HTTP قابل پخش و انتقال رویدادهای ارسالی از سرور (SSE) از طریق SSL/TLS می‌سازد.

در این آزمایشگاه کد، شما یک سرور MCP در سطح تولید با استفاده از MCPServer از MCP 2.0 Python SDK و مدیریت وابستگی uv خواهید ساخت. شما سرور MCP خود را به چهار ابزار Google Cloud (فراخوانی Vertex AI Gemini، بازرسی Google Cloud Storage، نوشتن حسابرسی Cloud Logging و بررسی سلامت منابع Google Cloud) مجهز خواهید کرد. سپس، سرور را کانتینرایز کرده و آن را در دو هدف زمان اجرای Google Cloud مستقر خواهید کرد: Cloud Run و Google Kubernetes Engine (GKE) Autopilot ، و سرور MCP را در Gemini Enterprise Agent Platform ثبت و استفاده خواهید کرد.

نکات برجسته MCP Spec 2026-07-28 و SDK v2.0 :

MCPServer جایگزین FastMCP می‌شود: در SDK رسمی پایتون mcp>=2.0.0 ، MCPServer ( from mcp.server.mcpserver import MCPServer ) جایگزین FastMCP می‌شود و در عین حال API آشنای مبتنی بر دکوراتور ( @mcp.tool ، @mcp.custom_route ) را حفظ می‌کند.
معماری بدون وضعیت (Stateless) : مراحل اولیه‌سازی اتصال ( initialize / initialized ) را حذف می‌کند. هر درخواست، نسخه پروتکل و قابلیت‌های کلاینت را در هدرها و _meta حمل می‌کند.
نقطه پایانی کشف : server/discover را برای مذاکره در مورد قابلیت‌ها و نسخه‌ها پیاده‌سازی می‌کند.
انتقال جریانی : انتقال ساده‌شده HTTP/SSE با حذف هدرهای Mcp-Session-Id برای هر اتصال.
نوع‌بندی نتیجه : تمام نتایج اجرای ابزار شامل یک resultType اجباری ( "complete" یا "input_required" ) هستند.

کاری که انجام خواهید داد

با استفاده از پایتون ۳.۱۲+ و uv مطابق با MCP Spec 2026-07-28 ، یک MCPServer با ۴ ابزار Google Cloud بسازید.
سرور MCP را با استفاده از یک ساخت چند مرحله‌ای داکر، کانتینرایز کنید.
سرور MCP را با احراز هویت اجباری IAM و SSL/TLS در Cloud Run ایمن و مستقر کنید.
سرور MCP را با استفاده از Workload Identity و Kubernetes Gateway API به همراه TLS ، ایمن‌سازی و روی GKE Autopilot مستقر کنید.
نقطه پایانی امن سرور MCP را با استفاده از هدرهای توکن حامل OIDC در پلتفرم Gemini Enterprise Agent ثبت کنید.

آنچه نیاز دارید

یک پروژه گوگل کلود با قابلیت پرداخت.
کیت توسعه نرم‌افزار گوگل کلود ( gcloud CLI) نصب و پیکربندی شده است.
پایتون ۳.۱۲+ و مدیر بسته uv نصب شده باشند.
docker نصب شد.
ابزار خط فرمان kubectl نصب شده است.

۲. محیط ابری گوگل را راه‌اندازی کنید

قبل از ایجاد منابع، محیط خود را احراز هویت کنید و APIهای لازم Google Cloud را فعال کنید.

احراز هویت gcloud CLI

وارد حساب گوگل کلود خود شوید:

gcloud auth login

شناسه پروژه فعال Google Cloud خود را تنظیم کنید:

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

فعال کردن سرویس‌های ابری گوگل

تمام 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.

اعتبارنامه‌های پیش‌فرض برنامه را تأیید کنید

محیط خود را احراز هویت کنید تا کتابخانه‌های کلاینت پایتون بتوانند در طول توسعه به صورت محلی به Vertex AI و Cloud Storage دسترسی داشته باشند:

gcloud auth application-default login

۳. ساخت سرور MCP با MCPServer و uv

در این مرحله، شما یک پروژه پایتون را با استفاده از uv راه‌اندازی می‌کنید و یک MCPServer می‌سازید که از چهار قابلیت 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"]

همگام‌سازی وابستگی‌ها با استفاده از 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,
    )

پیاده‌سازی ابزارها در این Codelab :

ابزارهای زنده‌ی GCP SDK : vertex_ai_generate_content ، gcs_bucket_inspector و cloud_logging_audit_writer با استفاده از کتابخانه‌های کلاینت گوگل کلود مربوطه ( google-genai ، google-cloud-storage و google-cloud-logging ) فراخوانی‌های زنده‌ی API را انجام می‌دهند.
ابزار شبیه‌سازی استاتیک : gcp_resource_health_checker عمداً به عنوان یک پاسخ استاتیک سبک نوشته شده است. این ابزار، بررسی سلامت بدون وابستگی فوری را برای کلاینت‌های آزمایشی بدون نیاز به سهمیه‌های اضافی Cloud Monitoring یا Resource Manager API فراهم می‌کند. در محیط عملیاتی، این قابلیت را می‌توان با google-cloud-resource-manager یا Cloud Monitoring Uptime Checks گسترش داد.

سرور MCP را به صورت محلی آزمایش کنید

مرحله 1: سرور MCP را راه اندازی کنید

در ترمینال اصلی خود، سرور را با استفاده از uv اجرا کنید:

uv run python -m src.mcp_server.server

شما باید لاگ‌های راه‌اندازی را مشاهده کنید که تأیید می‌کنند سرور Streamable HTTP روی پورت ۸۰۸۰ در حال گوش دادن است:

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 Spec 2026-07-28 )، هدرهای stateful initialize handshake و Mcp-Session-Id با درخواست‌های stateless جایگزین شده‌اند. می‌توانید با ارسال هدرهای MCP-Protocol-Version و Mcp-Method به همراه فراداده‌های کلاینت در params._meta ، مستقیماً 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 را در ترمینال اصلی خود فشار دهید تا سرور محلی متوقف شود.

۴. سرور MCP را کانتینرایز کنید

برای استقرار سرور MCP در Cloud Run و GKE Autopilot، سرور را با استفاده از یک Dockerfile چند مرحله‌ای که توسط uv پشتیبانی می‌شود، در یک تصویر کانتینر مینیمال بسته‌بندی کنید.

ایجاد داکرفایل

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

ساخت و ارسال تصویر به رجیستری مصنوعات

یک مخزن رجیستری مصنوعات ایجاد کنید:

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

پس از اتمام، تصویر کانتینر شما به طور ایمن در رجیستری مصنوعات ذخیره می‌شود:

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

۵. استقرار در Cloud Run با IAM و HTTPS

Cloud Run یک محیط کاملاً مدیریت‌شده بدون سرور با قابلیت خاتمه خودکار گواهینامه HTTPS/SSL و کنترل دقیق دسترسی Cloud IAM فراهم می‌کند.

ایجاد حساب کاربری سرویس اختصاصی

یک حساب کاربری سرویس گوگل با حداقل دسترسی برای 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

کانتینر را با احراز هویت اجباری IAM ( --no-allow-unauthenticated ) و SSL مدیریت‌شده پیش‌فرض، روی Cloud Run مستقر کنید:

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

امنیت نقطه پایانی را تأیید کنید

تلاش برای ارسال درخواست غیرمجاز به نقطه پایانی 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

برای تأیید ارتباط مجاز، یک توکن OIDC ID با استفاده از 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",...}]}}

اجرای ابزار تست زنده: بازرس ذخیره‌سازی ابری

اکنون که سرور مجازی MCP از راه دور احراز هویت شده و پاسخ می‌دهد، اجرای ابزار gcs_bucket_inspector را در زیرساخت Google Cloud آزمایش کنید.

مرحله ۱: ایجاد یک سطل ذخیره‌سازی ابری آزمایشی

یک سطل آزمایشی ایجاد کنید و یک فایل نمونه آپلود کنید:

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"

مرحله ۲: ایجاد کلاینت تست ابزار Remote 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())

مرحله ۳: اجرای کلاینت تست 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":"..."}]}

۶. استقرار در GKE Autopilot با هویت بار کاری و TLS

برای بارهای کاری Kubernetes، GKE Autopilot مدیریت تأمین گره را بر عهده دارد در حالی که Workload Identity کلیدهای حساب سرویس استاتیک را حذف می‌کند.

خوشه خلبان خودکار GKE را ارائه دهید

ارائه یک کلاستر خلبان خودکار GKE:

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

تنظیم هویت حجم کار

یک حساب سرویس گوگل (GSA) و یک حساب سرویس کوبرنتیز (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

مرحله ۱: رزرو IP استاتیک و ارائه گواهی SSL مدیریت‌شده توسط گوگل با nip.io

یک آدرس IP خارجی جهانی رزرو کنید و از سرویس DNS wildcard به نام nip.io استفاده کنید ( . .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 آماده‌سازی خود را به پایان برساند - در واقع، گواهی‌های مدیریت‌شده توسط گوگل تا زمانی که در مرحله ۳ به متعادل‌کننده بار دروازه متصل نشوند، در حالت PROVISIONING باقی می‌مانند. بلافاصله به مراحل بعدی بروید.

مرحله 2: ایجاد مانیفست استقرار و سرویس

deployment.yaml که شامل تعاریف Deployment و سرویس داخلی است، ایجاد کنید. از آنجا که MCP 2.0 ( MCP Spec 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: ایجاد Gateway API، HealthCheckPolicy و GCPBackendPolicy Manifest

gateway.yaml که شامل Gateway، HTTPRoute، HealthCheckPolicy و GCPBackendPolicy است، ایجاد کنید:

HealthCheckPolicy : متعادل‌کننده بار Google Cloud را برای بررسی /healthz روی پورت ۸۰۸۰ پیکربندی می‌کند.
GCPBackendPolicy (زمان انتظار بک‌اند) : timeoutSec: 300 را تنظیم می‌کند تا با زمان انتظار خواندن ۳۰۰ ثانیه‌ای MCP SDK v2 برای استریم‌های HTTP/SSE قابل پخش هماهنگ شود (در صورت حذف، GKE Gateway به طور پیش‌فرض روی ۳۰ ثانیه تنظیم می‌شود).

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 -

مرحله ۴: تأیید استقرار و آزمایش سرور MCP با استفاده از Port-Forward

از آنجا که گواهینامه‌های SSL تحت مدیریت گوگل و متعادل‌کننده‌های بار خارجی برنامه ابری گوگل، ۵ تا ۱۵ دقیقه برای آماده‌سازی و ایجاد مسیریابی DNS زمان نیاز دارند، می‌توانید بلافاصله پادهای 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 زنده را به صورت محلی با استفاده از پورت فورواردینگ آزمایش کنید:

در ترمینال خود، پورت محلی 8080 را به سرویس GKE ClusterIP ارسال کنید:

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

در ترمینال دوم، یک اسکریپت کلاینت آزمایشی با نام 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))

اسکریپت آزمایشی را روی نمونه‌ی محلیِ پورت‌فوروارد شده اجرا کنید:

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 سالم هستند، هویت بار کاری با موفقیت در برابر Vertex AI بدون اعتبارنامه‌های استاتیک احراز هویت می‌شود و انتقال HTTP قابل پخش مطابق انتظار عمل می‌کند!

مرحله ۵: تأیید نقطه پایانی دروازه عمومی HTTPS

وضعیت ارائه گواهی را با موارد زیر بررسی کنید:

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

پس از ACTIVE گواهی، نقطه پایانی عمومی HTTPS را با استفاده از کلاینت آزمایشی پایتون با پرچم --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.

۷. سرور MCP را با پلتفرم Google Cloud Agent ادغام کنید

اکنون که MCPServer شما به طور ایمن در نقاط انتهایی HTTPS در Cloud Run و GKE Autopilot مستقر شده است، ابزارهای MCP خود را با Google Cloud Agent Platform (Agent Registry) ثبت و کشف کنید تا نمایندگان سازمانی و مدل‌های هوش مصنوعی بتوانند به صورت پویا آنها را کشف و فراخوانی کنند.

۱. سرور ابری سفارشی MCP Run را در پلتفرم عامل ثبت کنید و با استفاده از کشف سرویس (Service Discovery) آن را آزمایش کنید.

مرحله ۱: استخراج مشخصات ابزار ( `toolspec.json` )

برای ثبت یک سرور MCP خارجی یا سفارشی در Agent Registry ، 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

مرحله ۳: ایجاد و اجرای کلاینت تست کشف سرویس

یک اسکریپت پایتون با نام src/mcp_server/test_agent_platform.py ایجاد کنید که به صورت پویا نقطه پایانی را از کاتالوگ mcp-servers رجیستری Agent دریافت کند، با 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"}

۲. کشف و ثبت خودکار سرور MCP با GKE

موتور کوبرنتیز گوگل (GKE) یکپارچه‌سازی خودکار با رجیستری عامل (Agent Registry) را فراهم می‌کند. با برچسب‌گذاری و حاشیه‌نویسی مانیفست استقرار GKE شما، GKE به‌طور خودکار یک اسکن درون‌نگرانه روی سرور MCP انجام می‌دهد، ابزارها را در کاتالوگ ثبت می‌کند و سرویس را بدون نیاز به تولید یا آپلود دستی toolspec.json در سمت مصرف‌کننده پیاده‌سازی می‌کند.

مرحله ۱: نحوه عملکرد کشف خودکار GKE

وقتی deployment.yaml شما در بخش قبلی اعمال شد، پیکربندی‌های زیر امکان کشف خودکار را فراهم کردند:

registry.gke.io/functional-type: "MCP_SERVER" : به کنترلر خوشه GKE اطلاع می‌دهد که استقرار را در Google Cloud Agent Registry ثبت کند.
modelcontextprotocol.info/urls : نقطه پایانی خارجی HTTPS Gateway ( https://${MCP_DOMAIN}/mcp ) را برای درون‌نگری کنترلر و مسیریابی مصرف‌کننده فراهم می‌کند.
modelcontextprotocol.info/capabilities : نقطه پایانی انتقال HTTP ( /mcp ) را اعلام می‌کند.
iam.gke.io/spiffe-identity-type: agent-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 با استفاده از Service Discovery

استقرار 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"}

۸. منابع را پاکسازی کنید

برای جلوگیری از تحمیل هزینه به حساب Google Cloud خود برای منابع استفاده شده در این آزمایشگاه کد، سرویس Cloud Run، کلاستر GKE، حساب‌های سرویس و مخزن کانتینر را حذف کنید.

سرویس Cloud Run را حذف کنید

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

خوشه خلبان خودکار GKE را حذف کنید

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

مخزن رجیستری مصنوعات را حذف کنید

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

حذف سطل ذخیره‌سازی ابری

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

۹. تبریک

تبریک! شما با موفقیت یک سرور MCP 2.0 را با استفاده MCPServer از MCP Python SDK و uv در Google Cloud ساختید، ایمن‌سازی کردید و مستقر نمودید!

آنچه را که پوشش دادید

یک MCPServer ساختم که از MCP Spec (2026-07-28) روی HTTP Streamable بدون وضعیت پشتیبانی می‌کند.
۴ ابزار ابری گوگل تولید کرده که Vertex AI Gemini، Cloud Storage، Cloud Logging و Resource Health را با هم ادغام می‌کنند.
استقرار کانتینرهای امن در Cloud Run با احراز هویت اجباری IAM و SSL/TLS.
احراز هویت بدون مخفی‌کاری روی GKE Autopilot با Workload Identity و Kubernetes Gateway API با TLS پیکربندی شد.
نقاط پایانی سرور MCP با استفاده از هدرهای مجوز OIDC در پلتفرم عامل سازمانی Gemini (ثبت عامل) ثبت و کشف شدند.

مراحل بعدی

برای منابع پیشرفته و تعاریف سریع ، مشخصات پروتکل زمینه مدل (Model Context Protocol Specification) را بررسی کنید.
درباره MCP Python SDK بیشتر بدانید.
بررسی عمیق امنیت Google Cloud Run و هویت حجم کار GKE .