اكتشاف خوادم MCP الديناميكية باستخدام Agent Registry وADK

1. مقدمة

في هذا الدرس التطبيقي حول الترميز، ستتعرّف على كيفية استخدام Google Cloud Agent Registry وحزمة تطوير الوكلاء (ADK) لاكتشاف خوادم بروتوكول Model Context‏ (MCP) المسجّلة واستدعائها ديناميكيًا في وقت التشغيل، ما يلغي الحاجة إلى نقاط النهاية المبرمَجة وإعدادات الأدوات.

يستخدم هذا التمرين ADK ولغة البرمجة Python لإنشاء تطبيق. ويمكن تطبيق المبادئ نفسها في لغات برمجة أخرى وباستخدام أُطر عمل أخرى لتطوير الوكلاء.

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

  • تفعيل Google Cloud Agent Registry وإعداده في مشروعك
  • نشر خادم MCP على Cloud Run وتسجيله في Agent Registry
  • إنشاء وكيل ADK يستدعي خادم MCP باستخدام عنوان URL مضمّن في الرمز البرمجي
  • تغيير الوكيل للعثور على خادم MCP في Agent Registry من خلال هويته
  • تغيير الوكيل لاكتشاف خوادم MCP في وقت التشغيل باستخدام البحث الدلالي

المتطلبات

  • متصفّح ويب حديث (مثل Chrome)
  • مشروع على Google Cloud تم تفعيل الفوترة فيه
  • حساب لديه أذونات لإنشاء الموارد وإدارتها وضبط سياسات "إدارة الهوية وإمكانية الوصول" في مشروع Google Cloud

2. إعداد البيئة

تفترض التعليمات الواردة في هذا الدرس التطبيقي حول الترميز أنّك تشغّل الأوامر في Cloud Shell في Google Cloud Console. ‫Cloud Shell هي بيئة سطر أوامر تعمل في Google Cloud وتكون محمَّلة مسبقًا بالأدوات اللازمة. إذا كان لديك gcloud CLI وgit وPython 3 مثبّتة في بيئتك المحلية، يمكنك أيضًا تشغيل الأوامر هناك.

إنشاء مشروع على Google Cloud

  1. في Google Cloud Console، في صفحة أداة اختيار المشاريع، أنشئ مشروعًا على السحابة الإلكترونية.
  2. تأكَّد من تفعيل الفوترة لمشروعك على السحابة الإلكترونية. كيفية التحقّق من تفعيل الفوترة في مشروع

بدء وحدة طرفية في Cloud Shell

  1. باستخدام أداة اختيار الموارد، اختَر مشروع Google Cloud الذي أنشأته.
  2. انقر على تفعيل Cloud Shell رمز تفعيل Cloud Shell في أعلى "وحدة تحكّم Google Cloud".
  3. بعد الاتصال بـ Cloud Shell، تأكَّد من المصادقة:
    gcloud auth list
  4. تأكَّد من ضبط مشروعك:
    gcloud config get project
  5. إذا لم يتم ضبط مشروعك على النحو المتوقّع أو إذا كنت تستخدم بيئة على جهازك المحلي بدلاً من Cloud Shell، اضبط رقم تعريف المشروع:
    export GOOGLE_CLOUD_PROJECT=<YOUR_PROJECT_ID>
gcloud config set project "${GOOGLE_CLOUD_PROJECT}"

اختيار مشروع على Google Cloud وضبط متغيّرات البيئة

  1. فعِّل واجهات Google Cloud API المطلوبة:
    gcloud services enable \
  agentregistry.googleapis.com \
  aiplatform.googleapis.com \
  cloudbuild.googleapis.com \
  run.googleapis.com \
  --project="${GOOGLE_CLOUD_PROJECT}"

إعداد بيئة Python

  1. أنشِئ بيئة افتراضية وفعِّلها وثبِّت التبعيات المطلوبة:
    mkdir -p adk-dynamic-discovery && cd adk-dynamic-discovery
python3 -m venv .venv
source .venv/bin/activate
pip install "fastmcp<5.0.0" "google-adk[mcp,a2a,agent-identity]>=2.5.0,<3.0.0"

3- إنشاء خادم Zoo MCP وتسجيله

في هذه الخطوة، عليك إنشاء خادم Zoo MCP وتسجيله. يعرض الخادم ثلاث أدوات: list_species لعرض قائمة بجميع الأنواع التي تعيش في حديقة الحيوانات، وget_animals_by_species لعرض قائمة بالحيوانات من نوع معيّن، وget_animal_details للحصول على معلومات إضافية عن حيوان معيّن.

يمكنك العثور على جميع الملفات لهذه الخطوة على GitHub.

إنشاء خادم MCP

  1. إعداد مجلد ومشروع Python لخادم MCP
    mkdir mcp-on-cloudrun && cd mcp-on-cloudrun
uv init --description "Example of deploying an MCP server on Cloud Run" --bare --python 3.13
uv add "fastmcp<5.0.0" --no-sync
  2. نزِّل رمز خادم MCP:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step3/server.py -o server.py

نشر خادم MCP على Cloud Run وتسجيله في Agent Registry

