Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

الوكلاء على نطاق واسع: بنية متعددة الوكلاء باستخدام بروتوكول A2A على Agent Runtime وتكامل ADK

1. مقدمة

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

يحلّ بروتوكول A2A (Agent2Agent) مشكلة التواصل، إذ يوحّد طريقة عثور الوكلاء على إمكانات بعضهم البعض والتعاون في مختلف الأُطر والمؤسسات.
يحلّ وقت تشغيل منصة وكلاء Gemini Enterprise مشكلة النشر، فهو منصة مُدارة بالكامل بدون خادم تستضيف الوكلاء مع توفير إمكانية الربط بين التطبيقات، والتدرّج التلقائي، ونقاط النهاية الآمنة، والجلسات الدائمة، وعدم الحاجة إلى إدارة البنية التحتية.

تتيح لك هذه الأدوات معًا إنشاء وكلاء متخصصين ونشرهم كخدمات A2A قابلة للاكتشاف وتجميعهم في أنظمة متعددة الوكلاء.

ما ستنشئه

وكيل حجز يدير حجوزات الطاولات في المطاعم (إنشاء الحجوزات والتحقّق منها وإلغاؤها) باستخدام حالة جلسة ADK التي تديرها جلسات منصة وكلاء Gemini Enterprise يمكنك نشر هذا الوكيل في وقت تشغيل منصة وكيل Gemini Enterprise حيث يمكن اكتشافه من خلال بطاقة الوكيل في بروتوكول A2A. بعد ذلك، يمكنك ترقية وكيل خدمة المطاعم Foodie Finds (من برنامج التدريب العملي المسبق، لا تقلق إذا لم تكن قد زرت برنامج التدريب العملي، فقد أعددنا لك مستودعًا أوليًا) لاستخدام "وكيل الحجز" كوكيل فرعي عن بُعد من نوع A2A. والنتيجة هي نظام متعدد الوكلاء يوجّه فيه المنسّق طلبات البحث عن القائمة إلى MCP Toolbox وطلبات الحجز إلى وكيل A2A البعيد.

ما ستتعلمه

إنشاء وكيل ADK يستخدم خدمة الجلسات المُدارة لإدارة بيانات الحجوزات
عرض وكيل ADK كخادم A2A باستخدام بطاقات الوكيل ومهاراته
نشر وكيل A2A في بيئة تشغيل وكيل Gemini Enterprise
استخدام وكيل A2A عن بُعد من وكيل ADK آخر باستخدام RemoteA2aAgent والتعامل مع الطلب الذي تمت مصادقته
اختبار الأنظمة المتعددة الوكلاء بشكل تدريجي: A2A محلي، وA2A تم نشره، ودمج جزئي، ونشر كامل

المتطلبات الأساسية

(يُنصح به) أكمِل codelabs التالية :
إنشاء وكلاء ذكاء اصطناعي دائمين باستخدام ADK وCloudSQL -> مزيد من التفاصيل حول جلسة ADK وحالتها
التوليد المعزّز بالاسترجاع (RAG) المستنِد إلى الذكاء الاصطناعي الوكيل باستخدام ADK وMCP Toolbox وCloud SQL: يمكنك مواصلة إنشاء الوكيل من خلال هذا الدرس التطبيقي حول الترميز، فالرمز المبدئي المقدَّم مطابق
حساب على Google Cloud مع حساب فوترة نشط
الإلمام بأساسيات مفاهيم Python وADK

2. إعداد البيئة - مواصلة الدرس التطبيقي السابق حول الترميز

إنّ السرد الذي نوفّره في هذا الدرس التطبيقي حول الترميز هو في الواقع استمرار للدرس التطبيقي حول الترميز الخاص بالمتطلبات الأساسية: التوليد المعزّز بالاسترجاع (RAG) المستنِد إلى الذكاء الاصطناعي الوكيل باستخدام ADK وMCP Toolbox وCloud SQL . يمكنك مواصلة العمل من الدرس التطبيقي حول الترميز السابق

يمكننا البدء في الإنشاء في دليل العمل السابق الخاص بالدرس التطبيقي حول الترميز ( يجب أن يكون دليل العمل build-agent-adk-toolbox-cloudsql). ولتجنُّب حدوث أي التباس، لنغيّر اسم الدليل إلى اسم الدليل نفسه الذي نستخدمه عند البدء من جديد.

mv ~/build-agent-adk-toolbox-cloudsql ~/adk-a2a-agent-runtime-starter
cloudshell workspace ~/adk-a2a-agent-runtime-starter && cd ~/adk-a2a-agent-runtime-starter
source .env

تأكَّد من توفّر ملفات المفاتيح من الدرس العملي السابق:

echo "--- Restaurant Agent ---"
cat restaurant_agent/agent.py | head -5
echo ""
echo "--- Toolbox Config ---"
cat tools.yaml | head -5

من المفترض أن يظهر الملف restaurant_agent/agent.py مع عملية الاستيراد LlmAgent، والملف tools.yaml مع إعدادات "مجموعة الأدوات".

بعد ذلك، لنعيد تهيئة بيئة Python

rm -rf .venv
uv sync

تأكَّد أيضًا من أنّ قاعدة البيانات قد تمّت تعبئتها وهي جاهزة:

uv run python scripts/verify_seed.py

إذا اتّبعت كل تفاصيل الاختبار من الدرس العملي السابق، قد يظهر لك ناتج مشابه لما يلي

Menu Items: 16/15
Embeddings: 16/15

✗ Database not ready

لا بأس. لا يأخذ فحص قاعدة البيانات في الاعتبار البيانات الإضافية التي تُدخلها من فحص نقل البيانات. طالما لديك 15 بيانات أو أكثر، كل شيء على ما يرام.

تفعيل واجهة برمجة التطبيقات المطلوبة

بعد ذلك، علينا التأكّد من تفعيل واجهة برمجة التطبيقات المطلوبة للتفاعل مع "منصة وكيل Gemini Enterprise".

gcloud services enable \
  cloudresourcemanager.googleapis.com

من المفترض أن تتوفّر لديك الملفات والبنية الأساسية اللازمة لمتابعة القسم التالي: A2A Protocol and Gemini Enterprise Agent Runtime !

3- إعداد البيئة - بداية جديدة باستخدام مستودع الرموز البرمجية الأوّلية

تجهّز هذه الخطوة بيئة Cloud Shell وتضبط مشروعك على Google Cloud وتستنسخ مستودع الرموز البرمجية المبدئي.

فتح Cloud Shell

افتح Cloud Shell في المتصفّح. توفّر Cloud Shell بيئة تم ضبطها مسبقًا تتضمّن جميع الأدوات التي تحتاج إليها في هذا الدرس التطبيقي حول الترميز. انقر على تفويض عندما يُطلب منك ذلك

