1. 📖 مقدمة

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

في هذا العرض التوضيحي، سنبدأ بتنفيذ A2A باستخدام حزمة تطوير البرامج (SDK) الخاصة بلغة Python. سنتعرّف على حالة استخدام عندما يكون لدينا مساعد شخصي للتسوّق يمكنه مساعدتنا في التواصل مع وكلاء بائعي البرغر والبيتزا للتعامل مع طلبنا.
تستفيد A2A من مبدأ العميل والخادم. في ما يلي مسار A2A النموذجي الذي ستتوقّعه في هذا البرنامج التعليمي

- سيجري برنامج A2A Client أولاً عملية البحث عن جميع بطاقات وكيل خادم A2A التي يمكن الوصول إليها، وسيستخدم معلوماتها لإنشاء برنامج اتصال
- عند الحاجة، سيرسل تطبيق A2A Client رسالة إلى خادم A2A، وسيقيّم الخادم هذه الرسالة على أنّها مهمة يجب إكمالها. إذا تم ضبط عنوان URL الخاص بمستلِم الإشعارات الفورية على عميل A2A وكان متوافقًا مع خادم A2A، سيتمكّن الخادم أيضًا من نشر حالة تقدّم المهمة إلى نقطة النهاية المستلِمة على العميل.
- بعد انتهاء المهمة، سيرسل خادم A2A عنصر الردّ إلى عميل A2A.
خلال هذا الدرس العملي، ستتّبع نهجًا خطوة بخطوة على النحو التالي:
- إعداد مشروع Google Cloud
- إعداد دليل العمل لبيئة الترميز
- نشر وكيل البرغر على Cloud Run
- نشر وكيل البيتزا على Cloud Run
- نشر وكيل شراء (Concierge) على Agent Engine
- التفاعل مع خدمة الاستشارة بشأن الشراء من خلال واجهة محلية
نظرة عامة على البنية
ستنفّذ بنية الخدمة التالية

ستنشر خدمتين تعملان كخادم A2A، وهما وكيل Burger (الذي يستند إلى إطار عمل وكيل CrewAI) ووكيل Pizza (الذي يستند إلى إطار عمل وكيل Langgraph). سيتفاعل المستخدم مباشرةً مع وكيل الشراء (Concierge) الذي سيتم تشغيله باستخدام إطار عمل Agent Development Kit (ADK) الذي سيعمل كعميل A2A.
سيكون لكلّ من هؤلاء الوكلاء بيئة وعملية نشر خاصة به.
المتطلبات الأساسية
- إجادة العمل باستخدام لغة Python
- فهم أساسي لبنية التطبيقات المتكاملة باستخدام خدمة HTTP
ما ستتعلمه
- البنية الأساسية لخادم A2A
- البنية الأساسية لبرنامج A2A Client
- نشر خدمة الوكيل على Cloud Run
- نشر خدمة الوكيل في Agent Engine
- كيفية ربط تطبيق A2A بخادم A2A
- بنية الطلب والاستجابة عند استخدام اتصال غير متواصل
المتطلبات
- متصفّح الويب Chrome
- حساب Gmail
- مشروع على السحابة الإلكترونية مع تفعيل حساب الفوترة
يستخدم هذا الدرس التطبيقي حول الترميز، المصمّم للمطوّرين من جميع المستويات (بما في ذلك المبتدئين)، لغة Python في تطبيق المثال. ومع ذلك، لا يُشترط معرفة لغة Python لفهم المفاهيم المقدَّمة.
2. 🚀 إعداد بيئة تطوير ورشة العمل
الخطوة 1: اختيار "المشروع النشط" في Cloud Console
في Google Cloud Console، ضِمن صفحة اختيار المشروع، اختَر أو أنشِئ مشروعًا على Google Cloud (راجِع القسم العلوي الأيمن من وحدة التحكّم).

انقر على هذا الرمز، وستظهر لك قائمة بجميع مشاريعك كما في المثال التالي:

القيمة الموضّحة في المربّع الأحمر هي معرّف المشروع وسيتم استخدام هذه القيمة في جميع أنحاء البرنامج التعليمي.
تأكَّد من تفعيل الفوترة لمشروعك على Cloud. للتأكّد من ذلك، انقر على رمز الهامبرغر ☰ في الشريط العلوي الأيمن الذي يعرض قائمة التنقّل وابحث عن قائمة "الفوترة".

إذا ظهرت لك الرسالة تم ربط"حساب فوترة تجريبي على Google Cloud Platform"، يعني ذلك أنّ مشروعك جاهز للاستخدام في هذا البرنامج التعليمي. إذا لم يكن كذلك، ارجع إلى بداية هذا الدليل التوجيهي/التعليمي وحصّل قيمة حساب الفوترة.

الخطوة 2: التعرّف على Cloud Shell
ستستخدم Cloud Shell في معظم أجزاء البرامج التعليمية. انقر على "تفعيل Cloud Shell" في أعلى وحدة تحكّم Google Cloud. إذا طُلب منك التفويض، انقر على تفويض.


بعد الاتصال بـ Cloud Shell، علينا التحقّق مما إذا كان قد تمّت مصادقة الصدفة ( أو الوحدة الطرفية) باستخدام حسابنا.
gcloud auth list
إذا ظهر لك حساب Gmail الشخصي كما في المثال أدناه، يعني ذلك أنّ كل شيء على ما يرام.
Credentialed Accounts
ACTIVE: *
ACCOUNT: alvinprayuda@gmail.com
To set the active account, run:
$ gcloud config set account `ACCOUNT`
إذا لم يكن كذلك، جرِّب إعادة تحميل المتصفّح وتأكَّد من النقر على تفويض عند مطالبتك بذلك ( قد يتمّ مقاطعة العملية بسبب مشكلة في الاتصال).
بعد ذلك، نحتاج أيضًا إلى التحقّق مما إذا كان الغلاف قد تم إعداده مسبقًا باستخدام معرّف المشروع الصحيح الذي لديك. إذا رأيت قيمة داخل ( ) قبل رمز $ في الوحدة الطرفية ( في لقطة الشاشة أدناه، القيمة هي "a2a-agent-engine")، تعرض هذه القيمة المشروع الذي تم إعداده لجلسة الغلاف النشطة.