بعد ذلك، عليك إنشاء Dockerfile للخادم ونشره على Cloud Run.

  1. إنشاء Dockerfile:
    cat << EOF > Dockerfile
# Use the official Python image
FROM python:3.13-slim

# Install uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

# Install the project into /app
COPY . /app
WORKDIR /app

# Allow statements and log messages to immediately appear in the logs
ENV PYTHONUNBUFFERED=1

# Install dependencies
RUN uv sync

# Run the FastMCP server
CMD ["uv", "run", "server.py"]
EOF
  2. أنشئ حساب خدمة مخصّصًا للخادم:
    gcloud iam service-accounts create zoo-mcp-server-sa \
    --display-name="Zoo MCP Server Service Account"
  3. انشر الخادم في منطقة us-west1.
    gcloud run deploy zoo-mcp-server \
    --service-account=zoo-mcp-server-sa@$GOOGLE_CLOUD_PROJECT.iam.gserviceaccount.com \
    --region=us-west1 \
    --source=. \
    --labels=dev-tutorial=codelab-dynamic-mcp-discovery \
    --invoker-iam-check \
    --no-allow-unauthenticated
    إذا ظهرت لك الرسالة التالية، أكِّد إنشاء مستودع Artifact Registry جديد. سيظهر هذا الطلب إذا لم يكن لديك مستودع Artifact Registry في مشروعك للمنطقة us-west1:
    Deploying from source requires an Artifact Registry Docker repository to store built containers. A repository named
[cloud-run-source-deploy] in region [us-west1] will be created.

Do you want to continue (Y/n)?
  4. سجِّل خادم MCP في Agent Registry في الموقع الجغرافي "العالمي":
    fastmcp inspect server.py --format mcp | jq 'del(..|nulls) | del(.tools[]?._meta?)' > toolspec.json
MCP_URL=$(gcloud run services describe zoo-mcp-server --region=us-west1 --format='value(status.url)')
gcloud agent-registry services create zoo-mcp-server \
    --display-name="zoo-mcp-server" \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --location=global \
    --mcp-server-spec-type="tool-spec" \
    --mcp-server-spec-content="toolspec.json" \
    --interfaces="url=${MCP_URL}/mcp,protocolBinding=jsonrpc"
    تُنشئ هذه الأوامر toolspec.json باستخدام بيان الأداة وتستدعي واجهة برمجة التطبيقات Agent Registry API لتسجيل الخادم.
  5. تأكَّد من ظهور الوكيل في السجلّ:
    gcloud agent-registry mcp-servers list \
    --location=global \
    --filter="display_name=zoo-mcp-server"
    ستظهر لك نتيجة YAML تتضمّن معلومات خادم MCP على النحو التالي:
    ---
attributes:
  agentregistry.googleapis.com/system/RuntimeReference:
    uri: //agentregistry.googleapis.com/projects/[PROJECT_NUMBER]/locations/global/services/zoo-mcp-server
createTime: '<create time datetime in ISO 8601>'
displayName: zoo-mcp-server
interfaces:
- protocolBinding: JSONRPC
  url: https://zoo-mcp-server-[PROJECT_NUMBER].us-west1.run.app/mcp
mcpServerId: urn:mcp:projects-[PROJECT_NUMBER]:projects:[PROJECT_NUMBER]:locations:global:agentregistry:services:zoo-mcp-server
name: projects/[PROJECT_ID]/locations/global/mcpServers/agentregistry-00000000-0000-0000-1234-56789abcdef
tools:
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: |-
    Returns every species in the zoo. Use it first for zoo-wide questions
    such as counting all animals or listing everything the zoo has.
  name: list_species
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: |-
    Retrieves all animals of a specific species from the zoo.
    Can also be used to collect the base data for aggregate queries
    of animals of a specific species - like counting the number of penguins
    or finding the oldest lion.
  name: get_animals_by_species
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: Retrieves the details of a specific animal by its name.
  name: get_animal_details
updateTime: '<last updated datetime in ISO 8601>'

4. إنشاء وكيل ADK أساسي لاستخدام خادم Zoo MCP

في هذه الخطوة، عليك إنشاء وكيل ADK للإجابة عن طلبات المستخدمين حول حديقة الحيوانات بمساعدة عنوان URL المُبرمَج لخادم Zoo MCP. يتطلّب منك الترميز الثابت معرفة عنوان URL لخادم MCP قبل نشر الوكيل.

يمكنك العثور على الملفات المصدر لهذه الخطوة على GitHub.

إنشاء تطبيق الوكيل

أنشئ هيكل مشروع ADK ووكيل Zoo الذي يستخدم خادم Zoo MCP.

  1. إعداد مجلد ومشروع لوكيل Zoo:
    mkdir ../sample-agent && cd ../sample-agent
uv init --description "Example of ADK agent to showcase dynamic MCP discovery" --bare --python 3.13
uv add "google-adk[mcp,a2a,agent-identity]>=2.5.0,<3.0.0" --no-sync
  2. أنشئ مجلد حزمة الوكيل ونزِّل ملف __init__.py:
    mkdir zoo_agent
curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step4/zoo_agent/__init__.py -o zoo_agent/__init__.py
  3. أنشئ ملف .env واملأه بمتغيّرات البيئة المستخدَمة في هذه الخطوة والخطوات التالية:
    MCP_URL=$(gcloud run services describe zoo-mcp-server --region=us-west1 --format='value(status.url)')
cat << EOF > zoo_agent/.env
GOOGLE_GENAI_USE_ENTERPRISE=TRUE
GOOGLE_CLOUD_LOCATION=global
GOOGLE_CLOUD_PROJECT="${GOOGLE_CLOUD_PROJECT}"
GOOGLE_CLOUD_PROJECT_NUMBER=$(gcloud projects describe ${GOOGLE_CLOUD_PROJECT} \
   --format='value(projectNumber)')
MCP_SERVER_URL="${MCP_URL}"
EOF
    يكتب هذا الأمر جميع متغيّرات البيئة المطلوبة في هذه الخطوة وخطوات الاكتشاف اللاحقة في zoo_agent/.env، لذا لن تحتاج إلى إعادة إنشاء الملف لاحقًا. في هذه الخطوة، يتم استخدام المتغيّرات التالية:
    • ‫GOOGLE_GENAI_USE_ENTERPRISE وGOOGLE_CLOUD_PROJECT وGOOGLE_CLOUD_LOCATION: تُستخدم ضمنيًا من قِبل ADK وحزمة Google GenAI SDK لتوجيه طلبات نموذج Gemini من خلال Vertex AI في مشروعك على السحابة الإلكترونية وموقعك الجغرافي على Google Cloud (global). وتخزّن أيضًا الموقع الجغرافي لـ Agent Registry حيث يتم تسجيل خادم MCP.
    • ‫GOOGLE_CLOUD_PROJECT_NUMBER: لا يُستخدَم في هذه الخطوة، ولكن تتم تعبئته تلقائيًا هنا لاستخدامه في الخطوة التالية.
    • MCP_SERVER_URL: يمكنك قراءة التعليمات مباشرةً في zoo_agent/agent.py لضبط McpToolset باستخدام عنوان URL الثابت لخدمة Cloud Run الخاص بخادم Zoo MCP.
  4. نزِّل agent.py الذي يتضمّن الرمز البرمجي لوكيل Zoo الذي يتصل بخادم Zoo MCP:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step4/zoo_agent/agent.py -o zoo_agent/agent.py

طريقة عمل مصادقة خادم MCP

بما أنّه تم نشر خدمة Cloud Run في zoo-mcp-server باستخدام علامتَي --invoker-iam-check و--no-allow-unauthenticated، يجب أن يتضمّن كل طلب HTTP إلى خادم MCP رمز تعريف موقَّعًا من Google في عنوان Authorization. يجب أن يحدّد الرمز المميّز المتصل الذي لديه إذن run.routes.invoke في الخدمة.

في zoo_agent/agent.py، تحصل فئة _GoogleAuthHeaders على بيانات اعتماد رمز التعريف بإحدى طريقتَين. أولاً، يتم استدعاء fetch_id_token_credentials() للحصول على رمز مميّز لرقم التعريف لعنوان URL الخاص بالخدمة من خادم البيانات الوصفية أو مفتاح حساب الخدمة. وفي حال تعذّر ذلك، يتم استخدام الرمز المميّز لرقم التعريف الخاص ببيانات الاعتماد التلقائية للتطبيق (ADC) للمستخدم. في كل مرة يتم فيها استدعاء المثيل، يتم تحديث الرمز المميّز إذا لزم الأمر ويعرض العنوان Authorization:

class _GoogleAuthHeaders:
    def __init__(self, audience: str):
        self._request = Request()
        self._id_token = False
        try:
            self._credentials = id_token.fetch_id_token_credentials(audience, request=self._request)
        except exceptions.DefaultCredentialsError:
            self._credentials, _ = google.auth.default()
            self._id_token = True
            if not isinstance(self._credentials, credentials.Credentials):
                raise

    def __call__(self, unused_context: ReadonlyContext | None = None) -> dict[str, str]:
        if not self._credentials.valid or (self._id_token and not self._credentials.id_token):
            self._credentials.refresh(self._request)
        token = self._credentials.id_token if self._id_token else self._credentials.token
        return {"Authorization": f"Bearer {token}"}

يمرّر الوكيل نسخة من الفئة إلى McpToolset كوسيطة header_provider. يستدعي ADK موفّر العناوين ويرفق العناوين التي يعرضها بكل طلب إلى خادم MCP. بما أنّ عميل HTTP يتم إنشاؤه بواسطة ADK، يعمل الرمز نفسه بغض النظر عن إصدار حزمة MCP Python SDK ومكتبة HTTP المثبّتة:

zoo_mcp_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url=f"{MCP_SERVER_URL}/mcp",
        timeout=30.0,
    ),
    header_provider=_GoogleAuthHeaders(MCP_SERVER_URL),
)

الجمهور المطلوب هو عنوان URL للخدمة (MCP_SERVER_URL)، بينما يتضمّن عنوان URL للاتصال مسار /mcp حيث يقدّم خادم FastMCP بروتوكول MCP. تتيح مهلة الـ 30 ثانية إمكانية التشغيل على البارد في Cloud Run.