بعد ذلك، انقر على "عرض" -> "وحدة طرفية" لفتح الوحدة الطرفية.يجب أن تبدو واجهتك مشابهة لما يلي

ستكون هذه واجهتنا الرئيسية، مع وضع بيئة التطوير المتكاملة في الأعلى والوحدة الطرفية في الأسفل.

إعداد دليل العمل

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

rm -rf ~/adk-a2a-agent-runtime-starter
git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git
cloudshell workspace ~/adk-a2a-agent-runtime-starter && cd ~/adk-a2a-agent-runtime-starter

أنشئ ملف .env من النموذج المقدَّم:

cp .env.example .env

ملاحظة مهمة: يتم توفير الخطوة أدناه للمساعدة في إعداد مشروع Google Cloud المرتبط بحساب فوترة تجريبي (إلزامي) بسرعة. إذا كنت تفضّل استخدام مشروعك السابق، يمكنك تخطّي الخطوة التالية وتنفيذ الخطوة التالية بدلاً منها:

إضافة اسم مشروعك كمتغيّر GOOGLE_CLOUD_PROJECT في الملف .env
تفعيل مشروعك في نافذة الجهاز باستخدام gcloud config set project your-project-id

بعد ذلك، يمكنك الانتقال مباشرةً إلى قسم تفعيل واجهة برمجة التطبيقات.

إذا لم تكن متأكدًا من ذلك، يمكنك مواصلة قراءة القسم أدناه.

لتبسيط عملية إعداد المشاريع في الوحدة الطرفية، نزِّل نص برمجي لإعداد المشاريع في دليل العمل:

curl -sL https://raw.githubusercontent.com/alphinside/cloud-trial-project-setup/main/setup_verify_trial_project.sh -o setup_verify_trial_project.sh

شغِّل النص البرمجي. يتحقّق من حساب الفوترة التجريبي، وينشئ مشروعًا جديدًا (أو يتحقّق من صحة مشروع حالي)، ويحفظ رقم تعريف مشروعك في ملف .env في الدليل الحالي، ويضبط المشروع النشط في gcloud.

bash setup_verify_trial_project.sh && source .env

سيؤدي النص البرمجي إلى ما يلي:

التأكّد من أنّ لديك حساب فوترة تجريبيًا نشطًا
التحقّق من وجود مشروع حالي في .env (إن وُجد)
إنشاء مشروع جديد أو إعادة استخدام المشروع الحالي
ربط حساب الفوترة التجريبي بمشروعك
احفظ رقم تعريف المشروع في .env
ضبط المشروع كمشروع gcloud نشط

تأكَّد من ضبط المشروع بشكل صحيح من خلال التحقّق من النص الأصفر بجانب دليل العمل في موجّه أوامر Cloud Shell. يجب أن يعرض رقم تعريف مشروعك.

تفعيل واجهة برمجة التطبيقات المطلوبة

بعد ذلك، علينا التأكّد من تفعيل واجهة برمجة التطبيقات المطلوبة للتفاعل مع "منصة وكيل Gemini Enterprise".

gcloud services enable \
  aiplatform.googleapis.com \
  cloudresourcemanager.googleapis.com

إعداد البنية الأساسية في حزمة Starter

أولاً، علينا تثبيت التبعيات في Python باستخدام uv، وهي حزمة Python سريعة وأداة إدارة المشاريع مكتوبة بلغة Rust ( مستندات uv). يستخدم هذا الدرس التطبيقي العملي هذه الحزمة لضمان السرعة والبساطة في الحفاظ على مشروع Python.

uv sync

بعد ذلك، شغِّل نص الإعداد الكامل الذي ينشئ مثيل Cloud SQL، ويضيف البيانات الأولية، وينشر خدمة Toolbox التي ستعمل كحالة أولية لروبوت المطعم.

bash scripts/full_setup.sh > logs/full_setup.log 2>&1 &

4. المفهوم: بروتوكول Agent2Agent (A2A) ووقت تشغيل وكيل Gemini Enterprise

قبل البدء في عملية الإنشاء، لنستغرق بضع لحظات لفهم التكنولوجيتَين الرئيسيتَين المعروضتَين في هذا الدرس التطبيقي حول الترميز لتوسيع نطاق تطبيقنا القائم على الوكلاء.

بروتوكول Agent2Agent (A2A)

بروتوكول Agent2Agent (A2A) هو معيار مفتوح مصمّم لإتاحة التواصل والتعاون السلسَين بين وكلاء الذكاء الاصطناعي. في حين يربط بروتوكول سياق النموذج (MCP) الوكلاء بالأدوات والبيانات، يربط بروتوكول A2A الوكلاء بوكلاء آخرين، ما يتيح لهم التعرّف على إمكانات بعضهم البعض وتفويض المهام والتعاون في مختلف الأُطر والمؤسسات.

يكمن الاختلاف الرئيسي بين تضمين نموذج وكيل كأداة (عبر بروتوكول Model Context) وإتاحته عبر بروتوكول Agent2Agent في أنّ الأدوات لا تحتفظ بأي بيانات وتنفّذ وظائف فردية، بينما يمكن لنماذج وكلاء A2A الاستدلال والاحتفاظ بالبيانات والتعامل مع التفاعلات المتعددة الجولات، مثل التفاوض أو التوضيح. يحتفظ الوكيل الذي يتم عرضه من خلال A2A بكامل إمكاناته بدلاً من أن يتم تقليلها إلى استدعاء دالة.

تحدّد A2A ثلاثة مفاهيم أساسية:

بطاقة الوكيل: مستند JSON يصف دور الوكيل ومهاراته ونقطة النهاية الخاصة به. تستردّ الوكلاء الآخرون هذه البطاقة لاكتشاف الإمكانات.
الرسالة: هي طلب من مستخدم أو وكيل يتم إرساله إلى نقطة نهاية A2A، ما يؤدي إلى بدء مهمة.
المهمة: هي وحدة عمل لها دورة حياة (تم إرسالها ← قيد التنفيذ ← مكتملة/فشلت) وعناصر تحتوي على النتائج.

لمزيد من التفاصيل، يُرجى الاطّلاع على المقالة ما هي ميزة "التطبيق إلى التطبيق"؟

وقت تشغيل منصة وكيل Gemini Enterprise

Agent Runtime هي خدمة مُدارة بالكامل على Google Cloud لتوزيع وكلاء الذكاء الاصطناعي وتوسيع نطاقهم وإدارتهم في مرحلة الإنتاج باستخدام ميزات أمان المؤسسات (مثل عناصر التحكّم في خدمة سحابة VPC وCMEK). وتتولّى هذه الخدمة إدارة البنية التحتية حتى تتمكّن من التركيز على منطق الوكيل.

تقدّم Agent Runtime ما يلي:

النشر المُدار: يمكنك نشر الوكلاء الذين تم إنشاؤهم باستخدام ADK أو LangGraph أو أي إطار عمل Python من خلال طلب واحد من حزمة SDK.
استضافة A2A: نشر الوكلاء كنقاط نهاية متوافقة مع A2A مع عرض بطاقة الوكيل تلقائيًا وإمكانية الوصول المصادق عليه
الجلسات المستمرة: VertexAiSessionService تخزّن سجلّ المحادثات والحالة في جميع الطلبات
التدرّج التلقائي: يتم التدرّج من صفر للتعامل مع عدد الزيارات، بدون إدارة البنية الأساسية
إمكانية الملاحظة: تتبُّع وتسجيل ورصد مدمجة من خلال حزمة إمكانية الملاحظة في Google Cloud
والعديد من الميزات الأخرى، يمكنك الاطّلاع على هذه المستندات للحصول على التفاصيل

في هذا الدرس التطبيقي حول الترميز، يمكنك نشر وكيل الحجز في Agent Runtime. تؤدي عملية النشر إلى تسلسل (تخزين) رمز الوكيل وتحميله. يوفر Agent Runtime نقطة نهاية بلا خادم تعرض بروتوكول A2A، وتتفاعل الوكلاء (أو العملاء) الآخرون معها من خلال طلبات HTTP عادية، تتم المصادقة عليها باستخدام بيانات اعتماد Google Cloud.

5- إنشاء وكيل الحجز

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

إنشاء بنية الوكيل

استخدِم adk create لإنشاء بنية دليل الوكيل مع النموذج الصحيح وإعدادات المشروع:

source .env
uv run adk create reservation_agent \
    --model gemini-2.5-flash \
    --project ${GOOGLE_CLOUD_PROJECT} \
    --region ${GOOGLE_CLOUD_LOCATION}

سيؤدي ذلك إلى إنشاء دليل reservation_agent/ يتضمّن __init__.py وagent.py و.env تم إعدادها مسبقًا لنموذج Gemini على "منصة الوكيل".

adk-a2a-agent-runtime-starter/
├── reservation_agent/
│   ├── __init__.py
│   ├── agent.py
│   └── .env
├── logs
├── scripts
└── ...

بعد ذلك، لنعدّل رمز الوكيل.

كتابة رمز الوكيل

افتح ملف الوكيل الذي تم إنشاؤه:

cloudshell edit reservation_agent/agent.py

بعد ذلك، استبدِل المحتوى بما يلي:

# reservation_agent/agent.py
from google.adk.agents import LlmAgent
from google.adk.tools import ToolContext

# App-scoped state prefix ensures reservations persist across all sessions.
# See https://adk.dev/sessions/state/ for state scope details.
STATE_PREFIX = "app:reservation:"


def create_reservation(
    phone_number: str,
    name: str,
    party_size: int,
    date: str,
    time: str,
    tool_context: ToolContext,
) -> dict:
    """Create a new restaurant reservation.

    Args:
        phone_number: Customer's phone number, used as the reservation ID.
        name: Name for the reservation.
        party_size: Number of guests.
        date: Reservation date (e.g., '2025-07-15' or 'this Friday').
        time: Reservation time (e.g., '7:00 PM').

    Returns:
        Confirmation of the reservation.
    """
    reservation = {
        "name": name,
        "party_size": party_size,
        "date": date,
        "time": time,
        "status": "confirmed",
    }
    tool_context.state[f"{STATE_PREFIX}{phone_number}"] = reservation
    return {
        "status": "confirmed",
        "message": f"Reservation created for {name}, party of {party_size} on {date} at {time}. Phone: {phone_number}.",
    }


def check_reservation(phone_number: str, tool_context: ToolContext) -> dict:
    """Look up an existing reservation by phone number.

    Args:
        phone_number: The phone number used when the reservation was created.
        tool_context: ADK tool context for state access.

    Returns:
        The reservation details, or a message if not found.
    """
    reservation = tool_context.state.get(f"{STATE_PREFIX}{phone_number}")
    if reservation:
        return {"found": True, "reservation": reservation}
    return {"found": False, "message": f"No reservation found for {phone_number}."}


def cancel_reservation(phone_number: str, tool_context: ToolContext) -> dict:
    """Cancel an existing reservation by phone number.

    Args:
        phone_number: The phone number used when the reservation was created.
        tool_context: ADK tool context for state access.

    Returns:
        Confirmation of cancellation, or a message if not found.
    """
    key = f"{STATE_PREFIX}{phone_number}"
    reservation = tool_context.state.get(key)
    if not reservation:
        return {"success": False, "message": f"No reservation found for {phone_number}."}
    if reservation.get("status") == "cancelled":
        return {"success": False, "message": f"Reservation for {phone_number} is already cancelled."}
    reservation["status"] = "cancelled"
    tool_context.state[key] = reservation
    return {"success": True, "message": f"Reservation for {reservation['name']} ({phone_number}) has been cancelled."}


root_agent = LlmAgent(
    name="reservation_agent",
    model="gemini-2.5-flash",
    instruction="""You are a friendly reservation assistant for "Foodie Finds" restaurant.
You help diners create, check, and cancel table reservations.

When a diner wants to make a reservation, collect these details:
- Name for the reservation
- Phone number (used as the reservation ID)
- Party size (number of guests)
- Date
- Time

Always confirm the details before creating the reservation.
When checking or cancelling, ask for the phone number if not provided.
Be concise and professional.""",
    tools=[create_reservation, check_reservation, cancel_reservation],
)

6. إعداد إعدادات خادم A2A

تحديد بطاقة وكيل A2A

بطاقة الوكيل هي وصف منظَّم لإمكانات الوكيل، وتستخدمها الوكلاء والعملاء الآخرون للتعرّف على وظائف الوكيل. أنشئ إعدادات البطاقة:

cloudshell edit reservation_agent/a2a_config.py

انسخ ما يلي والصقه في reservation_agent/a2a_config.py:

# reservation_agent/a2a_config.py
from a2a.types import AgentSkill
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card

reservation_skill = AgentSkill(
    id="manage_reservations",
    name="Restaurant Reservations",
    description="Create, check, and cancel table reservations at Foodie Finds restaurant",
    tags=["reservations", "restaurant", "booking"],
    examples=[
        "Book a table for 4 on Friday at 7pm",
        "Check reservation for 555-0101",
        "Cancel my reservation, phone number 555-0101",
    ],
    input_modes=["text/plain"],
    output_modes=["text/plain"],
)

agent_card = create_agent_card(
    agent_name="Reservation Agent",
    description="Handles restaurant table reservations — create, check, and cancel bookings for Foodie Finds restaurant.",
    skills=[reservation_skill],
)

إنشاء أداة تنفيذ A2A

يربط المنفِّذ بروتوكول A2A ووكيل ADK. يتلقّى الطلبات من خلال بروتوكول A2A، وينفّذها من خلال وكيل ADK، ويعرض النتائج على شكل مهام A2A:

cloudshell edit reservation_agent/executor.py

انسخ ما يلي والصقه في reservation_agent/executor.py:

# reservation_agent/executor.py
import os
from typing import NoReturn

import vertexai
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, UnsupportedOperationError
from a2a.utils import new_agent_text_message
from a2a.utils.errors import ServerError
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService, VertexAiSessionService
from google.genai import types

from reservation_agent.agent import root_agent as reservation_agent


class ReservationAgentExecutor(AgentExecutor):
    """Bridge between the A2A protocol and the ADK reservation agent.

    Uses InMemorySessionService for local testing, VertexAiSessionService
    when deployed to Agent Runtime (detected via GOOGLE_CLOUD_AGENT_ENGINE_ID).
    """

    def __init__(self) -> None:
        self.agent = None
        self.runner = None

    def _init_agent(self) -> None:
        if self.agent is not None:
            return

        self.agent = reservation_agent
        engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")

        if engine_id:
            project = os.environ.get("GOOGLE_CLOUD_PROJECT")
            location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
            vertexai.init(project=project, location=location)
            session_service = VertexAiSessionService(
                project=project, location=location, agent_engine_id=engine_id,
            )
            app_name = engine_id
        else:
            session_service = InMemorySessionService()
            app_name = self.agent.name

        self.runner = Runner(
            app_name=app_name,
            agent=self.agent,
            artifact_service=InMemoryArtifactService(),
            session_service=session_service,
            memory_service=InMemoryMemoryService(),
        )

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        if self.agent is None:
            self._init_agent()

        query = context.get_user_input()
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        user_id = context.message.metadata.get("user_id", "a2a-user") if context.message.metadata else "a2a-user"

        if not context.current_task:
            await updater.submit()
        await updater.start_work()

        try:
            session = await self._get_or_create_session(context.context_id, user_id)
            content = types.Content(role="user", parts=[types.Part(text=query)])

            async for event in self.runner.run_async(
                session_id=session.id, user_id=user_id, new_message=content,
            ):
                if event.is_final_response():
                    parts = event.content.parts
                    answer = " ".join(p.text for p in parts if p.text) or "No response."
                    await updater.add_artifact([TextPart(text=answer)], name="answer")
                    await updater.complete()
                    break
        except Exception as e:
            await updater.update_status(
                TaskState.failed, message=new_agent_text_message(f"Error: {e!s}"),
            )
            raise

    async def _get_or_create_session(self, context_id: str, user_id: str):
        app_name = self.runner.app_name
        if context_id:
            session = await self.runner.session_service.get_session(
                app_name=app_name, session_id=context_id, user_id=user_id,
            )
            if session:
                return session
        session = await self.runner.session_service.create_session(
            app_name=app_name, user_id=user_id, session_id=context_id,
        )
        return session

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> NoReturn:
        raise ServerError(error=UnsupportedOperationError())

يرصد المنفّذ بيئته تلقائيًا: عند ضبط GOOGLE_CLOUD_AGENT_ENGINE_ID (يُدرج Agent Runtime هذا المتغير في وقت النشر)، يستخدم VertexAiSessionService للجلسات المستمرة. على المستوى المحلي، يتم الرجوع إلى InMemorySessionService.

يجب أن يحتوي دليل reservation_agent الآن على ما يلي:

reservation_agent/
├── __init__.py
├── agent.py
├── a2a_config.py
├── executor.py
└── .env

7. إعداد وكيل A2A باستخدام Agent Platform SDK واختباره محليًا

تتضمّن هذه الخطوة تغليف وكيل الحجز كوكيل متوافق مع A2A باستخدام فئة A2aAgent في حزمة تطوير البرامج (SDK) لمنصّة الوكيل (لا يزال اسم حزمة تطوير البرامج (SDK) يستخدم المصطلح vertex لتحقيق التوافق مع الإصدارات القديمة)، ثم اختبار مسار بروتوكول A2A الكامل محليًا، أي استرداد بطاقة الوكيل وإرسال الرسائل واسترداد المهام. هذا هو كائن A2aAgent نفسه الذي ستنشره في Agent Runtime في الخطوة التالية.

إضافة عناصر تابعة

ثبِّت حزمة تطوير البرامج (SDK) لمنصّة Agent Platform مع توفير Agent Runtime وADK، بالإضافة إلى حزمة تطوير البرامج (SDK) الخاصة ببروتوكول A2A:

uv add "google-cloud-aiplatform[agent_engines,adk]==1.149.0" "a2a-sdk==0.3.26"

التعرّف على مكوّنات A2A

يتطلّب تغليف وكيل ADK لتطبيق A2A ثلاثة مكوّنات:

بطاقة الوكيل: هي "بطاقة عمل" تصف إمكانات الوكيل ومهاراته وعنوان URL لنقطة النهاية. تستخدم الوكلاء الآخرون هذا الوصف لمعرفة ما يفعله وكيلك.
Agent Executor: هو الجسر بين بروتوكول A2A ومنطق وكيل ADK. يتلقّى الطلبات من A2A، وينفّذها من خلال وكيل ADK، ويعرض النتائج كمهام A2A.
‫A2aAgent: فئة حزمة تطوير البرامج (SDK) الخاصة بمنصة Agent التي تجمع بين البطاقة والمنفّذ في وحدة قابلة للنشر.

إنشاء النص البرمجي للاختبار

أنشئ النص البرمجي التالي للاختبار محليًا

cloudshell edit scripts/test_a2a_agent_local.py

انسخ ما يلي والصقه في scripts/test_a2a_agent_local.py:

# scripts/test_a2a_agent_local.py
import asyncio
import json
import os
from pprint import pprint

from dotenv import load_dotenv
from starlette.requests import Request
from vertexai.preview.reasoning_engines import A2aAgent

from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor

load_dotenv()


# --- Helper functions for building mock requests ---

def receive_wrapper(data: dict):
    async def receive():
        byte_data = json.dumps(data).encode("utf-8")
        return {"type": "http.request", "body": byte_data, "more_body": False}
    return receive

def build_post_request(data: dict = None, path_params: dict = None) -> Request:
    scope = {"type": "http", "http_version": "1.1", "headers": [(b"content-type", b"application/json")], "app": None}
    if path_params:
        scope["path_params"] = path_params
    return Request(scope, receive_wrapper(data))

def build_get_request(path_params: dict) -> Request:
    scope = {"type": "http", "http_version": "1.1", "query_string": b"", "app": None}
    if path_params:
        scope["path_params"] = path_params
    async def receive():
        return {"type": "http.disconnect"}
    return Request(scope, receive)


# --- Helper: poll for task completion ---