إذا كانت القيمة المعروضة صحيحة، يمكنك تخطّي الأمر التالي. ومع ذلك، إذا كان غير صحيح أو غير متوفّر، شغِّل الأمر التالي
gcloud config set project <YOUR_PROJECT_ID>
بعد ذلك، استنسِخ دليل العمل الخاص بهذا الدرس التطبيقي حول الترميز من Github، ونفِّذ الأمر التالي. سيتم إنشاء دليل العمل في الدليل purchasing-concierge-a2a
git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a
الخطوة 3: التعرّف على "محرّر Cloud Shell" وإعداد دليل عمل التطبيق
الآن، يمكننا إعداد محرّر الرموز البرمجية لتنفيذ بعض مهام الترميز. سنستخدم "محرّر Cloud Shell" لهذا الغرض.
انقر على الزر فتح المحرِّر، وسيؤدي ذلك إلى فتح محرِّر Cloud Shell 
بعد ذلك، انتقِل إلى القسم العلوي من "محرّر Cloud Shell" وانقر على ملف->فتح مجلد (File->Open Folder)، وابحث عن دليل اسم المستخدم وابحث عن دليل purchasing-concierge-a2a ثم انقر على الزر "حسنًا" (OK). سيؤدي ذلك إلى جعل الدليل الذي تم اختياره هو دليل العمل الرئيسي. في هذا المثال، اسم المستخدم هو alvinprayuda، وبالتالي يظهر مسار الدليل أدناه


من المفترض أن يظهر "محرّر Cloud Shell" الآن على النحو التالي

الآن، افتح الوحدة الطرفية للمحرّر. يمكنك إجراء ذلك من خلال النقر على Terminal -> New Terminal في شريط القوائم، أو استخدام Ctrl + Shift + C، وسيؤدي ذلك إلى فتح نافذة طرفية في الجزء السفلي من المتصفح.

يجب أن تكون المحطة الطرفية النشطة الحالية داخل دليل العمل purchasing-concierge-a2a. سنستخدم الإصدار 3.12 من لغة Python في هذا الدرس التطبيقي حول الترميز، كما سنستخدم أداة إدارة مشاريع Python uv لتسهيل عملية إنشاء إصدار Python وبيئة افتراضية وإدارتهما. تم تثبيت حزمة uv هذه مسبقًا على Cloud Shell.
نفِّذ هذا الأمر لتثبيت الاعتمادات المطلوبة في البيئة الافتراضية في الدليل .venv.
uv sync --frozen
راجِع ملف pyproject.toml للاطّلاع على التبعيات المحدّدة لهذا الدليل التعليمي، وهي a2a-sdk, google-adk, and gradio.
الآن، علينا تفعيل واجهات برمجة التطبيقات المطلوبة من خلال الأمر الموضّح أدناه. قد يستغرق هذا بعض الوقت.
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
عند تنفيذ الأمر بنجاح، من المفترض أن تظهر لك رسالة مشابهة للرسالة الموضّحة أدناه:
Operation "operations/..." finished successfully.
3- 🚀 نشر وكلاء بائعين عن بُعد على خادم A2A على Cloud Run
في هذه الخطوة، سننفّذ وكيلَي البائعَين البعيدَين المحدّدَين بالمربّع الأحمر. سيتم تشغيل وكيل البرغر باستخدام إطار عمل الوكيل CrewAI، وسيتم تشغيل وكيل البيتزا باستخدام وكيل Langgraph