اختبار الوكيل باستخدام ADK Playground في بيئتك المحلية

قبل اختبار الوكيل، تأكَّد من أنّ حساب المستخدم الخاص بك يمكنه الاتصال بخادم MCP.

  1. في الوحدة الطرفية المحلية فقط (تخطَّ هذه الخطوة في Cloud Shell): ابدأ ADC باستخدام بيانات اعتماد المستخدم:
    gcloud auth application-default login
    اتّبِع التعليمات لإكمال عملية إعداد ADC.
  2. (اختياري) امنح حساب المستخدم الإذن باستدعاء خادم MCP البعيد:
    gcloud run services add-iam-policy-binding zoo-mcp-server \
   --region=us-west1 \
   --project=${GOOGLE_CLOUD_PROJECT} \
   --member=user:$(gcloud config get-value account) \
   --role='roles/run.invoker'
  3. شغِّل الوكيل باستخدام الأمر adk run:
    adk run zoo_agent
    أو شغِّل واجهة الويب الخاصة بـ ADK باستخدام الأمر adk web:
    adk web --port 8080
    ثم افتح عنوان URL الذي يطبعه الأمر في المتصفّح.إذا كنت تستخدم Cloud Shell، أضِف الوسيطة --allow_origins للسماح بفتح واجهة الويب من Cloud Shell:
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    بعد تنفيذ الأمر، ابحث عن رمز معاينة الويبرمز &quot;معاينة الويب&quot; في شريط أدوات Cloud Shell (على يسار الصفحة). انقر عليه واختَر معاينة على المنفذ 8080 (Preview on port 8080). سيفتح Cloud Shell علامة تبويب متصفّح جديدة تعرض واجهة الويب الخاصة بـ ADK.
  4. ابدأ الدردشة مع موظّف الدعم. على سبيل المثال، يمكنك أن تسأله عن الحيوانات في حديقة الحيوانات:
    How many animals are in the zoo?
    يستدعي الوكيل list_species ثم get_animals_by_species لكل نوع من أجل حساب الإجابة.

تتيح لك واجهة الويب في ADK استكشاف طريقة عمل الوكيل، بما في ذلك طلباته إلى خادم Zoo MCP. يمكنك الاطّلاع على مستندات واجهة المستخدم على الويب لمعرفة كيفية إجراء ذلك.

5- اكتشاف نقطة نهاية خادم MCP بشكل ديناميكي

في هذه الخطوة، يمكنك تغيير الوكيل لاكتشاف نقطة نهاية خادم Zoo MCP المسجّلة في Agent Registry بدلاً من ترميز عنوان URL لنقطة النهاية بشكل ثابت.

يمكنك العثور على جميع الملفات لهذه الخطوة على GitHub.

يمكنك استخدام دمج ADK مع Agent Registry للعثور على عنوان URL الخاص بالخادم واسترداده. لتنفيذها، تحتاج إلى المعلومات التالية:

  • رقم المشروع وموقع Agent Registry حيث تم تسجيل خادم MCP‏ (global)
  • رقم تعريف الخدمة الذي استخدمته لتسجيل خادم MCP ‏ (zoo-mcp-server).

تعديل الوكيل لاستخدام "سجلّ الوكلاء"

  1. نزِّل ملف agent.py المعدَّل من GitHub لإلغاء الإصدار الأول من الوكيل:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step5/zoo_agent/agent.py -o zoo_agent/agent.py

    متغيّرات البيئة في هذه الخطوة

    من خلال هذا التحديث، يعتمد الوكيل على متغيّرات البيئة التالية من zoo_agent/.env:
    • GOOGLE_CLOUD_PROJECT_NUMBER: أصبح مطلوبًا الآن لإنشاء اسم الموارد المنتظم (URN) الخاص بهوية خادم MCP ‏(MCP_SERVER_URN).
    • ‫GOOGLE_CLOUD_PROJECT وGOOGLE_GENAI_USE_ENTERPRISE وGOOGLE_CLOUD_LOCATION: تستخدمها حزمة ADK لضبط إمكانية الوصول إلى Vertex AI. يُستخدم الموقع الجغرافي أيضًا للعثور على خادم MCP.
    • لم يعُد مطلوبًا: لم يعُد MCP_SERVER_URL مستخدَمًا لأنّ الوكيل يسترد عنوان URL لنقطة نهاية خادم MCP من Agent Registry بدلاً من قراءته من البيئة.

    طريقة عمل هوية خادم MCP وعملية العثور عليه في AgentRegistry

    في Agent Registry، يحصل كل خادم MCP مسجَّل على معرّف فريد مخزَّن في حقل mcpServerId، ويكون بتنسيق اسم الموارد الموحّد (URN). يمكنك الاطّلاع عليه في ناتج الأمر gcloud agent-registry mcp-servers list في الخطوة 3. يعيد الوكيل إنشاء هذا المعرّف من متغيّرات البيئة:
    PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
