إنشاء خادم MCP وتأمينه ونشره على Google Cloud

1. قبل البدء

بروتوكول سياق النموذج (MCP) هو معيار مفتوح يتيح لنماذج الذكاء الاصطناعي والوكلاء الوصول بأمان إلى الأدوات وقواعد البيانات وسياق المؤسسة. منذ طرح البروتوكول في عام 2024، اعتمدته على نطاق واسع شركات توفير الخدمات السحابية ونماذج اللغات الكبيرة. يحدّد أحدث مواصفات MCP Spec 2026-07-28 (الإصدار 2.0 من MCP) بنية غير مرتبطة بحالة، وعمليات نقل مبسطة، وتصنيفًا صارمًا للنتائج، مع الحفاظ على التوافق مع الإصدارات السابقة.

توفّر حزمة تطوير البرامج (SDK) الرسمية بلغة Python لمنصة MCP (mcp>=2.0.0) MCPServer (mcp.server.mcpserver.MCPServer)، والتي تحلّ محلّ FastMCP في الإصدار 2.0 من منصة MCP كإطار عمل عالي الأداء وسهل الاستخدام للمطوّرين من أجل إنشاء خوادم MCP جاهزة للإنتاج تتوافق مع مواصفات MCP الحديثة (28-07-2026) مع عمليات نقل HTTP قابلة للبث وأحداث مرسَلة من الخادم (SSE) عبر SSL/TLS.

في هذا الدرس التطبيقي حول الترميز، ستنشئ خادم MCP يمكن استخدامه في بيئة إنتاج باستخدام MCPServer من حزمة تطوير البرامج (SDK) MCP 2.0 Python و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.

الإجراءات التي ستنفذّها

  • أنشئ MCPServer باستخدام 4 أدوات من Google Cloud باستخدام Python 3.12 والإصدارات الأحدث وuv يتوافق مع مواصفات MCP‏ 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 تم تفعيل الفوترة فيه
  • تم تثبيت حزمة تطوير البرامج (SDK) من Google Cloud (gcloud CLI) وضبطها.
  • يجب تثبيت الإصدار 3.12 أو إصدار أحدث من Python وأداة إدارة الحِزم uv.
  • تم تثبيت "docker".
  • تم تثبيت أداة سطر الأوامر kubectl.

2. إعداد بيئة Google Cloud

قبل إنشاء الموارد، يجب مصادقة بيئتك وتفعيل واجهات Google Cloud APIs اللازمة.

مصادقة gcloud CLI

سجِّل الدخول إلى حسابك على Google Cloud:

gcloud auth login

اضبط رقم تعريف مشروع على السحابة الإلكترونية النشط في Google Cloud:

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

تفعيل "خدمات Google Cloud"

فعِّل جميع واجهات برمجة التطبيقات المطلوبة لخدمات Cloud Run وGKE وVertex AI وArtifact Registry وCloud Build وCloud Logging وStorage وCompute Engine:

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

تأكَّد من تفعيل واجهات برمجة التطبيقات بنجاح:

Operation "operations/..." finished successfully.

المصادقة على "بيانات الاعتماد التلقائية للتطبيق"

صادِق على بيئتك لكي تتمكّن مكتبات برامج Python من الوصول إلى Vertex AI وCloud Storage محليًا أثناء عملية التطوير:

gcloud auth application-default login

3- إنشاء خادم MCP باستخدام MCPServer وuv

في هذه الخطوة، ستنشئ مشروع Python باستخدام 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,
    )

اختبار خادم 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.

افتح نافذة محطة طرفية ثانية وأرسِل طلب HTTP POST قابلاً للبث:

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

ستردّ MCPServer بالرمز HTTP/1.1 200 OK وتعرض نتيجة JSON-RPC tools/list مباشرةً:

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

content-type: application/json

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

الخطوة 3: تشغيل "عميل اختبار MCP"

لاختبار إمكانية العثور على الأدوات وتنفيذها على خادمك المحلي عبر Streamable HTTP، أنشئ نصًا برمجيًا لاختبار العميل src/mcp_server/test_client.py:

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


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

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

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

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


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

شغِّل نص برمجي لعميل الاختبار باستخدام uv:

uv run python src/mcp_server/test_client.py

من المفترض أن تظهر لك نتيجة توضّح نجاح عملية الربط واكتشاف الأداة وتنفيذها:

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

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

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

بعد التحقّق، اضغط على CTRL+C في نافذة الوحدة الطرفية الرئيسية لإيقاف الخادم المحلي.

4. تضمين خادم MCP

لنشر خادم بروتوكول سياق النموذج (MCP) على Cloud Run وGKE Autopilot، يجب تجميع الخادم في صورة حاوية بسيطة باستخدام ملف 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 باستخدام إدارة الهوية وإمكانية الوصول وبروتوكول HTTPS

يوفّر Cloud Run بيئة مُدارة بالكامل بدون خادم مع إنهاء تلقائي لشهادة 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 مع فرض مصادقة "إدارة الهوية وإمكانية الوصول" (--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}"

استرداد عنوان URL يستخدم HTTPS:

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

التحقّق من أمان نقطة النهاية

ستؤدي محاولة إرسال طلب غير مصرّح به إلى نقطة نهاية 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: إنشاء حزمة في Test Cloud Storage

أنشئ مجموعة اختبار وحمِّل ملفًا نموذجيًا:

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

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

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

الخطوة 2: إنشاء عميل اختبار "أداة GCS عن بُعد"

أنشئ نصًا برمجيًا لاختبار العميل باسم src/mcp_server/test_gcs_tool.py للمصادقة على Cloud Run واستدعاء الأداة gcs_bucket_inspector:

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


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

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

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

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

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

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

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

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


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