4. 🚀 نشر وكيل بائع البرغر - خادم A2A
يتوفّر رمز مصدر وكيل البرجر ضمن الدليل remote_seller_agents/burger_agent.
جميع الملفات المتوفّرة في الدليل remote_seller_agents/burger_agent كافية لنشر الوكيل على Cloud Run لكي يمكن الوصول إليه كخدمة. نفِّذ الأمر التالي لنشره
gcloud run deploy burger-agent \
--source remote_seller_agents/burger_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
إذا طُلب منك إنشاء مستودع حاويات للنشر من المصدر، أجب بـ Y. بعد نشر التطبيق بنجاح، سيظهر سجلّ على النحو التالي.
Service [burger-agent] revision [burger-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://burger-agent-xxxxxxxxx.us-central1.run.app
سيكون الجزء xxxx هنا معرّفًا فريدًا عند نشر الخدمة.
افتح علامة تبويب متصفّح جديدة وانتقِل إلى مسار https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json لخدمات وكيل برجر التي تم نشرها من خلال المتصفّح. هذا هو عنوان URL للوصول إلى بطاقة وكيل A2A التي تم نشرها على الخادم.
في حال تم نشرها بنجاح، ستظهر لك الاستجابة كما هو موضّح أدناه في المتصفّح عند الوصول إلى بطاقة الوكيل

هذه هي معلومات بطاقة وكيل البرغر التي يجب أن تكون متاحة لأغراض الاكتشاف.
لاحظ أنّ قيمة url لا تزال مضبوطة على http://0.0.0.0:8080/ هنا. يجب أن تكون قيمة url هذه هي المعلومات الرئيسية التي يستخدمها تطبيق A2A Client لإرسال الرسائل من العالم الخارجي، ولكن لم يتم ضبطها بشكل صحيح.
علينا تعديل هذه القيمة إلى عنوان URL الخاص بخدمة وكيل البرجر من خلال إضافة متغيّر بيئة إضافي HOST_OVERRIDE.
تعديل قيمة عنوان URL الخاص بـ Burger Agent على بطاقة Agent من خلال متغير البيئة
لإضافة HOST_OVERRIDE إلى خدمة وكيل البرغر، اتّبِع الخطوات التالية
- ابحث عن Cloud Run في شريط البحث أعلى وحدة تحكّم السحابة الإلكترونية

- انقر على خدمة burger-agent التي تم نشرها سابقًا على Cloud Run
- انسخ عنوان URL الخاص بخدمة البرغر، ثم انقر على تعديل ونشر إصدار جديد.

- بعد ذلك، انقر على قسم المتغيرات والأسرار.

- بعد ذلك، انقر على إضافة متغيّر واضبط قيمة
HOST_OVERRIDEعلى عنوان URL للخدمة ( العنوان الذي يتضمّن النمطhttps://burger-agent-xxxxxxxxx.us-central1.run.app).

- أخيرًا، انقر على الزر نشر لإعادة نشر خدمتك.

عند الوصول إلى بطاقة وكيل burger-agent مرة أخرى في المتصفح https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json ، ستكون القيمة url قد تم ضبطها بشكل صحيح

5- 🚀 نشر وكيل بائع البيتزا - خادم A2A
وبالمثل، يقع رمز مصدر وكيل البيتزا ضمن الدليل remote_seller_agents/pizza_agent.
على غرار خطوة نشر وكيل البرغر السابقة، تكفي جميع الملفات المتوفرة ضمن الدليل remote_seller_agents/pizza_agent لنشر الوكيل على Cloud Run كي يمكن الوصول إليه كخدمة. نفِّذ الأمر التالي لنشره
gcloud run deploy pizza-agent \
--source remote_seller_agents/pizza_agent \
--port=8080 \
--allow-unauthenticated \
--min 1 \
--region us-central1 \
--update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
--update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}
بعد نشر التطبيق بنجاح، سيظهر سجلّ على النحو التالي.
Service [pizza-agent] revision [pizza-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic. Service URL: https://pizza-agent-xxxxxxxxx.us-central1.run.app
سيكون الجزء xxxx هنا معرّفًا فريدًا عند نشر الخدمة.
وينطبق الأمر نفسه على وكيل البرغر، فعند محاولة الانتقال إلى مسار https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json لخدمات وكيل البيتزا التي تم نشرها من خلال المتصفح للوصول إلى بطاقة وكيل الخادم A2A، لم يتم ضبط قيمة url لوكيل البيتزا على بطاقة الوكيل بشكل صحيح بعد. علينا أيضًا إضافة HOST_OVERRIDE إلى متغير البيئة
تعديل قيمة عنوان URL الخاص بـ Pizza Agent على بطاقة Agent من خلال متغير البيئة
لإضافة HOST_OVERRIDE إلى خدمة وكيل البيتزا، اتّبِع الخطوات التالية
- ابحث عن Cloud Run في شريط البحث أعلى وحدة تحكّم السحابة الإلكترونية

- انقر على خدمة pizza-agent التي تم نشرها سابقًا على Cloud Run

- انقر على تعديل ونشر نسخة جديدة.

- انسخ عنوان URL الخاص بخدمة البيتزا، ثم انقر على قسم المتغيرات والأسرار.

- بعد ذلك، انقر على إضافة متغيّر واضبط قيمة
HOST_OVERRIDEعلى عنوان URL للخدمة ( العنوان الذي يتضمّن النمطhttps://pizza-agent-xxxxxxxxx.us-central1.run.app).

- أخيرًا، انقر على الزر نشر لإعادة نشر خدمتك.

الآن، عند الوصول إلى بطاقة وكيل pizza-agent مرة أخرى في المتصفح https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json، ستكون قيمة url قد تم ضبطها بشكل صحيح

في هذه المرحلة، نكون قد نشرنا بنجاح كلاً من خدمة البرغر وخدمة البيتزا على Cloud Run.
6. 🚀 نشر "وكيل الشراء" - عميل A2A إلى Agent Engine
في هذه الخطوة، سننفّذ وكيل خدمة الاستقبال والإرشاد الخاص بالشراء. هذا هو الوكيل الذي سنتفاعل معه.

يتوفّر رمز المصدر لوكيل شراء (Concierge) في الدليل purchasing_concierge. يمكن فحص عملية تهيئة الوكيل في البرنامج النصي purchasing_concierge/purchasing_agent.py.
اتّبِع الخطوات التالية لتفعيلها :
- أولاً، علينا إنشاء مساحة تخزين مرحلية في Cloud Storage
gcloud storage buckets create gs://purchasing-concierge-{your-project-id} --location=us-central1
- الآن، علينا إعداد متغير .env أولاً، لننسخ ملف .env.example إلى ملف .env
cp .env.example .env
- الآن، افتح ملف .env وسيظهر لك المحتوى التالي
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL={your-pizza-agent-url}
BURGER_SELLER_AGENT_URL={your-burger-agent-url}
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
سيتواصل هذا الوكيل مع وكيل البرغر والبيتزا، لذا علينا تقديم بيانات الاعتماد المناسبة لكليهما. علينا تعديل PIZZA_SELLER_AGENT_URL وBURGER_SELLER_AGENT_URL باستخدام عنوان URL لخدمة Cloud Run من الخطوات السابقة.
إذا نسيت ذلك، انتقِل إلى وحدة تحكّم Cloud Run. اكتب "Cloud Run" في شريط البحث أعلى وحدة التحكّم وانقر بزر الماوس الأيمن على رمز Cloud Run لفتحه في علامة تبويب جديدة.

من المفترض أن تظهر لك خدمات وكيل البائع عن بُعد التي تم نشرها سابقًا كما هو موضّح أدناه

للاطّلاع على عنوان URL المتاح للجميع لهذه الخدمات، انقر على إحدى الخدمات وستتم إعادة توجيهك إلى صفحة "تفاصيل الخدمة". يمكنك الاطّلاع على عنوان URL في المنطقة العلوية بجانب معلومات المنطقة.

يجب أن يبدو متغير البيئة النهائي مشابهًا لهذا المتغير
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}
- أصبحنا الآن جاهزين لنشر وكيل خدمة التسوق. سننفّذها على محرك الوكيل، وسيكون رمز التنفيذ داخل النص البرمجي
deploy_to_agent_engine.py.
يمكننا نشرها عن طريق تنفيذ النص البرمجي:
uv run deploy_to_agent_engine.py
بعد نشر التطبيق بنجاح، سيظهر سجلّ على النحو التالي. سيظهر لك اسم مورد Agent Engine على النحو التالي: "projects/xxxx/locations/us-central1/reasoningEngines/yyyy"
AgentEngine created. Resource name: projects/xxxx/locations/us-central1/reasoningEngines/yyyy
To use this AgentEngine in another session:
agent_engine = vertexai.agent_engines.get('projects/xxxx/locations/us-central1/reasoningEngines/yyyy)
Deployed remote app resource: projects/xxxx/locations/us-central1/reasoningEngines/xxxx
وعندما نفحصه في لوحة بيانات "محرك الوكيل" (ابحث عن "محرك الوكيل" في شريط البحث)، سيظهر النشر السابق.

يمكنك أيضًا التأكّد من عرض اسم مورد Agent Engine هناك، ثمّ يمكننا استخدام اسم المورد هذا لاختباره.
بعد ذلك، عدِّل AGENT_ENGINE_RESOURCE_NAME في ملف .env بهذه القيمة. تأكَّد من تقديم اسم مورد محرك الوكيل الصحيح. يجب أن يظهر ملف .env على النحو التالي:
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME=projects/xxxx/locations/us-central1/reasoningEngines/yyyy
اختبار الوكيل الذي تم نشره على Agent Engine
يمكن التفاعل مع محرك الوكيل من خلال الأمر curl وحزمة تطوير البرامج (SDK). على سبيل المثال، شغِّل الأمر التالي لمحاولة التفاعل مع الوكيل الذي تم نشره.
يمكنك محاولة إرسال هذا الاستعلام للتحقّق مما إذا تم نشر الوكيل بنجاح. شغِّل النص البرمجي التالي test_agent_engine.sh
bash test_agent_engine.sh
يمكنك فحص النص البرمجي، وسترى أنّنا نحاول أن نطلب من الوكيل "أريد قائمة بوجبات البرغر المتاحة".
في حال نجاح العملية، سيتم عرض عدة أحداث استجابة يتم بثها على وحدة التحكّم على النحو التالي
{
"content": {
"parts": [
{
"text": "Here is our burger menu:\n- Classic Cheeseburger: IDR 85K\n- Double Cheeseburger: IDR 110K\n- Spicy Chicken Burger: IDR 80K\n- Spicy Cajun Burger: IDR 85K"
}
],
"role": "model"
},
"usage_metadata": {
"candidates_token_count": 51,
"candidates_tokens_details": [
{
"modality": "TEXT",
"token_count": 51
}
],
"prompt_token_count": 907,
"prompt_tokens_details": [
{
"modality": "TEXT",
"token_count": 907
}
],
"total_token_count": 958,
"traffic_type": "ON_DEMAND"
},
"invocation_id": "e-14679918-af68-45f1-b942-cf014368a733",
"author": "purchasing_agent",
"actions": {
"state_delta": {},
"artifact_delta": {},
"requested_auth_configs": {}
},
"id": "dbe7fc43-b82a-4f3e-82aa-dd97afa8f15b",
"timestamp": 1754287348.941454
}
سنحاول استخدام واجهة المستخدم في الخطوة التالية، ولكن لنناقش أولاً المكوّنات الأساسية والمسار النموذجي لبرامج A2A.
7. 🚀 اختبار الدمج وفحص الحمولة
لنلقِ نظرة الآن على وكيل الشراء (Concierge) من خلال التفاعل مع الوكيل عن بُعد باستخدام واجهة مستخدم على الويب. نفِّذ الأمر التالي لنشر تطبيق Gradio. يتطلّب تشغيل هذا التطبيق أن تكون قد ملأت ملف .env بشكل صحيح.
uv run purchasing_concierge_ui.py
ستظهر النتيجة التالية في حال نجاح العملية
* Running on local URL: http://0.0.0.0:8080 * To create a public link, set `share=True` in `launch()`.
بعد ذلك، انقر مع الضغط على Ctrl على عنوان URL http://0.0.0.0:8080 في الوحدة الطرفية أو انقر على زر معاينة الويب لفتح واجهة مستخدم الويب.

جرِّب إجراء محادثة على النحو التالي :
- عرض قائمة طعام البرغر والبيتزا
- أريد طلب بيتزا دجاج مشوي واحدة وبرغر كاجون حار واحد
ويمكنك مواصلة المحادثة إلى أن تنتهي من الطلب. اطّلِع على سير التفاعل وما هي استدعاءات الأدوات وردودها. الصورة التالية هي مثال على نتيجة التفاعل.




يمكننا أن نرى أنّ التواصل مع وكيلَين مختلفَين يؤدي إلى سلوكَين مختلفَين، ويمكن أن يتعامل A2A مع هذا الأمر بشكل جيد. يقبل وكيل بائع البيتزا طلب وكيل الشراء مباشرةً، بينما يحتاج وكيل البرغر إلى تأكيد منّا قبل المتابعة في تنفيذ طلبنا، وبعد تأكيدنا، يمكن للوكيل إرسال التأكيد إلى وكيل البرغر.
بعد أن انتهينا من المفاهيم الأساسية لـ A2A، سنتعرّف على كيفية تنفيذها كبنية قائمة على البرنامج والخادم.
8. 💡 [شرح الرمز البرمجي] مفهوم خادم A2A وتنفيذه
يمكن فحص عملية تهيئة وكيل البائع عن بُعد في النص البرمجي remote_seller_agents/*/agent.py. في ما يلي مقتطف الرمز البرمجي لوكلاء البائعين.
Burger Agent
from crewai import Agent, Crew, LLM, Task, Process
from crewai.tools import tool
...
model = LLM(
model="vertex_ai/gemini-2.5-flash-lite", # Use base model name without provider prefix
)
burger_agent = Agent(
role="Burger Seller Agent",
goal=(
"Help user to understand what is available on burger menu and price also handle order creation."
),
backstory=("You are an expert and helpful burger seller agent."),
verbose=False,
allow_delegation=False,
tools=[create_burger_order],
llm=model,
)
agent_task = Task(
description=self.TaskInstruction,
agent=burger_agent,
expected_output="Response to the user in friendly and helpful manner",
)
crew = Crew(
tasks=[agent_task],
agents=[burger_agent],
verbose=False,
process=Process.sequential,
)
inputs = {"user_prompt": query, "session_id": sessionId}
response = crew.kickoff(inputs)
return response
...
Pizza Agent
from langchain_google_vertexai import ChatVertexAI
from langgraph.prebuilt import create_react_agent
...
self.model = ChatVertexAI(
model="gemini-2.5-flash-lite",
location=os.getenv("GOOGLE_CLOUD_LOCATION"),
project=os.getenv("GOOGLE_CLOUD_PROJECT"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
self.model,
tools=self.tools,
checkpointer=memory,
prompt=self.SYSTEM_INSTRUCTION,
)
...
كما تلاحظ، تم إنشاء هذين الوكيلَين باستخدام إطارَي عمل مختلفَين تمامًا ( CrewAI وLanggraph) مقارنةً بوكيل العميل ( ADK). مع A2A، لا يشكّل ذلك مشكلة، إذ لا نحتاج إلى أن يشارك الوكلاء رمزهم الداخلي للتواصل مع بعضهم البعض، ولا يهمّ إطار العمل المستخدَم أو اللغة المستخدَمة أو مكان نشرهم.
المكوّنات الأساسية لخادم A2A
لنناقش الآن المفهوم الأساسي ومكوّنات خادم A2A
بطاقة الوكيل
يجب أن يتضمّن كل خادم A2A بطاقة وكيل يمكن الوصول إليها من خلال المورد /.well-known/agent.json. يهدف ذلك إلى دعم مرحلة الاستكشاف على تطبيق A2A Client، ما يتيح تقديم معلومات وسياقات كاملة حول كيفية الوصول إلى الوكيل ومعرفة جميع إمكاناته. وهي تشبه إلى حدّ ما مستندات واجهة برمجة التطبيقات الموثّقة جيدًا باستخدام Swagger أو Postman.
هذه هي بطاقة وكيل الذكاء الاصطناعي الخاصة بمطعم البرغر التي تم نشرها
{
"capabilities": {
"streaming": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"description": "Helps with creating burger orders",
"name": "burger_seller_agent",
"protocolVersion": "0.2.6",
"skills": [
{
"description": "Helps with creating burger orders",
"examples": [
"I want to order 2 classic cheeseburgers"
],
"id": "create_burger_order",
"name": "Burger Order Creation Tool",
"tags": [
"burger order creation"
]
}
],
"url": "https://burger-agent-109790610330.us-central1.run.app",
"version": "1.0.0"
}
توضّح بطاقات العملاء الافتراضيين هذه العديد من المكوّنات المهمة، مثل مهارات العميل الافتراضي وإمكانات البث والوسائط المتوافقة وإصدار البروتوكول وغير ذلك.
يمكن استخدام كل هذه المعلومات لتطوير آلية تواصل مناسبة كي يتمكّن عميل A2A من التواصل بشكلٍ سليم. تضمن طريقة التواصل وآلية المصادقة المتوافقتان إمكانية إنشاء التواصل بشكل صحيح، ويمكن تضمين معلومات skills الوكيل في طلب نظام عميل A2A لمنح وكيل العميل سياقًا حول إمكانات الوكيل البعيد ومهاراته التي سيتم استدعاؤها. يمكنك العثور على حقول أكثر تفصيلاً لبطاقة الوكيل هذه في هذه المستندات.
في الرمز البرمجي، يتم إنشاء بطاقة الوكيل باستخدام حزمة تطوير البرامج (SDK) الخاصة بـ A2A بلغة Python. اطّلِع على مقتطف الرمز البرمجي remote_seller_agents/burger_agent/main.py أدناه لمعرفة طريقة التنفيذ.
...
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="create_burger_order",
name="Burger Order Creation Tool",
description="Helps with creating burger orders",
tags=["burger order creation"],
examples=["I want to order 2 classic cheeseburgers"],
)
agent_host_url = (
os.getenv("HOST_OVERRIDE")
if os.getenv("HOST_OVERRIDE")
else f"http://{host}:{port}/"
)
agent_card = AgentCard(
name="burger_seller_agent",
description="Helps with creating burger orders",
url=agent_host_url,
version="1.0.0",
defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
...
يمكننا الاطّلاع على عدّة حقول، مثل:
-
AgentCapabilities: بيان بالوظائف الاختيارية الإضافية التي تتيحها خدمة الوكيل، مثل إمكانية بث المحتوى و/أو إتاحة الإشعارات الفورية AgentSkill: الأدوات أو الوظائف التي يتيحها الوكيل-
Input/OutputModes: نوع الوسائط المتوافق مع الإدخال/الإخراج Url: عنوان للتواصل مع الوكيل
في هذا الإعداد، نوفّر إنشاء عنوان URL ديناميكي لمضيف الوكيل، ما يسهّل التبديل بين الاختبار المحلي والنشر على السحابة الإلكترونية، ولهذا السبب نحتاج إلى إضافة المتغيّر HOST_OVERRIDE في الخطوة السابقة.
قائمة انتظار المهام وبرنامج تنفيذ الوكيل
قد يعالج خادم A2A الطلبات من وكلاء أو مستخدمين مختلفين ويكون قادرًا على عزل كل مهمة تمامًا. للاطّلاع على سياقات هذه الصور بشكل أفضل، يمكنك فحص الصورة أدناه

وبالتالي، يجب أن يكون كل خادم A2A قادرًا على تتبُّع المهام الواردة وتخزين المعلومات المناسبة عنها. توفّر حزمة تطوير البرامج (SDK) من A2A وحدات لمواجهة هذا التحدي في خادم A2A. أولاً، يمكننا إنشاء منطق حول كيفية التعامل مع الطلب الوارد. من خلال وراثة فئة AgentExecutor المجردة، يمكننا التحكّم في كيفية إدارة تنفيذ المهام وإلغائها. يمكن فحص عملية التنفيذ هذه في الوحدة remote_seller_agents/burger_agent/agent_executor.py ( مسار مشابه لحالة بائع البيتزا)
...
class BurgerSellerAgentExecutor(AgentExecutor):
"""Burger Seller AgentExecutor."""
def __init__(self):
self.agent = BurgerSellerAgent()
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
try:
result = self.agent.invoke(query, context.context_id)
print(f"Final Result ===> {result}")
parts = [Part(root=TextPart(text=str(result)))]
await event_queue.enqueue_event(
completed_task(
context.task_id,
context.context_id,
[new_artifact(parts, f"burger_{context.task_id}")],
[context.message],
)
)
except Exception as e:
print("Error invoking agent: %s", e)
raise ServerError(error=ValueError(f"Error invoking agent: {e}")) from e
async def cancel(
self, request: RequestContext, event_queue: EventQueue
) -> Task | None:
raise ServerError(error=UnsupportedOperationError())
...
في الرمز أعلاه، ننفّذ مخطط معالجة أساسيًا يتم فيه استدعاء العامل مباشرةً عند ورود الطلب وإرسال أحداث المهام المكتملة بعد انتهاء الاستدعاء. ومع ذلك، لم نطبّق طريقة الإلغاء هنا لأنّها كانت تُعتبر عملية قصيرة الأمد.
بعد إنشاء المنفّذ، يمكننا الاستفادة مباشرةً من DefaultRequestHandler وInMemoryTaskStore وA2AStarletteApplication المضمّنة لتشغيل خادم HTTP. يمكن فحص عملية التنفيذ هذه في remote_seller_agents/burger_agent/__main__.py
...
request_handler = DefaultRequestHandler(
agent_executor=BurgerSellerAgentExecutor(),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=agent_card, http_handler=request_handler
)
uvicorn.run(server.build(), host=host, port=port)
...
ستوفّر لك هذه الوحدة تنفيذ مسار /.well-known/agent.json للوصول إلى بطاقة الوكيل، بالإضافة إلى نقطة نهاية POST لدعم بروتوكول A2A
ملخّص
باختصار، حتى الآن، تم نشر خادم A2A باستخدام حزمة تطوير البرامج (SDK) للغة Python، ويمكنه توفير الوظيفتَين أدناه:
- نشر بطاقة الوكيل على مسار
/.well-known/agent.json - التعامل مع طلب JSON-RPC باستخدام نظام صفوف انتظار المهام في الذاكرة
يمكن فحص نقطة الدخول عند بدء هذه الوظائف في النص البرمجي __main__.py ( على remote_seller_agents/burger_agent أو remote_seller_agents/pizza_agent) .
9. 💡 [شرح الرمز البرمجي] نشر Agent Engine
في ما يلي مقتطف الرمز البرمجي لوكيل شراء (Concierge) في purchasing_concierge/purchasing_agent.py:
from google.adk import Agent
...
def create_agent(self) -> Agent:
return Agent(
model="gemini-2.5-flash-lite",
name="purchasing_agent",
instruction=self.root_instruction,
before_model_callback=self.before_model_callback,
before_agent_callback=self.before_agent_callback,
description=(
"This purchasing agent orchestrates the decomposition of the user purchase request into"
" tasks that can be performed by the seller agents."
),
tools=[
self.send_task,
],
)
...
تم إنشاء هذا الوكيل باستخدام "حزمة تطوير الوكلاء" (ADK) وتم نشره على Agent Engine.
Vertex AI Agent Engine هي مجموعة من الخدمات التي تتيح للمطوّرين نشر وكلاء الذكاء الاصطناعي وإدارتهم وتوسيع نطاقهم في مرحلة الإنتاج. يتولّى هذا الإطار البنية الأساسية لتوسيع نطاق الوكلاء في مرحلة الإنتاج، ما يتيح لنا التركيز على إنشاء التطبيقات. يمكنك الاطّلاع على مزيد من المعلومات حول هذا الموضوع في هذا المستند . إذا كنّا نحتاج سابقًا إلى إعداد الملفات اللازمة لنشر خدمة الوكيل (مثل نص خادم main البرمجي وDockerfile)، يمكننا في هذه الحالة نشر الوكيل مباشرةً من نص Python البرمجي بدون الحاجة إلى تطوير خدمة الخلفية الخاصة بنا باستخدام مجموعة من ADK وAgent Engine.
في هذا البرنامج التعليمي، ننشر باستخدام النص البرمجي deploy_to_agent_engine.py الذي يتم عرض المحتوى أدناه
import vertexai
from vertexai.preview import reasoning_engines
from vertexai import agent_engines
from dotenv import load_dotenv
import os
from purchasing_concierge.agent import root_agent
load_dotenv()
PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
STAGING_BUCKET = os.getenv("STAGING_BUCKET")
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
staging_bucket=STAGING_BUCKET,
)
adk_app = reasoning_engines.AdkApp(
agent=root_agent,
)
remote_app = agent_engines.create(
agent_engine=adk_app,
display_name="purchasing-concierge",
requirements=[
"google-cloud-aiplatform[adk,agent_engines]",
"a2a-sdk==0.2.16",
],
extra_packages=[
"./purchasing_concierge",
],
env_vars={
"GOOGLE_GENAI_USE_VERTEXAI": os.environ["GOOGLE_GENAI_USE_VERTEXAI"],
"PIZZA_SELLER_AGENT_URL": os.environ["PIZZA_SELLER_AGENT_URL"],
"BURGER_SELLER_AGENT_URL": os.environ["BURGER_SELLER_AGENT_URL"],
},
)
print(f"Deployed remote app resource: {remote_app.resource_name}")
في ما يلي الخطوات اللازمة لنشر وكيل ADK في محرك الوكيل. أولاً، علينا إنشاء كائن AdkApp من حزمة تطوير البرامج (ADK) root_agent. بعد ذلك، يمكننا تنفيذ طريقة agent_engines.create من خلال توفير عنصر adk_app، وتحديد المتطلبات في الحقل requirements، وتحديد مسار دليل الوكيل في extra_packages ( يمكنك أيضًا توفير أدلة وملفات أخرى إذا لزم الأمر هنا)، وتوفير متغيرات env اللازمة.
10. 💡 [شرح الرمز البرمجي] مفهوم عميل A2A وتنفيذه

الصورة المعروضة أعلاه هي التدفق النموذجي لتفاعلات A2A:
- سيحاول العميل العثور على أي بطاقة وكيل منشورة في عنوان URL الخاص بالوكيل البعيد المقدَّم في المسار
/.well-known/agent.json - بعد ذلك، سيرسل الخادم رسالة إلى هذا الوكيل عند الضرورة تتضمّن الرسالة ومعلمات البيانات الوصفية اللازمة ( مثل معرّف الجلسة والسياق السابق وما إلى ذلك). وسيتعامل الخادم مع هذه الرسالة على أنّها مهمة يجب إكمالها.
- يعالج خادم A2A الطلب، وإذا كان الخادم يتيح إرسال الإشعارات الفورية، سيكون بإمكانه أيضًا نشر بعض الإشعارات أثناء معالجة المهمة ( لا تتضمّن هذه التجربة العملية هذه الوظيفة).
- بعد الانتهاء، سيرسل خادم A2A عنصر الردّ إلى العميل.
بعض العناصر الأساسية للتفاعلات المذكورة أعلاه هي ما يلي (يمكنك الاطّلاع على مزيد من التفاصيل هنا) :
- الرسالة: هي تبادل للمعلومات بين العميل وموظف الدعم عن بُعد
- المهمة: هي وحدة العمل الأساسية التي تديرها خدمة A2A، ويتم تحديدها من خلال معرّف فريد.
- الناتج: هو نتيجة (مثل مستند أو صورة أو بيانات منظَّمة) ينشئها الوكيل نتيجةً لمهمة، ويتألف من أجزاء
- الجزء: هو أصغر وحدة محتوى ضمن "رسالة" أو "عنصر". يمكن أن يكون الجزء نصًا أو صورة أو فيديو أو ملفًا أو غير ذلك.
التعرّف على البطاقات
عند إعداد خدمة "العميل من التطبيق إلى التطبيق"، تتمثل العملية النموذجية في محاولة الحصول على معلومات بطاقة الوكيل وتخزينها لتسهيل الوصول إليها عند الحاجة. في هذا الدرس التطبيقي حول الترميز، ننفّذها على before_agent_callback، ويمكنك الاطّلاع على التنفيذ في purchasing_concierge/purchasing_agent.py، راجِع مقتطف الرمز أدناه
...
async def before_agent_callback(self, callback_context: CallbackContext):
if not self.a2a_client_init_status:
httpx_client = httpx.AsyncClient(timeout=httpx.Timeout(timeout=30))
for address in self.remote_agent_addresses:
card_resolver = A2ACardResolver(
base_url=address, httpx_client=httpx_client
)
try:
card = await card_resolver.get_agent_card()
remote_connection = RemoteAgentConnections(
agent_card=card, agent_url=card.url
)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
except httpx.ConnectError:
print(f"ERROR: Failed to get agent card from : {address}")
agent_info = []
for ra in self.list_remote_agents():
agent_info.append(json.dumps(ra))
self.agents = "\n".join(agent_info)
...
في هذه الخطوة، نحاول الوصول إلى جميع بطاقات الوكيل المتاحة باستخدام وحدة A2ACardResolver المضمّنة في عميل A2A، ثم نجمع معلومات الاتصال اللازمة لإرسال رسالة إلى الوكيل، وبعد ذلك نحتاج أيضًا إلى إدراج جميع الوكلاء المتاحين ومواصفاتهم في الطلب حتى يعرف وكيلنا أنّه يمكنه التواصل مع هؤلاء الوكلاء.
أداة "طلب المهمة وإرسالها"
هذا هو الطلب والأداة اللذان نقدّمهما إلى وكيل ADK هنا
...
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_active_agent(context)
return f"""You are an expert purchasing delegator that can delegate the user product inquiry and purchase request to the
appropriate seller remote agents.
Execution:
- For actionable tasks, you can use `send_task` to assign tasks to remote agents to perform.
- When the remote agent is repeatedly asking for user confirmation, assume that the remote agent doesn't have access to user's conversation context.
So improve the task description to include all the necessary information related to that agent
- Never ask user permission when you want to connect with remote agents. If you need to make connection with multiple remote agents, directly
connect with them without asking user permission or asking user preference
- Always show the detailed response information from the seller agent and propagate it properly to the user.
- If the remote seller is asking for confirmation, rely the confirmation question to the user if the user haven't do so.
- If the user already confirmed the related order in the past conversation history, you can confirm on behalf of the user
- Do not give irrelevant context to remote seller agent. For example, ordered pizza item is not relevant for the burger seller agent
- Never ask order confirmation to the remote seller agent
Please rely on tools to address the request, and don't make up the response. If you are not sure, please ask the user for more details.
Focus on the most recent parts of the conversation primarily.
If there is an active agent, send the request to that agent with the update task tool.
Agents:
{self.agents}
Current active seller agent: {current_agent["active_agent"]}
"""
...
async def send_task(self, agent_name: str, task: str, tool_context: ToolContext):
"""Sends a task to remote seller agent
This will send a message to the remote agent named agent_name.
Args:
agent_name: The name of the agent to send the task to.
task: The comprehensive conversation context summary
and goal to be achieved regarding user inquiry and purchase request.
tool_context: The tool context this method runs in.
Yields:
A dictionary of JSON data.
"""
if agent_name not in self.remote_agent_connections:
raise ValueError(f"Agent {agent_name} not found")
state = tool_context.state
state["active_agent"] = agent_name
client = self.remote_agent_connections[agent_name]
if not client:
raise ValueError(f"Client not available for {agent_name}")
session_id = state["session_id"]
task: Task
message_id = ""
metadata = {}
if "input_message_metadata" in state:
metadata.update(**state["input_message_metadata"])
if "message_id" in state["input_message_metadata"]:
message_id = state["input_message_metadata"]["message_id"]
if not message_id:
message_id = str(uuid.uuid4())
payload = {
"message": {
"role": "user",
"parts": [
{"type": "text", "text": task}
], # Use the 'task' argument here
"messageId": message_id,
"contextId": session_id,
},
}
message_request = SendMessageRequest(
id=message_id, params=MessageSendParams.model_validate(payload)
)
send_response: SendMessageResponse = await client.send_message(
message_request=message_request
)
print(
"send_response",
send_response.model_dump_json(exclude_none=True, indent=2),
)
if not isinstance(send_response.root, SendMessageSuccessResponse):
print("received non-success response. Aborting get task ")
return None
if not isinstance(send_response.root.result, Task):
print("received non-task response. Aborting get task ")
return None
return send_response.root.result
...
في الطلب، نقدّم لوكيل خدمة التسوق جميع أسماء الوكلاء عن بُعد وأوصافهم المتاحة، وفي الأداة self.send_task، نقدّم آلية لاسترداد العميل المناسب للتواصل مع الوكيل وإرسال البيانات الوصفية المطلوبة باستخدام العنصر SendMessageRequest.
بروتوكولات الاتصال
تعريف المهمة هو نطاق يملكه خادم A2A. ومع ذلك، من منظور عميل A2A، يرى ذلك على أنّه رسالة يتم إرسالها إلى الخادم، ويعود إلى الخادم تحديد كيفية تعريف الرسائل الواردة من العميل على أنّها مهمة وما إذا كان إكمال المهمة يتطلّب تفاعلاً من العميل. يمكنك الاطّلاع على مزيد من التفاصيل حول دورة حياة المهمة في هذا المستند. يمكنك الاطّلاع أدناه على المفهوم الأعلى مستوى لهذا الإجراء:


يتم تنفيذ عملية تبادل الرسائل والمهام هذه باستخدام تنسيق البيانات الأساسية فوق معيار JSON-RPC كما هو موضّح في المثال التالي لبروتوكول message/send :
{
# identifier for this request
"id": "abc123",
# version of JSON-RPC protocol
"jsonrpc": "2.0",
# method name
"method": "message/send",
# parameters/arguments of the method
"params": {
"message": "hi, what can you help me with?"
}
}
تتوفّر طرق مختلفة، مثلاً لدعم أنواع مختلفة من التواصل (مثل المزامنة أو البث أو عدم التزامن) أو لضبط الإشعارات بشأن حالة المهمة. يمكن ضبط خادم A2A بمرونة للتعامل مع معايير تعريف المهام هذه. يمكنك الاطّلاع على تفاصيل هذه الطرق في هذا المستند.
11. 🎯 التحدي
الآن، هل يمكنك إعداد الملف اللازم ونشر تطبيق Gradio في Cloud Run بنفسك؟ حان الوقت لخوض التحدي.
12. 🧹 تنظيف
لتجنُّب تحمّل رسوم في حسابك على Google Cloud مقابل الموارد المستخدَمة في هذا الدرس التطبيقي حول الترميز، اتّبِع الخطوات التالية:
- في Google Cloud Console، انتقِل إلى صفحة إدارة الموارد.
- في قائمة المشاريع، اختَر المشروع الذي تريد حذفه، ثم انقر على حذف.
- في مربّع الحوار، اكتب رقم تعريف المشروع، ثم انقر على إيقاف لحذف المشروع.
- بدلاً من ذلك، يمكنك الانتقال إلى Cloud Run وAgent Engine في وحدة التحكّم، واختيار الخدمة التي نشرتها للتو وحذفها.