PROJECT_NUMBER = os.environ["GOOGLE_CLOUD_PROJECT_NUMBER"]
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")
MCP_SERVER_URN = f"urn:mcp:projects-{PROJECT_NUMBER}:projects:{PROJECT_NUMBER}:locations:{LOCATION}:agentregistry:services:zoo-mcp-server"
    يرمّز اسم الموارد الموحّد (URN) رقم مشروع السجلّ (projects-{PROJECT_NUMBER}) ورقم المشروع والموقع الجغرافي الذي تم تسجيل الخادم فيه (projects:{PROJECT_NUMBER}:locations:{LOCATION}) ومعرّف الخدمة الذي استخدمته في الأمر gcloud agent-registry services create ‏(agentregistry:services:zoo-mcp-server).يدرج الوكيل خوادم MCP المسجَّلة باستخدام registry.list_mcp_servers() ويطابق mcpServerId لكل خادم مع MCP_SERVER_URN. بعد ذلك، يمرّر مورد الخادم المطابق name (projects/.../locations/.../mcpServers/...) إلى registry.get_mcp_toolset()، الذي يقرأ عنوان URL لنقطة النهاية من واجهة الخادم المسجّلة (https://...run.app/mcp) ويعرض McpToolset مرتبطًا بعنوان URL هذا:
    header_provider = _GoogleAuthHeaders()
registry = AgentRegistry(
    project_id=PROJECT_ID,
    location=LOCATION,
    header_provider=header_provider,
)

mcp_server_name = next(
    (
        s["name"]
        for s in registry.list_mcp_servers().get("mcpServers", [])
        if s.get("mcpServerId") == MCP_SERVER_URN
    ),
    None,
)
if not mcp_server_name:
    raise RuntimeError(f"MCP server '{MCP_SERVER_URN}' not found in Agent Registry.")
zoo_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)
header_provider.bind(zoo_mcp_toolset.connection_params.url)
zoo_mcp_toolset.connection_params.timeout = 30.0
    يضيف get_mcp_toolset() بادئة إلى أسماء الأدوات باستخدام الاسم المعروض للخادم، وبالتالي يرى الوكيل الأدوات على النحو التالي: zoo_mcp_server_list_species وzoo_mcp_server_get_animals_by_species وzoo_mcp_server_get_animal_details. تمنع البادئة تضارب الأسماء عندما يستخدم الوكيل أدوات من خوادم MCP متعدّدة. يرفع السطر الأخير مهلة الاتصال من القيمة التلقائية التي تبلغ 5 ثوانٍ لإتاحة المجال لعمليات التشغيل على البارد في Cloud Run.

    كيفية تغيُّر التفويض باستخدام AgentRegistry

    يصادق AgentRegistry على طلبات البيانات من واجهة برمجة التطبيقات الخاصة به، مثل list_mcp_servers()، باستخدام ADC، لذا لا يلزم توفّر موفّر الرأس إلا لطلبات البيانات من خادم MCP الذي تم اكتشافه. بما أنّ AgentRegistry ينشئ McpToolset داخليًا، عليك تمرير موفّر العنوان إلى دالة الإنشاء AgentRegistry.في الخطوة السابقة، كان الجمهور هو قيمة MCP_SERVER_URL التي عرفتها مسبقًا. أما الآن، فلا يتم التعرّف على عنوان URL إلا بعد اكتشافه. تستدعي حزمة ADK موفّر الرأس عندما يرسل الوكيل طلبًا إلى خادم MCP، وليس عند إنشاء AgentRegistry. يتيح ذلك للوكيل إنشاء مقدّم الخدمة أولاً وربطه بنقطة النهاية التي تم اكتشافها بعد عرض get_mcp_toolset(). تحصل الطريقة bind() على بيانات الاعتماد بالطريقة نفسها التي تحصل بها الدالة الإنشائية في الخطوة السابقة، مع أخذ الجمهور من عنوان URL لنقطة النهاية التي تم اكتشافها بدون المسار /mcp:
        def bind(self, endpoint_url: str) -> None:
        parsed = urlparse(endpoint_url)
        audience = f"{parsed.scheme}://{parsed.netloc}"
        try:
            self._credentials = id_token.fetch_id_token_credentials(audience, request=self._request)
        except exceptions.DefaultCredentialsError:
            self._credentials, _ = google.auth.default()
            self._id_token = True
            if not isinstance(self._credentials, credentials.Credentials):
                raise
    الطريقة __call__ هي نفسها الطريقة في الخطوة السابقة، باستثناء أنّها تظهر خطأ إذا تم استدعاؤها قبل bind().
  2. شغِّل الوكيل لاختبار التغيير. استخدِم الأمر adk run:
    adk run zoo_agent
    أو شغِّل واجهة الويب الخاصة بـ ADK باستخدام الأمر adk web:
    adk web --port 8080
    ثم افتح عنوان URL الذي يطبعه الأمر في المتصفّح.إذا كنت تستخدم Cloud Shell، أضِف الوسيطة --allow_origins للسماح بفتح واجهة الويب من Cloud Shell:
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    اطرح على الوكيل السؤال نفسه الذي طرحته في الخطوة السابقة. الإجابة هي نفسها، ولكن في واجهة الويب الخاصة بـ ADK، تظهر طلبات الأدوات الآن على النحو التالي: zoo_mcp_server_list_species وzoo_mcp_server_get_animals_by_species.