async def wait_for_task(a2a_agent, task_id, max_retries=30):
    """Poll on_get_task until the task reaches a terminal state."""
    for _ in range(max_retries):
        request = build_get_request({"id": task_id})
        result = await a2a_agent.on_get_task(request=request, context=None)
        state = result.get("status", {}).get("state", "")
        if state in ["completed", "failed"]:
            return result
        await asyncio.sleep(1)
    return result


def print_task_answer(result):
    """Extract and print the answer from task artifacts."""
    print(f"Status: {result.get('status', {}).get('state')}")
    for artifact in result.get("artifacts", []):
        if artifact.get("parts") and "text" in artifact["parts"][0]:
            print(f"Answer: {artifact['parts'][0]['text']}")


# --- Local test ---

async def main():
    # Create and set up the A2A agent locally
    a2a_agent = A2aAgent(agent_card=agent_card, agent_executor_builder=ReservationAgentExecutor)
    a2a_agent.set_up()

    # 1. Get agent card
    print("=" * 50)
    print("1. Retrieving agent card...")
    print("=" * 50)
    request = build_get_request(None)
    card_response = await a2a_agent.handle_authenticated_agent_card(request=request, context=None)
    print(f"Agent: {card_response.get('name')}")
    print(f"Skills: {[s.get('name') for s in card_response.get('skills', [])]}")

    # 2. Create a reservation
    print("\n" + "=" * 50)
    print("2. Creating a reservation...")
    print("=" * 50)
    message_data = {
        "message": {
            "messageId": f"msg-{os.urandom(4).hex()}",
            "content": [{"text": "Book a table for 2 on Saturday at 6pm. Name: Bob, Phone: 555-0202"}],
            "role": "ROLE_USER",
        },
    }
    request = build_post_request(message_data)
    response = await a2a_agent.on_message_send(request=request, context=None)
    task_id = response["task"]["id"]
    context_id = response["task"].get("contextId")
    print(f"Task ID: {task_id}")

    # 3. Wait for result
    print("\n" + "=" * 50)
    print("3. Waiting for task result...")
    print("=" * 50)
    result = await wait_for_task(a2a_agent, task_id)
    print_task_answer(result)

    # 4. Check the reservation (same context for session continuity)
    print("\n" + "=" * 50)
    print("4. Checking the reservation...")
    print("=" * 50)
    check_data = {
        "message": {
            "messageId": f"msg-{os.urandom(4).hex()}",
            "content": [{"text": "Check the reservation for 555-0202"}],
            "role": "ROLE_USER",
            "contextId": context_id,
        },
    }
    request = build_post_request(check_data)
    check_response = await a2a_agent.on_message_send(request=request, context=None)
    check_result = await wait_for_task(a2a_agent, check_response["task"]["id"])
    print_task_answer(check_result)

    # 5. Cancel the reservation
    print("\n" + "=" * 50)
    print("5. Cancelling the reservation...")
    print("=" * 50)
    cancel_data = {
        "message": {
            "messageId": f"msg-{os.urandom(4).hex()}",
            "content": [{"text": "Cancel the reservation for 555-0202"}],
            "role": "ROLE_USER",
            "contextId": context_id,
        },
    }
    request = build_post_request(cancel_data)
    cancel_response = await a2a_agent.on_message_send(request=request, context=None)
    cancel_result = await wait_for_task(a2a_agent, cancel_response["task"]["id"])
    print_task_answer(cancel_result)

    print("\n" + "=" * 50)
    print("All tests passed!")
    print("=" * 50)


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

يستورد النص البرمجي للاختبار بطاقة الوكيل والمنفّذ اللذين أنشأتهما في الخطوة السابقة، بدون تكرار. سيتم إنشاء A2aAgent محلي، ومحاكاة طلبات بروتوكول A2A من خلال طلبات HTTP وهمية، والتحقّق من جميع عمليات الحجز الثلاث.

بما أنّه لم يتم ضبط GOOGLE_CLOUD_AGENT_ENGINE_ID محليًا، يستخدم المنفّذ InMemorySessionService. عند النشر في Agent Runtime، يتم التبديل تلقائيًا إلى VertexAiSessionService لتوفير جلسات مستمرة.

إجراء الاختبار

PYTHONPATH=. uv run python scripts/test_a2a_agent_local.py

يتضمّن الناتج خمس مراحل:

بطاقة الوكيل: تسترد هذه البطاقة إمكانات الوكيل ومهاراته.
إنشاء حجز: لحجز طاولة وإرجاع مهمة تتضمّن التأكيد
Get task result: لاسترداد المهمة المكتملة مع الإجابة
التحقّق من الحجز: للبحث عن الحجز باستخدام رقم الهاتف
إلغاء الحجز: لإلغاء الحجز وتأكيده

مثال على الناتج كما هو موضّح أدناه

==================================================
1. Retrieving agent card...
==================================================
Agent: Reservation Agent
Skills: ['Restaurant Reservations']

==================================================
2. Creating a reservation...
==================================================
Task ID: f7f7004d-cfea-49c2-b57d-5bca9959e193

==================================================
3. Waiting for task result...
==================================================
Status: TASK_STATE_COMPLETED
Answer: Your reservation for Bob, party of 2, on Saturday at 6:00 PM has been confirmed. The phone number associated is 555-0202.

==================================================
4. Checking the reservation...
==================================================
Status: TASK_STATE_COMPLETED
Answer: I found a reservation for Bob, party of 2, on Saturday at 6:00 PM. The reservation status is confirmed.

==================================================
5. Cancelling the reservation...
==================================================
Status: TASK_STATE_COMPLETED
Answer: Your reservation for Bob (555-0202) has been cancelled.

==================================================
All tests passed!
==================================================

في هذه المرحلة، تكون قد تحقّقت من أنّ بطاقة وكيل A2A تصف المهارات الصحيحة، وأنّ جميع عمليات الحجز الثلاث تعمل من خلال تدفق الرسائل/المهام في بروتوكول A2A، وأنّ الحالة تظل محفوظة في جميع الرسائل ضمن السياق نفسه.

8. نشر "وكيل الحجز" في "بيئة تشغيل الوكيل"

تؤدي هذه الخطوة إلى نشر وكيل الحجز في بيئة تشغيل "منصة وكيل Gemini Enterprise"، وهي منصة بلا خادم مُدارة بالكامل تستضيف الوكيل وتعرضه كنقطة نهاية آمنة من تطبيق إلى تطبيق. بعد النشر، يمكن لأي عميل معتمَد اكتشاف الوكيل والتفاعل معه من خلال نقاط نهاية HTTP العادية من تطبيق إلى تطبيق.

إنشاء حزمة مرحلية

أنشئ حزمة Cloud Storage لتنظيم Agent Runtime. يستخدم Agent Runtime هذا الحزمة لتحميل رمز الوكيل وموارد الاعتمادية أثناء عملية النشر:

STAGING_BUCKET="${GOOGLE_CLOUD_PROJECT}-adk-a2a-agent-runtime"
gsutil mb -l $REGION -p $GOOGLE_CLOUD_PROJECT gs://$STAGING_BUCKET 2>/dev/null || echo "Bucket already exists"
echo "STAGING_BUCKET=$STAGING_BUCKET" >> .env
source .env

إنشاء نص برمجي للنشر

بعد ذلك، سنحتاج إلى إعداد نص البرمجة الخاص بالنشر

cloudshell edit scripts/deploy_a2a_agent_runtime.py

انسخ ما يلي والصقه في scripts/deploy_a2a_agent_runtime.py:

# scripts/deploy_a2a_agent_runtime.py
import os
from pathlib import Path

import vertexai
from dotenv import load_dotenv
from google.genai import types
from vertexai.preview.reasoning_engines import A2aAgent

from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor

load_dotenv()

PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
STAGING_BUCKET = os.environ.get("STAGING_BUCKET", f"{PROJECT_ID}-adk-a2a-agent-runtime")
BUCKET_URI = f"gs://{STAGING_BUCKET}"

a2a_agent = A2aAgent(
    agent_card=agent_card,
    agent_executor_builder=ReservationAgentExecutor,
)


def main():
    vertexai.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)
    client = vertexai.Client(
        project=PROJECT_ID,
        location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )

    print("Deploying Reservation Agent to Agent Runtime...")
    print("This may take 3-5 minutes.")

    remote_agent = client.agent_engines.create(
        agent=a2a_agent,
        config={
            "display_name": agent_card.name,
            "description": agent_card.description,
            "requirements": [
                "google-cloud-aiplatform[agent_engines,adk]==1.149.0",
                "a2a-sdk==0.3.26",
                "google-adk==1.29.0",
                "cloudpickle",
                "pydantic"
            ],
            "extra_packages": [
                "./reservation_agent",
            ],
            "http_options": {
                "api_version": "v1beta1",
            },
            "staging_bucket": BUCKET_URI,
        },
    )

    resource_name = remote_agent.api_resource.name
    print(f"\nDeployment complete!")
    print(f"Resource name: {resource_name}")

    env_path = Path(".env")
    lines = env_path.read_text().splitlines() if env_path.exists() else []
    lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_RESOURCE_NAME=")]
    lines.append(f"RESERVATION_AGENT_RESOURCE_NAME={resource_name}")
    env_path.write_text("\n".join(lines) + "\n")
    print("Written RESERVATION_AGENT_RESOURCE_NAME to .env")


if __name__ == "__main__":
    main()

يستورد النص البرمجي للنشر agent_card وReservationAgentExecutor نفسيهما المستخدَمان في الاختبار المحلي، ما يمنع تكرار الرموز. يُسلسل Agent Runtime الكائن A2aAgent (يُخزّنه) مع التبعيات لنشره. في نهاية نص البرمجة الخاص بالنشر، سيتم كتابة قيمة RESERVATION_AGENT_RESOURCE_NAME في الملف .env

النشر في Agent Runtime

شغِّل نص النشر البرمجي:

PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py

تستغرق عملية النشر من 3 إلى 5 دقائق. يوفّر النص البرمجي نقطة نهاية بلا خادم على Agent Runtime تستضيف وكيل الحجز. بعد عملية نشر ناجحة، ستظهر لك نتيجة مشابهة لما هو موضّح أدناه

Deploying Reservation Agent to Agent Runtime...
This may take 3-5 minutes.

Deployment complete!
Resource name: projects/your-project-number/locations/us-central1/reasoningEngines/your-agent-deployment-unique-id
Written RESERVATION_AGENT_RESOURCE_NAME to .env

يمكنك عرض الوكيل الذي تم نشره في وحدة تحكّم السحابة الإلكترونية. ابحث عن Agent Platform في شريط البحث في وحدة التحكّم

بعد ذلك، مرِّر مؤشر الماوس فوق Agents في علامة التبويب اليمنى، ثم انقر على Deployments.

سيظهر Reservation Agent في قائمة النشر كما هو موضّح أدناه

اختبار الوكيل الذي تم نشره

الآن، نحن مستعدون لاختبار الوكيل الذي تم نشره، وإنشاء نص برمجي للاختبار:

cloudshell edit scripts/test_a2a_agent_runtime.py

انسخ ما يلي والصقه في scripts/test_a2a_agent_runtime.py:

# scripts/test_a2a_agent_runtime.py
import asyncio
import os
import time

import vertexai
from a2a.types import TaskState
from dotenv import load_dotenv
from google.genai import types

load_dotenv()

PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]


async def main():
    vertexai.init(project=PROJECT_ID, location=REGION)
    client = vertexai.Client(
        project=PROJECT_ID, location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )

    agent = client.agent_engines.get(name=RESOURCE_NAME)

    # 1. Get agent card
    print("=" * 50)
    print("1. Retrieving agent card...")
    print("=" * 50)
    card = await agent.handle_authenticated_agent_card()
    print(f"Agent: {card.name}")
    print(f"URL: {card.url}")
    print(f"Skills: {[s.name for s in card.skills]}")

    # 2. Send a reservation request
    print("\n" + "=" * 50)
    print("2. Sending reservation request...")
    print("=" * 50)
    message_data = {
        "messageId": "msg-remote-001",
        "role": "user",
        "parts": [{"kind": "text", "text": "Book a table for 3 on Sunday at noon. Name: Carol, Phone: 555-0303"}],
    }
    response = await agent.on_message_send(**message_data)

    task_object = None
    for chunk in response:
        if isinstance(chunk, tuple) and len(chunk) > 0 and hasattr(chunk[0], "id"):
            task_object = chunk[0]
            break

    task_id = task_object.id
    print(f"Task ID: {task_id}")
    print(f"Status: {task_object.status.state}")

    # 3. Poll for result
    print("\n" + "=" * 50)
    print("3. Waiting for result...")
    print("=" * 50)
    result = None
    for _ in range(30):
        try:
            result = await agent.on_get_task(id=task_id)
            if result.status.state in [TaskState.completed, TaskState.failed]:
                break
        except Exception:
            pass
        time.sleep(1)

    print(f"Final status: {result.status.state}")
    if result.artifacts:
        for artifact in result.artifacts:
            if artifact.parts and hasattr(artifact.parts[0], "root") and hasattr(artifact.parts[0].root, "text"):
                print(f"Answer: {artifact.parts[0].root.text}")

    print("\n" + "=" * 50)
    print("Remote agent test passed!")
    print("=" * 50)


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

بعد ذلك، لننفّذ الاختبار

source .env
uv run python scripts/test_a2a_agent_runtime.py

تعرض النتيجة بطاقة الوكيل مع مهارة "حجوزات المطاعم"، يليها إكمال المهمة بتأكيد الحجز.

