1. 📖 مقدمة
تم تصميم بروتوكول Agent2Agent (A2A) لتوحيد معايير التواصل بين وكلاء الذكاء الاصطناعي، لا سيما أولئك الذين يتم نشرهم في أنظمة خارجية. في السابق، تم وضع بروتوكولات مماثلة للأدوات تحت اسم بروتوكول سياق النموذج (MCP)، وهو معيار ناشئ لربط النماذج اللغوية الكبيرة بالبيانات والموارد. تحاول A2A إكمال MCP، حيث تركّز A2A على مشكلة مختلفة، بينما تركّز MCP على تقليل التعقيد لربط الوكلاء بالأدوات والبيانات، وتركز A2A على كيفية تمكين الوكلاء من التعاون في أساليبهم الطبيعية. تتيح هذه الميزة للوكلاء التواصل كوكلاء (أو كمستخدمين) بدلاً من التواصل كأدوات، على سبيل المثال، تتيح إجراء محادثة ذهابًا وإيابًا عندما تريد طلب شيء ما.
تم تصميم A2A لتكمّل MCP، وفي المستندات الرسمية، يُنصح بأن تستخدم التطبيقات MCP للأدوات وA2A للوكلاء، ويتم تمثيلهم من خلال AgentCard ( سنتناول هذا الموضوع لاحقًا). ويمكن للأُطر بعد ذلك استخدام A2A للتواصل مع المستخدمين والوكلاء البعيدين والوكلاء الآخرين.
في هذا العرض التوضيحي، سنبدأ بتنفيذ ميزة "التطبيق بالتطبيق" باستخدام حزمة تطوير البرامج (SDK) الخاصة بلغة Python. سنتناول حالة استخدام عندما يكون لدينا مساعد شخصي للتسوّق يمكنه مساعدتنا في التواصل مع وكلاء بائعي البرغر والبيتزا للتعامل مع طلبنا.
تستفيد A2A من مبدأ العميل والخادم. في ما يلي مسار A2A النموذجي الذي ستتوقّعه في هذا الدليل التعليمي
- سيجري برنامج A2A Client أولاً عملية البحث عن جميع بطاقات وكيل خادم A2A التي يمكن الوصول إليها، وسيستخدم معلوماتها لإنشاء برنامج اتصال.
- عند الحاجة، سيرسل تطبيق A2A Client رسالة إلى خادم A2A، وسيقيّم الخادم هذه الرسالة على أنّها مهمة يجب إكمالها. إذا تم ضبط عنوان URL الخاص بمستلِم الإشعارات الفورية على عميل A2A وكان متوافقًا مع خادم A2A، سيتمكّن الخادم أيضًا من نشر حالة تقدّم المهمة إلى نقطة النهاية المستلِمة على العميل.
- بعد انتهاء المهمة، سيرسل خادم A2A عنصر الردّ إلى عميل A2A.
خلال هذا الدرس العملي، ستتّبع نهجًا خطوة بخطوة على النحو التالي:
- إعداد مشروع Google Cloud
- إعداد دليل العمل لبيئة الترميز
- نشر وكيل البرغر على Cloud Run
- نشر وكيل البيتزا على Cloud Run
- نشر "مرشد التسوّق" في Agent Engine
- التفاعل مع خدمة الاستشارة بشأن الشراء من خلال الواجهة المحلية
نظرة عامة على البنية
ستنشر بنية الخدمة التالية
ستنشر خدمتين تعملان كخادم A2A، وهما وكيل Burger ( الذي يستند إلى إطار عمل وكيل CrewAI) ووكيل Pizza ( الذي يستند إلى إطار عمل وكيل Langgraph). سيتفاعل المستخدم مباشرةً مع خدمة Purchasing concierge فقط التي سيتم تشغيلها باستخدام إطار عمل Agent Development Kit (ADK) الذي سيعمل كعميل A2A.
سيكون لكلّ من هؤلاء الوكلاء بيئة وعملية نشر خاصة به.
المتطلبات الأساسية
- إجادة العمل باستخدام لغة Python
- فهم بنية أساسية كاملة الميزات باستخدام خدمة HTTP
ما ستتعلمه
- البنية الأساسية لخادم A2A
- البنية الأساسية لعميل A2A
- نشر خدمة الوكيل على Cloud Run
- تفعيل خدمة الوكيل في "محرك الوكيل"
- كيفية ربط تطبيق 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 Server على 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 لخدمات وكيل Burger التي تم نشرها من خلال المتصفّح. هذا هو عنوان URL للوصول إلى بطاقة وكيل خادم A2A التي تم نشرها.
في حال تم نشرها بنجاح، ستظهر لك الاستجابة كما هو موضّح أدناه في المتصفّح عند الوصول إلى بطاقة الوكيل
هذه هي معلومات بطاقة وكيل البرغر التي يجب أن تكون متاحة لأغراض الاكتشاف.
لاحظ أنّ قيمة url لا تزال مضبوطة على http://0.0.0.0:8080/ هنا. يجب أن تكون قيمة url هذه هي المعلومات الرئيسية التي يستخدمها تطبيق A2A Client لإرسال الرسائل من العالم الخارجي، ولكن لم يتم ضبطها بشكل صحيح.
علينا تعديل هذه القيمة إلى عنوان URL الخاص بخدمة وكيل البرغر من خلال إضافة متغيّر بيئة إضافي HOST_OVERRIDE.
تعديل قيمة عنوان URL الخاص بـ Burger Agent على بطاقة Agent Card من خلال متغيّر البيئة
لإضافة 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 الخاص بـ "وكيل البيتزا" على بطاقة الوكيل من خلال متغيّر البيئة
لإضافة 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. 🚀 نشر تطبيق Purchasing Concierge - A2A Client to Agent Engine
في هذه الخطوة، سننفّذ وكيل خدمة التسوق. هذا هو الوكيل الذي سنتفاعل معه.
يتوفّر رمز المصدر الخاص بوكيل خدمة التسوّق في الدليل 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. 🚀 اختبار الدمج وفحص الحمولة
لنلقِ نظرة الآن على خدمة المساعدة في الشراء من خلال التفاعل مع الوكيل عن بُعد باستخدام واجهة مستخدم على الويب. نفِّذ الأمر التالي لنشر تطبيق 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. يهدف ذلك إلى دعم مرحلة الاستكشاف على تطبيق "العميل من التطبيق إلى التطبيق"، والتي من المفترض أن تقدّم معلومات وسياقات كاملة حول كيفية الوصول إلى الوكيل ومعرفة جميع إمكاناته. الأمر مشابه إلى حدّ ما مع توفّر وثائق واجهة برمجة التطبيقات بشكلٍ جيد باستخدام 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) الخاصة بلغة Python في A2A. راجِع مقتطف الرمز البرمجي 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- 💡 [شرح الرمز] نشر "محرك الوكيل"
في ما يلي مقتطف الرمز البرمجي الخاص بوكيل خدمة الاستقبال والإرشاد في 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 وتنفيذه
الصورة المعروضة أعلاه هي التدفق النموذجي لتفاعلات التطبيق مع تطبيق آخر:
- سيحاول العميل العثور على أي بطاقة وكيل منشورة في عنوان URL الخاص بالوكيل البعيد المقدَّم في المسار
/.well-known/agent.json - بعد ذلك، سيرسل الخادم رسالة إلى هذا الوكيل عند الضرورة تتضمّن الرسالة ومعلمات البيانات الوصفية اللازمة ( مثل معرّف الجلسة والسياق السابق وما إلى ذلك). وسيتعامل الخادم مع هذه الرسالة على أنّها مهمة يجب إكمالها.
- يعالج خادم A2A الطلب، وإذا كان الخادم يتيح إرسال الإشعارات الفورية، سيكون بإمكانه أيضًا نشر بعض الإشعارات أثناء معالجة المهمة ( هذه الوظيفة خارج نطاق هذا الدرس العملي).
- بعد الانتهاء، سيرسل خادم A2A عنصر الردّ إلى العميل.
بعض العناصر الأساسية للتفاعلات المذكورة أعلاه هي ما يلي (يمكنك الاطّلاع على مزيد من التفاصيل هنا) :
- الرسالة: هي تبادل للمعلومات بين العميل وموظف الدعم عن بُعد
- المهمة: هي وحدة العمل الأساسية التي تديرها A2A، ويتم تحديدها من خلال معرّف فريد.
- الناتج: هو نتيجة (مثل مستند أو صورة أو بيانات منظَّمة) ينشئها الوكيل نتيجةً لمهمة، ويتألف من أجزاء.
- الجزء: هو أصغر وحدة محتوى ضمن "رسالة" أو "عنصر". يمكن أن يكون الجزء نصًا أو صورة أو فيديو أو ملفًا أو غير ذلك.
استكشاف البطاقات
عند إعداد خدمة A2A Client، تتمثل العملية النموذجية في محاولة الحصول على معلومات بطاقة الوكيل وتخزينها لتسهيل الوصول إليها عند الحاجة. في هذا الدرس العملي، ننفّذها على 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 في وحدة التحكّم، واختيار الخدمة التي نشرتها للتو وحذفها.