سمح لك هذا التغيير بتجاوز الحاجة إلى عنوان URL صريح لخادم MCP. ومع ذلك، يجب معرفة رقم المشروع وموقع السجلّ ومعرّف الخدمة المستخدَم لتسجيل خادم MCP، ويتم إجراء عملية الاكتشاف بشكل ثابت أثناء إعداد الوكيل.

6. استخدام البحث الدلالي لخادم MCP في ميزة "الاكتشاف الديناميكي"

في هذه الخطوة، يمكنك تحسين ميزة الاكتشاف الديناميكي لخادم MCP من خلال استبدال المطابقة التامة لاسم الموارد الموحّد (URN) عند بدء التشغيل بالبحث الدلالي أثناء التشغيل. يغني هذا الأسلوب عن الحاجة إلى معرفة رقم المشروع الذي تم نشر الخادم فيه أو هوية الخادم الدقيقة (MCP_SERVER_URN)، ويدمج عملية اكتشاف الأدوات مباشرةً في سير عمل وقت التشغيل العادي للوكيل بدلاً من معالجتها عند الإعداد.

يمكنك العثور على جميع الملفات لهذه الخطوة على GitHub.

تعديل الوكيل لاستخدام ميزة "الاكتشاف الدلالي" أثناء التشغيل

  1. نزِّل ملف agent.py المعدَّل من GitHub لاستبدال عملية التنفيذ الحالية:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step6/zoo_agent/agent.py -o zoo_agent/agent.py

    متغيّرات البيئة في هذه الخطوة

    بما أنّ الوكيل لم يعُد ينشئ اسمًا موحدًا للموارد (URN) لخادم MCP، تم تبسيط الإعدادات المطلوبة بشكل أكبر:
    • ‫GOOGLE_CLOUD_PROJECT وGOOGLE_GENAI_USE_ENTERPRISE وGOOGLE_CLOUD_LOCATION: تستخدمها حزمة ADK لضبط إمكانية الوصول إلى Vertex AI. يُستخدم الموقع الجغرافي أيضًا للعثور على خادم MCP.
    • لم يعُد مطلوبًا: لم يعُد GOOGLE_CLOUD_PROJECT_NUMBER مطلوبًا (بالإضافة إلى MCP_SERVER_URL الذي تمت إزالته في الخطوة السابقة).
    بدلاً من استدعاء registry.list_mcp_servers() أثناء إعداد الوحدة ومقارنة كل إدخال بـ MCP_SERVER_URN الذي تم إنشاؤه، يحدّد agent.py الآن دالة أداة الاستكشاف discover_zoo_mcp_servers التي تستدعي registry.search_mcp_servers() باستخدام search_type="SEMANTIC":
    def discover_zoo_mcp_servers(query: str, tool_context: ToolContext) -> dict[str, Any]:
    """Discovers Zoo MCP server(s) matching the query using semantic search.

    Args:
        query: Semantic description of zoo capabilities needed (e.g. 'zoo animal info')
    """
    cached = _fresh_discovery(tool_context.state)
    if cached:
        return {
            "status": "already_discovered",
            "message": "Zoo MCP servers are already discovered.",
            "servers": cached["servers"],
        }

    try:
        response = registry.search_mcp_servers(
            search_string=query,
            search_type="SEMANTIC",
            page_size=5,
        )
    except Exception as e:
        return {
            "status": "error",
            "message": f"Agent Registry search failed: {e}",
        }

    found_servers = response.get("mcpServers", [])
    if not found_servers:
        return {
            "status": "not_found",
            "message": f"No MCP servers found in Agent Registry at {LOCATION} matching '{query}'.",
        }

    valid_servers = [s["name"] for s in found_servers if "name" in s]
    tool_context.state[DISCOVERY_STATE_KEY] = {
        "servers": valid_servers,
        "discovered_at": time.time(),
    }
    return {
        "status": "discovered",
        "message": f"Successfully discovered {len(valid_servers)} Zoo MCP server(s).",
        "servers": valid_servers,
    }
    يتيح ذلك للوكيل البحث في "سجلّ الوكلاء" (Agent Registry) أثناء التشغيل باستخدام وصف بلغة طبيعية للإمكانات التي يحتاجها (على سبيل المثال، "zoo animals, exhibits, habitat, care, feeding schedule") ويحفظ أسماء موارد الخادم التي تم اكتشافها في حالة جلسة ADK ‏(tool_context.state).

    كيفية دمج ميزة "الاكتشاف الديناميكي" في سير عمل الوكيل

    لتحميل الأدوات من الخوادم التي يتم اكتشافها أثناء التشغيل وتوجيه الوكيل خلال عملية الاكتشاف ثم الاستدعاء، يجمع agent.py بين تنفيذ BaseToolset مخصّص (RegistryDiscoveredToolset) وموفّر تعليمات ديناميكي (zoo_guide_instruction):
    1. مجموعة الأدوات الديناميكية (RegistryDiscoveredToolset): تنشئ فئات فرعية من BaseToolset وتضبط self._use_invocation_cache = False لكي تعيد ADK تقييم الأدوات المتاحة فورًا بعد أن تعدّل discover_zoo_mcp_servers حالة الجلسة بدلاً من تخزين قائمة أدوات فارغة في البداية مؤقتًا. عند استدعاء get_tools()، يقرأ أسماء الخوادم التي تم اكتشافها من الحالة، وينشئ McpToolset مصادقًا لكل خادم في سلسلة الوحدات العاملة (worker thread)، ويعرض جميع الأدوات النشطة. ولا يؤثر تعذُّر الوصول إلى خادم واحد في الخوادم الأخرى. تنفِّذ دالة _build_toolset() عمليات إدخال/إخراج محظورة، لذا يشغّلها _tools_of() في سلسلة الوحدات العاملة.
      class RegistryDiscoveredToolset(BaseToolset):
    def __init__(self, agent_registry: AgentRegistry):
        super().__init__()
        self._registry = agent_registry
        self._toolsets: dict[str, McpToolset] = {}
        self._use_invocation_cache = False

    def _build_toolset(self, server_name: str) -> McpToolset:
        header_provider = _GoogleAuthHeaders()
        server_registry = AgentRegistry(
            project_id=self._registry.project_id,
            location=self._registry.location,
            header_provider=header_provider,
        )
        toolset = server_registry.get_mcp_toolset(server_name)
        header_provider.bind(toolset.connection_params.url)
        toolset.connection_params.timeout = 30.0
        return toolset

    async def _tools_of(self, server_name: str, ctx: ReadonlyContext) -> list[BaseTool]:
        try:
            if server_name not in self._toolsets:
                self._toolsets[server_name] = await asyncio.to_thread(
                    self._build_toolset, server_name
                )
            return await self._toolsets[server_name].get_tools_with_prefix(ctx)
        except Exception as e:
            print(f"Skipping MCP server {server_name}: {e}")
            return []

    async def get_tools(self, readonly_context: ReadonlyContext | None = None) -> list[BaseTool]:
        if not readonly_context:
            return []
        record = _fresh_discovery(readonly_context.state)
        if not record or not record.get("servers"):
            if self._toolsets:
                await self.close()
            return []
        per_server = await asyncio.gather(
            *(self._tools_of(name, readonly_context) for name in record["servers"])
        )
        return [tool for tools in per_server for tool in tools]
    2. تعليمات الوكيل الديناميكية (zoo_guide_instruction): لتقييم حالة الجلسة قبل كل دور للنموذج. في حال عدم العثور على أي خوادم بعد، يتم إلحاق _DISCOVERY_PROMPT لتوجيه الوكيل إلى طلب discover_zoo_mcp_servers أولاً. بعد نجاح عملية الاكتشاف، يتم تعديل الطلب لإدراج خوادم MCP النشطة وتوجيه الوكيل لاستخدام أدوات zoo المثبَّتة مباشرةً بدون تشغيل عملية الاكتشاف مرة أخرى:
      def zoo_guide_instruction(ctx: ReadonlyContext) -> str:
    record = _fresh_discovery(ctx.state)
    if record is None:
        return _ROLE + _DISCOVERY_PROMPT
    return _ROLE + (
        f"\nYour active Zoo MCP servers: {', '.join(record['servers'])}. "
        "Do not run discovery again; use the active zoo tools directly to fulfill the visitor's request."
    )

    تفويض الخوادم التي يتم اكتشافها ديناميكيًا

    يمكن أن يعرض البحث الدلالي عدة خوادم، وتستدعي ADK موفّر عنوان بدون إخباره بالخادم الذي يتم إرسال الطلب إليه. لذلك، ينشئ _build_toolset() _GoogleAuthHeaders وAgentRegistry مخصّصَين لكل خادم يتم اكتشافه ويربط مقدّم الخدمة بنقطة نهاية ذلك الخادم. مقدّم الخدمة هو نفسه في الخطوة السابقة، مع علامة _bound التي تجعله يتعذّر استخدامه قبل bind().

    منطق الاحتفاظ بالبيانات وإعادة تحميلها

    لتجنُّب إرسال طلبات إلى Agent Registry في كل مرة مع الاستمرار في إعادة تحميل نقاط النهاية التي تم اكتشافها بشكل دوري، ينفّذ agent.py عملية الاحتفاظ بالحالة مع مدة بقاء (TTL):
    DISCOVERY_STATE_KEY = "app:zoo_mcp_servers"