==================================================
1. Retrieving agent card...
==================================================
Agent: Reservation Agent
URL: https://us-central1-aiplatform.googleapis.com/v1beta1/projects/your-project-id/locations/us-central1/reasoningEngines/your-agent-unique-id/a2a
Skills: ['Restaurant Reservations']

==================================================
2. Sending reservation request...
==================================================
Task ID: b34585d0-5f03-4cb0-85a3-40710a0d224d
Status: TaskState.completed

==================================================
3. Waiting for result...
==================================================
Final status: TaskState.completed
Answer: Your reservation for Carol, party of 3 on Sunday at noon with phone number 555-0303 is confirmed.

==================================================
Remote agent test passed!
==================================================

يعمل وكيل الحجز الآن بنجاح كنقطة نهاية A2A مُدارة على Agent Runtime.

9- دمج "وكيل الحجز" المستند إلى A2A مع "وكيل المطعم" الجذر

تعمل هذه الخطوة على ترقية وكيل المطعم لاستخدام وكيل الحجز الذي تم نشره كوكيل فرعي عن بُعد من النوع A2A. يتم تشغيل أداة التنسيق محليًا، بينما يتم تشغيل وكيل الحجز على Agent Runtime، وهو تكامل جزئي يتحقّق من اتصال A2A قبل النشر الكامل.

حلّ عنوان URL لبطاقة وكيل A2A

يحتاج RemoteA2aAgent إلى عنوان URL لبطاقة وكيل الحجز الذي تم نشره من أجل التعرّف على إمكاناته. أنشئ نصًا برمجيًا يستردّ عنوان URL هذا من Agent Runtime ويكتبه في .env الخاص بوكيل المطعم:

cloudshell edit scripts/resolve_agent_card_url.py

انسخ ما يلي والصقه في scripts/resolve_agent_card_url.py:

# scripts/resolve_agent_card_url.py
import asyncio
import os
from pathlib import Path

import vertexai
from dotenv import load_dotenv
from google.genai import types

load_dotenv()

PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]


async def main():
    vertexai.init(project=PROJECT_ID, location=REGION)
    client = vertexai.Client(
        project=PROJECT_ID, location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )

    agent = client.agent_engines.get(name=RESOURCE_NAME)
    card = await agent.handle_authenticated_agent_card()
    card_url = f"{card.url}/v1/card"

    print(f"Agent: {card.name}")
    print(f"Card URL: {card_url}")

    # Write to restaurant_agent/.env
    # Write to both restaurant_agent/.env (for adk web) and root .env (for Cloud Run deploy)
    for env_path in [Path("restaurant_agent/.env"), Path(".env")]:
        lines = env_path.read_text().splitlines() if env_path.exists() else []
        lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_CARD_URL=")]
        lines.append(f"RESERVATION_AGENT_CARD_URL={card_url}")
        env_path.write_text("\n".join(lines) + "\n")
        print(f"Written RESERVATION_AGENT_CARD_URL to {env_path}")


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

تشغيل النص البرمجي لتعبئة ملف .env بعنوان URL الخاص ببطاقة الوكيل

uv run python scripts/resolve_agent_card_url.py
source .env

تعديل وكيل المطعم

افتح ملف وكيل المطعم:

cloudshell edit restaurant_agent/agent.py

بعد ذلك، استبدِل المحتوى بالإصدار المعدَّل الذي يتضمّن وكيل الحجز عن بُعد كوكيل فرعي:

# restaurant_agent/agent.py
import os

import httpx
from google.adk.agents import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.auth import default
from google.auth.transport.requests import Request as AuthRequest
from toolbox_adk import ToolboxToolset

TOOLBOX_URL = os.environ.get("TOOLBOX_URL", "http://127.0.0.1:5000")
RESERVATION_AGENT_CARD_URL = os.environ.get("RESERVATION_AGENT_CARD_URL", "")

toolbox = ToolboxToolset(TOOLBOX_URL)


class GoogleCloudAuth(httpx.Auth):
    """Auto-refreshing Google Cloud authentication for httpx.

    Refreshes the access token before each request if expired,
    so long-running agents never hit 401 errors.
    """

    def __init__(self):
        self.credentials, _ = default(
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )

    def auth_flow(self, request):
        # Refresh the token if it is expired or missing
        if not self.credentials.valid:
            self.credentials.refresh(AuthRequest())
            
        request.headers["Authorization"] = f"Bearer {self.credentials.token}"
        yield request


reservation_remote_agent = RemoteA2aAgent(
    name="reservation_agent",
    description="Handles restaurant table reservations — create, check, and cancel bookings. Delegate to this agent when the user wants to book a table, check a reservation, or cancel a reservation.",
    agent_card=RESERVATION_AGENT_CARD_URL,
    httpx_client=httpx.AsyncClient(auth=GoogleCloudAuth(), timeout=60),
)

root_agent = LlmAgent(
    name="restaurant_agent",
    model="gemini-2.5-flash",
    instruction="""You are a friendly and knowledgeable concierge at "Foodie Finds," a restaurant. Your job:
- Help diners browse the menu by category or cuisine type.
- Provide full details about specific dishes, including ingredients, price, and dietary information.
- Recommend dishes based on natural language descriptions of what the diner is craving.
- Add new menu items when asked.
- For reservation requests (booking, checking, or cancelling tables), delegate to the reservation_agent.

When a diner asks about a specific dish by name or cuisine, use the get-item-details tool.
When a diner asks for a specific category or cuisine type, use the search-menu tool.
When a diner describes what kind of food they want — by flavor, texture, dietary needs, or cravings — use the search-menu-by-description tool for semantic search.

When in doubt between search-menu and search-menu-by-description, prefer search-menu-by-description — it searches dish descriptions and finds more relevant matches.
If a dish is not available (available is false), let the diner know and suggest similar alternatives from the search results.
Be conversational, knowledgeable, and concise.""",
    tools=[toolbox],
    sub_agents=[reservation_remote_agent],
)

في ما يلي التغييرات الرئيسية مقارنةً بالإصدار السابق:

GoogleCloudAuth: معالج httpx.Auth مخصّص يعيد تحميل رمز الدخول إلى Google Cloud قبل كل طلب. يتطلّب Agent Runtime إجراء مكالمات A2A مصادَق عليها، وتنتهي صلاحية الرموز المميزة بعد فترة زمنية معيّنة.
يقرأ RemoteA2aAgent RESERVATION_AGENT_CARD_URL من .env (الذي كتبه النص البرمجي لحلّ المشكلة) ويستخدم httpx_client الذي تمت مصادقته
مسجَّل كوكيل فرعي: يفوّض منسّق "حزمة تطوير التطبيقات" طلبات الحجز إليه تلقائيًا
تعديل التعليمات لتتضمّن الإشارة إلى تفويض الحجز