الخطوة 3: تشغيل Remote GCS Test Client

شغِّل نص الاختبار البرمجي باستخدام 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 عملية توفير العُقد، بينما تزيل ميزة Workload Identity مفاتيح حساب الخدمة الثابتة.

توفير مجموعة GKE Autopilot

توفير مجموعة GKE Autopilot:

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

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

إعداد Workload Identity

أنشئ حساب خدمة Google (GSA) وحساب خدمة Kubernetes (KSA)، ثم اربطهما باستخدام Workload Identity:

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

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

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

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

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

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

تطبيق واجهة برمجة التطبيقات الخاصة بالنشر والبوابة في Kubernetes مع بروتوكول أمان طبقة النقل (TLS)

الخطوة 1: حجز عنوان IP ثابت وتوفير شهادة طبقة المقابس الآمنة التي تديرها Google باستخدام nip.io

احجز عنوان IP خارجيًا عامًا واستفِد من خدمة نظام أسماء النطاقات (DNS) ذات الأحرف البَديلة 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 قبل المتابعة. في الواقع، تبقى الشهادات المُدارة من Google في الحالة PROVISIONING إلى أن يتم ربطها بخدمة موازنة التحميل لبوابة الإنترنت في الخطوة 3. يُرجى الانتقال فورًا إلى الخطوات التالية.

الخطوة 2: إنشاء ملف بيان النشر والخدمة

أنشئ ملف deployment.yaml يحتوي على تعريفات النشر والخدمة الداخلية. بما أنّ الإصدار 2.0 من MCP (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

أنشئ gateway.yaml يحتوي على Gateway وHTTPRoute وHealthCheckPolicy وGCPBackendPolicy:

  • استبدِل HealthCheckPolicy بما يلي: يضبط هذا الأمر جهاز موازنة الحمل في Google Cloud لفحص /healthz على المنفذ 8080.
  • GCPBackendPolicy (Backend Timeout): يضبط timeoutSec: 300 ليتوافق مع مهلة القراءة البالغة 300 ثانية في الإصدار 2 من حزمة MCP SDK لجداول بيانات HTTP / SSE القابلة للبث (يتم ضبط بوابة GKE تلقائيًا على 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 يستغرق من 5 إلى 15 دقيقة، يمكنك اختبار وحدات GKE التي تعمل باستخدام kubectl port-forward على الفور.

أولاً، تأكَّد من أنّ وحداتك وGateway تعملان:

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

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

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

بعد ذلك، اختبِر خدمة GKE المباشرة محليًا باستخدام ميزة "توجيه المنفذ":

  1. في نافذة الأوامر، أعِد توجيه المنفذ المحلي 8080 إلى خدمة ClusterIP في GKE:
kubectl port-forward svc/mcp-server-service 8080:80
  1. في نافذة طرفية ثانية، أنشئ نصًا برمجيًا لعميل اختبار باسم 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 (سجلّ الوكلاء) حتى تتمكّن وكلاء المؤسسات ونماذج الذكاء الاصطناعي من اكتشافها واستدعائها بشكل ديناميكي.

1. تسجيل خادم MCP مخصّص على Cloud Run في Agent Platform واختباره باستخدام Service Discovery

الخطوة 1: استخراج مواصفات الأداة (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

الخطوة 3: إنشاء عميل اختبار "اكتشاف الخدمات" وتشغيله

أنشئ نصًا برمجيًا بلغة Python باسم src/mcp_server/test_agent_platform.py يحلّ نقطة النهاية ديناميكيًا من فهرس mcp-servers في "سجلّ الوكلاء"، ويصادق على الهوية باستخدام خدمة "إدارة الهوية وإمكانية الوصول" (IAM) في Google Cloud، ويستدعي أداة باستخدام 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 Registry. من خلال تصنيف بيان نشر 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: اختبار خادم MCP في GKE باستخدام "اكتشاف الخدمات"

اختبِر عملية نشر 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 مقابل الموارد المستخدَمة في هذا الدرس التطبيقي حول الترميز، احذف خدمة 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}"

حذف حسابات الخدمة وعمليات الربط في "إدارة الهوية وإمكانية الوصول"

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

Delete Agent Registry Custom Service

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

حذف حزمة Cloud Storage

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

حذف عنوان IP الثابت وشهادة SSL

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

التحقّق من اكتمال عملية تنظيف الحذف:

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

9. تهانينا

تهانينا! لقد تمكّنت بنجاح من إنشاء خادم MCP 2.0 وتأمينه ونشره باستخدام MCPServer من حزمة تطوير البرامج (SDK) الخاصة بلغة Python في MCP وuv على Google Cloud.

المواضيع التي تناولتها

  • تم إنشاء MCPServer يتيح مواصفات MCP (بتاريخ 2026-07-28) عبر بروتوكول HTTP بدون حالة وقابل للبث.
  • تم إنشاء 4 أدوات إنتاجية على Google Cloud تدمج Vertex AI Gemini وCloud Storage وCloud Logging وResource Health.
  • عمليات نشر الحاويات الآمنة على Cloud Run مع فرض مصادقة إدارة الهوية وإمكانية الوصول (IAM) وبروتوكول أمان طبقة النقل (TLS)/طبقة المقابس الآمنة (SSL)
  • تم ضبط المصادقة بدون استخدام أسرار على GKE Autopilot باستخدام Workload Identity وKubernetes Gateway API مع بروتوكول أمان طبقة النقل (TLS).
  • نقاط نهاية خادم MCP المسجَّلة والمكتشَفة باستخدام Gemini Enterprise Agent Platform (قاعدة بيانات الوكلاء) باستخدام عناوين تخويل OIDC

الخطوات التالية