DISCOVERY_TTL_SECONDS = 24 * 60 * 60

def _fresh_discovery(state: Any) -> dict[str, Any] | None:
    if not state:
        return None
    record = state.get(DISCOVERY_STATE_KEY)
    if not record or not record.get("servers"):
        return None
    if time.time() - record.get("discovered_at", 0) > DISCOVERY_TTL_SECONDS:
        return None
    return record
    • الاحتفاظ بالبيانات على مستوى التطبيق: يتم تخزين أسماء الخوادم التي تم اكتشافها والطابع الزمني لـ discovered_at في حالة الجلسة ضمن مفتاح app:zoo_mcp_servers. يؤدي استخدام بادئة app: في ADK إلى تحديد نطاق سجلّ الاكتشاف المخزَّن مؤقتًا على مستوى تشغيل التطبيق.
    • التحقّق من صلاحية TTL وإعادة التحميل التلقائي: يتحقّق _fresh_discovery() من توفّر السجلّ المحفوظ وأنّ عمره لا يتجاوز DISCOVERY_TTL_SECONDS (24 ساعة). عندما يكون السجلّ جديدًا، يعيد RegistryDiscoveredToolset استخدام اتصالات McpToolset المخزّنة مؤقتًا (self._toolsets)، ويتجاهل zoo_guide_instruction طلب الاكتشاف، وأي طلب متكرر إلى discover_zoo_mcp_servers يعرض على الفور "already_discovered". بعد انتهاء مدة البقاء التي تبلغ 24 ساعة، يعرض _fresh_discovery() الرمز None، ما يؤدي إلى إغلاق RegistryDiscoveredToolset.get_tools() لجلسات عملاء MCP الحالية (await self.close()) ويطلب zoo_guide_instruction من الوكيل إجراء عملية اكتشاف جديدة.
  2. شغِّل الوكيل لاختبار سير عمل الاكتشاف الدلالي. استخدِم الأمر adk run:
    adk run zoo_agent
    أو شغِّل واجهة الويب الخاصة بـ ADK باستخدام الأمر adk web:
    adk web --port 8080
    ثم افتح عنوان URL الذي يطبعه الأمر في المتصفّح.إذا كنت تستخدم Cloud Shell، أضِف الوسيطة --allow_origins للسماح بفتح واجهة الويب من Cloud Shell:
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    عند طرح سؤال على الوكيل حول حديقة الحيوانات (مثلاً، How many animals are in the zoo?)، ستلاحظ أنّه يستدعي أولاً discover_zoo_mcp_servers، ويحمّل الأدوات ديناميكيًا من خادم MCP الذي تم اكتشافه، ثم يستدعي أدوات حديقة الحيوانات، مثل zoo_mcp_server_list_species وzoo_mcp_server_get_animals_by_species، للإجابة عن سؤالك.اطرح سؤال متابعة، مثلاً Which is the oldest lion?. يجيب الوكيل باستخدام أدوات حديقة الحيوانات التي تم تحميلها مسبقًا ولا يستدعي discover_zoo_mcp_servers مرة أخرى، لأنّ سجلّ الاستكشاف في حالة الجلسة لا يزال حديثًا.