اختبار الوكيل المدمج محليًا

كان الوكيل الأوّلي يتطلّب الدمج مع MCP Toolbox، وكان من المفترض أن يتم توفير الملف المطلوب من درس تطبيقي حول الترميز السابق أو من مستودع الرموز الأوّلي. ما علينا سوى التأكّد من أنّ عملية صندوق الأدوات تعمل بشكلٍ سليم.

إذا كان TOOLBOX_URL في .env يشير إلى خدمة Cloud Run (من الدرس التطبيقي حول الترميز السابق أو ربما من full_setup.sh في مستودع نقطة البدء)، يمكنك تخطّي هذه الخطوة، وسيتصل الوكيل بـ Toolbox الذي تم تفعيله.

إذا كنت بحاجة إلى أداة Toolbox محلية بدلاً من ذلك، تحقَّق مما إذا كانت إحدى الأدوات تعمل حاليًا قبل بدء مثيل جديد:

if curl -s http://127.0.0.1:5000/api/toolsets > /dev/null 2>&1; then
  echo "Toolbox already running on port 5000"
else
  set -a; source .env; set +a
  ./toolbox --config=tools.yaml > logs/toolbox.log 2>&1 &
  echo "Toolbox started"
fi

بعد ذلك، يمكننا محاولة التفاعل مع وكيل المطعم من خلال واجهة مستخدم الويب الخاصة بـ "حزمة تطوير التطبيقات"

uv run adk web --allow_origins "regex:https://.*\.cloudshell\.dev" --port 8080

افتح واجهة مستخدم الويب الخاصة بـ ADK باستخدام "معاينة الويب" في Cloud Shell (انقر على زر "معاينة الويب"، وغيِّر المنفذ إلى 8080)، ثم انقر على restaurant_agent

اختبار محادثة مختلطة:

طلب بحث في القائمة

What Italian dishes do you have?

طلب الحجز

I want to create reservation under name Bob, phone number 123456

التحقّق من الحجز

إنشاء جلسة جديدة ( بدء محادثة جديدة):

Check the reservation for 123456

أوقِف عملية adk web من خلال الضغط على Ctrl+C مرّتين. بعد ذلك، لنُكمل النظام من خلال نشر الوكيل بالكامل

10. نشر وكيل المطعم المعدَّل على Cloud Run

تعيد هذه الخطوة نشر وكيل المطعم على Cloud Run مع دمج A2A، ما يتيح إكمال نشر النظام المتعدد الوكلاء بالكامل.

منح أذونات الوصول إلى Agent Runtime

يجب أن يمتلك حساب خدمة Cloud Run الإذن باستدعاء Agent Runtime. امنح الدور roles/aiplatform.user لحساب الخدمة التلقائي في Compute Engine:

PROJECT_NUMBER=$(gcloud projects describe $GOOGLE_CLOUD_PROJECT --format='value(projectNumber)')
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
  --member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
  --role="roles/aiplatform.user"

النشر على Cloud Run

في عملية الإعداد هذه، نفترض أنّ خدمة وكيل المطعم متوفّرة من الدرس العملي السابق أو من خلال تنفيذ scripts/full_setup.sh إذا بدأت من جديد. يتم إعادة نشر هذا الرمز مع الرمز المعدَّل (عملية دمج RemoteA2aAgent الجديدة) وإضافة عنوان URL لبطاقة وكيل الحجز كمتغيّر بيئة جديد، مع الحفاظ على متغيّرات البيئة الحالية (TOOLBOX_URL وGOOGLE_CLOUD_PROJECT وما إلى ذلك):

gcloud run deploy restaurant-agent \
  --source . \
  --region=$REGION \
  --allow-unauthenticated \
  --update-env-vars="RESERVATION_AGENT_CARD_URL=$RESERVATION_AGENT_CARD_URL" \
  --min-instances=0 \
  --max-instances=1 \
  --memory=1Gi \
  --port=8080

اختبار النظام الذي تم نشره بالكامل

احصل على عنوان URL للخدمة التي تم نشرها:

AGENT_URL=$(gcloud run services describe restaurant-agent --region=$REGION --format='value(status.url)')
echo "Agent URL: $AGENT_URL"

افتح عنوان URL في المتصفّح. يتم تحميل واجهة مستخدم ADK على الويب، وهي الواجهة نفسها التي استخدمتها محليًا، ولكنها تعمل الآن على Cloud Run.

يمكنك الدردشة مع موظّف الدعم

طلب بحث في القائمة

What spicy dishes do you have?

طلب الحجز

Book a table for 4 on Friday at 7pm. Name: Eve, Phone: 555-0505

التحقّق من الحجز

إنشاء جلسة جديدة ( بدء محادثة جديدة):

Check reservation for 555-0505

تم نشر النظام المتعدّد الوكلاء بالكامل. ينسّق وكيل المطعم على Cloud Run بين خدمتَين خلفيتَين: MCP Toolbox لعمليات قائمة الطعام ووكيل الحجز A2A على Agent Runtime.

11. تهانينا!

أنشأت نظامًا متعدد الوكلاء ونشرته باستخدام بروتوكول A2A على Google Cloud.

ما تعلّمته

إنشاء وكيل ADK يستخدم حالة الجلسة (ToolContext) لإدارة بيانات الحجز بدون قاعدة بيانات
تم نشر وكيل A2A في Agent Runtime باستخدام Agent Platform SDK
استخدام وكيل A2A عن بُعد من وكيل ADK آخر باستخدام RemoteA2aAgent كوكيل فرعي
اختبار النظام بشكل تدريجي: A2A محلي → نشر A2A → دمج جزئي → نشر كامل

إخلاء مساحة

لتجنُّب تحمّل رسوم في حسابك على Google Cloud، احذف الموارد التي تم إنشاؤها في هذا الدرس العملي.

الخيار 1: حذف المشروع (مقترَح)

gcloud projects delete $GOOGLE_CLOUD_PROJECT

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

# Delete the Agent Runtime deployment
uv run python -c "
import vertexai
from google.genai import types
vertexai.init(project='$GOOGLE_CLOUD_PROJECT', location='$REGION')
client = vertexai.Client(
    project='$GOOGLE_CLOUD_PROJECT', location='$REGION',
    http_options=types.HttpOptions(api_version='v1beta1'),
)
agent = client.agent_engines.get(name='$RESERVATION_AGENT_RESOURCE_NAME')
agent.delete(force=True)
print('Agent Runtime deployment deleted.')
"

# Delete Cloud Run services
gcloud run services delete restaurant-agent --region=$REGION --quiet
gcloud run services delete toolbox-service --region=$REGION --quiet

# Delete Cloud SQL instance
gcloud sql instances delete $DB_INSTANCE --quiet

# Delete GCS staging bucket
gsutil rm -r gs://$STAGING_BUCKET