7. تَنظيم

الخيار 1: حذف المشروع

إذا أنشأت مشروعًا مخصّصًا لهذا الدرس التطبيقي حول الترميز، فإنّ أسهل طريقة لإلغاء الفوترة هي حذف المشروع:

gcloud projects delete "${GOOGLE_CLOUD_PROJECT}" --quiet

الخيار 2: حذف موارد فردية

إذا استخدمت مشروعًا حاليًا وأردت الاحتفاظ به، احذف الموارد الفردية التي تم إنشاؤها خلال هذا الدرس التطبيقي حول الترميز:

  1. احذف تسجيل خادم MCP من Agent Registry. بما أنّك سجّلت الخادم يدويًا، لن يؤدي حذف خدمة Cloud Run إلى إزالة التسجيل:
    gcloud agent-registry services delete zoo-mcp-server \
    --location=global \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  2. احذف خدمة خادم MCP في Cloud Run:
    gcloud run services delete zoo-mcp-server \
    --region=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  3. احذف حساب الخدمة الذي تم إنشاؤه لخادم MCP:
    gcloud iam service-accounts delete \
    "zoo-mcp-server-sa@${GOOGLE_CLOUD_PROJECT}.iam.gserviceaccount.com" \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  4. إذا تم إنشاء مستودع cloud-run-source-deploy Artifact Registry خصيصًا خلال هذا الدرس التطبيقي حول الترميز، احذفه لتجنُّب رسوم تخزين نُسخ الحاويات:
    gcloud artifacts repositories delete cloud-run-source-deploy \
    --location=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  5. يمكنك اختياريًا إزالة دليل مشروع على جهاز المستخدم الذي تم إنشاؤه في Cloud Shell:
    cd ~ && rm -rf adk-dynamic-discovery

8. تهانينا

في هذا الدرس التطبيقي حول الترميز، تعرّفت على كيفية اكتشاف خوادم MCP واستخدامها ديناميكيًا مع Google Cloud Agent Registry و"حزمة تطوير الوكلاء" (ADK):

  1. تسجيل خادم MCP: نشر خادم MCP آمن على Cloud Run وتسجيله في Agent Registry
  2. العثور على خادم MCP حسب الهوية: تم حلّ مشكلة نقطة نهاية خادم MCP من Agent Registry بدلاً من ترميز عنوان URL الخاص به بشكل ثابت.
  3. العثور على خوادم MCP بشكل دلالي: يمكن للوكيل العثور على أدوات MCP وتحميلها أثناء التشغيل باستخدام البحث باللغة الطبيعية، مع الاحتفاظ بها وإعادة تحميلها.

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