1. 📖 מבוא
פרוטוקול Agent2Agent (A2A) נועד לתקנן את התקשורת בין סוכני AI, במיוחד בין סוכנים שמוצבים במערכות חיצוניות. בעבר, פרוטוקולים כאלה הוגדרו עבור כלים שנקראים Model Context Protocol (MCP). זהו תקן חדש לחיבור של מודלים גדולים של שפה (LLM) לנתונים ולמשאבים. A2A מנסה להשלים את MCP. בעוד ש-MCP מתמקד בהפחתת המורכבות של חיבור סוכנים לכלים ולנתונים, A2A מתמקד בהפעלת שיתוף פעולה בין סוכנים בשיטות הטבעיות שלהם. הוא מאפשר לנציגים לתקשר כנציגים (או כמשתמשים) ולא ככלים. לדוגמה, הוא מאפשר תקשורת הלוך ושוב כשרוצים להזמין משהו.
התקשורת בין אפליקציות (A2A) נועדה להשלים את הפרוטוקול של מרכז הבקרה של המנהל (MCP). במסמכים הרשמיים מומלץ להשתמש ב-MCP עבור כלים וב-A2A עבור סוכנים – שמיוצגים על ידי AgentCard (נרחיב על כך בהמשך). לאחר מכן, מסגרות העבודה יכולות להשתמש ב-A2A כדי לתקשר עם המשתמש, עם הסוכנים המרוחקים ועם סוכנים אחרים.
בהדגמה הזו נתחיל בהטמעה של A2A באמצעות python SDK. נבחן תרחיש שימוש שבו יש לנו עוזר אישי לקניות שיכול לעזור לנו לתקשר עם סוכנים של מוכרי המבורגרים ופיצות כדי לטפל בהזמנה שלנו.
ב-A2A נעשה שימוש בעקרון הלקוח-שרת. זהו התהליך הטיפוסי של העברת נתונים מאפליקציה לאפליקציה שצפוי במדריך הזה
- לקוח A2A יבצע קודם גילוי בכל כרטיס סוכן של שרת A2A שאפשר לגשת אליו, וישתמש במידע שלו כדי ליצור לקוח חיבור.
- כשנדרש, A2A Client ישלח הודעה אל A2A Server, והשרת יעריך את ההודעה הזו כמשימה שצריך להשלים. אם כתובת ה-URL של מקבל ההתראות בדחיפה מוגדרת בלקוח A2A ונתמכת על ידי שרת A2A, השרת יוכל גם לפרסם את מצב התקדמות המשימה בנקודת הקצה המקבלת בלקוח
- אחרי שהמשימה מסתיימת, שרת A2A שולח את ארטיפקט התגובה ללקוח A2A
במהלך ה-codelab, תשתמשו בגישה שלב אחר שלב באופן הבא:
- הכנת פרויקט Google Cloud
- הגדרת ספרייה בעבודה לסביבת קידוד
- פריסת סוכן המבורגר ב-Cloud Run
- פריסת סוכן הפיצה ב-Cloud Run
- פריסת Concierge לרכישות ב-Agent Engine
- אינטראקציה עם שירות הקונסיירז' לרכישות דרך ממשק מקומי
סקירה כללית של הארכיטקטורה
תפרסו את ארכיטקטורת השירות הבאה
תפרסו 2 שירותים שיפעלו כשרת A2A: סוכן המבורגרים ( שמבוסס על מסגרת סוכנים של CrewAI) וסוכן פיצה ( שמבוסס על מסגרת סוכנים של Langgraph). המשתמש יקיים אינטראקציה ישירה רק עם שירות הקונסיירז' לרכישות, שיפעל באמצעות מסגרת Agent Development Kit (ADK) וישמש כלקוח A2A.
לכל אחד מהסוכנים האלה יהיו סביבה ופריסה משלו.
דרישות מוקדמות
- נוח לעבוד עם Python
- הבנה של ארכיטקטורת full-stack בסיסית באמצעות שירות HTTP
מה תלמדו
- המבנה המרכזי של שרת A2A
- המבנה המרכזי של לקוח A2A
- פריסת שירות סוכן ב-Cloud Run
- פריסת סוכן שירות ב-Agent Engine
- איך לקוח A2A מתחבר לשרת A2A
- מבנה הבקשה והתגובה בחיבור לא סטרימינג
מה צריך
- דפדפן האינטרנט Chrome
- חשבון Gmail
- פרויקט ב-Cloud שמופעל בו חשבון לחיוב
ב-codelab הזה, שמיועד למפתחים בכל הרמות (כולל מתחילים), נעשה שימוש ב-Python באפליקציה לדוגמה. עם זאת, לא נדרש ידע ב-Python כדי להבין את המושגים שמוצגים.
2. 🚀 הכנה להגדרת סדנת פיתוח
שלב 1: בוחרים פרויקט פעיל ב-Cloud Console
במסוף Google Cloud, בדף לבחירת הפרויקט, בוחרים או יוצרים פרויקט ב-Google Cloud (ראו את הקטע הימני העליון במסוף).
לוחצים עליו ורואים רשימה של כל הפרויקטים, כמו בדוגמה הזו:
הערך שמסומן בתיבה האדומה הוא מזהה הפרויקט, והוא ישמש לאורך כל המדריך.
הקפידו לוודא שהחיוב מופעל בפרויקט שלכם ב-Cloud. כדי לבדוק את זה, לוחצים על סמל ההמבורגר ☰ בסרגל הימני העליון כדי להציג את תפריט הניווט, ומחפשים את תפריט החיוב.
אם מופיעה ההודעה "Google Cloud Platform Trial Billing Account" is linked, הפרויקט מוכן לשימוש במדריך הזה. אם לא, חוזרים לתחילת המדריך הזה ומממשים את חשבון החיוב
שלב 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>
לאחר מכן, משכפלים את ספריית העבודה של התבנית בשביל ה-codelab הזה מ-GitHub, ומריצים את הפקודה הבאה. היא תיצור את ספריית העבודה בספרייה purchasing-concierge-a2a
git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a
שלב 3: היכרות עם Cloud Shell Editor והגדרת ספריית עבודה של האפליקציה
עכשיו אפשר להגדיר את עורך הקוד כדי לבצע פעולות שקשורות לקוד. נשתמש ב-Cloud Shell Editor לצורך הזה
לוחצים על הלחצן Open Editor כדי לפתוח את Cloud Shell Editor 
.
אחרי זה, עוברים לחלק העליון של Cloud Shell Editor ולוחצים על File->Open Folder (קובץ > פתיחת תיקייה), מוצאים את ספריית שם המשתמש ואת הספרייה purchasing-concierge-a2a ואז לוחצים על הלחצן OK. הפעולה הזו תגדיר את הספרייה שנבחרה כספריית העבודה הראשית. בדוגמה הזו, שם המשתמש הוא alvinprayuda, ולכן נתיב הספרייה מוצג למטה
עכשיו Cloud Shell Editor אמור להיראות כך
עכשיו פותחים את הטרמינל של כלי העריכה. כדי לעשות את זה, לוחצים על Terminal -> New Terminal בסרגל התפריטים , או משתמשים בקיצור הדרך Ctrl + Shift + C. חלון טרמינל ייפתח בחלק התחתון של הדפדפן.
הטרמינל הפעיל הנוכחי צריך להיות בתוך ספריית העבודה purchasing-concierge-a2a. ב-codelab הזה נשתמש ב-Python 3.12 וב-uv python project manager כדי לפשט את הצורך ביצירה ובניהול של גרסת Python וסביבה וירטואלית. חבילת uv כבר מותקנת מראש ב-Cloud Shell.
מריצים את הפקודה הזו כדי להתקין את התלות הנדרשת בסביבה הווירטואלית בספרייה .venv
uv sync --frozen
כדי לראות את התלויות שהוגדרו במדריך הזה, שהן a2a-sdk, google-adk, and gradio, בודקים את הקובץ pyproject.toml.
עכשיו צריך להפעיל את ממשקי ה-API הנדרשים באמצעות הפקודה שמוצגת למטה. הפעולה עשויה להימשך זמן מה.
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 כדי לשלוח הודעות מהעולם החיצוני, אבל הוא לא מוגדר בצורה נכונה.
אנחנו צריכים לעדכן את הערך הזה לכתובת ה-URL של שירות הסוכן שלנו על ידי הוספת משתנה סביבה נוסף HOST_OVERRIDE.
עדכון הערך של כתובת ה-URL של סוכן המבורגר בכרטיס הסוכן באמצעות משתנה סביבה
כדי להוסיף את HOST_OVERRIDE לשירות של סוכן המבורגרים, פועלים לפי השלבים הבאים
- מחפשים את Cloud Run בסרגל החיפוש בחלק העליון של מסוף הענן.
- לוחצים על שירות Cloud Run burger-agent שנפרס קודם.
- מעתיקים את כתובת ה-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.
בדומה לשלב הקודם של פריסת סוכן ה-burger, כל הקבצים שנמצאים בספרייה 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 בסרגל החיפוש בחלק העליון של מסוף הענן.
- לוחצים על שירות Cloud Run pizza-agent שנפרס בעבר.
- לוחצים על עריכה ופריסה של גרסה חדשה.
- מעתיקים את כתובת ה-URL של שירות הפיצה ולוחצים על הקטע Variable & Secrets (משתנה וסודות).
- אחר כך לוחצים על הוספת משתנה ומגדירים את
HOST_OVERRIDEהערך לכתובת ה-URL של השירות ( הכתובת עם התבניתhttps://pizza-agent-xxxxxxxxx.us-central1.run.app)
- לבסוף, לוחצים על לחצן הפריסה כדי לפרוס מחדש את השירות
עכשיו, כשתיגשו שוב לכרטיס של סוכן הפיצה בדפדפן https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json, הערך url כבר יוגדר כמו שצריך
בשלב הזה, כבר פרסנו בהצלחה את שירותי ההמבורגר והפיצה ב-Cloud Run.
6. 🚀 פריסת Concierge לרכישה – לקוח 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
אפשר לבדוק את הסקריפט ולראות שאנחנו מנסים לשאול את הסוכן "List available burger menu please"
אם הפעולה בוצעה ללא שגיאות, יוצגו כמה אירועי תגובה שמוזרמים במסוף, כמו בדוגמה הבאה
{
"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. 🚀 בדיקות שילוב ובדיקת מטען ייעודי (payload)
עכשיו נבדוק את עוזר הקנייה שלנו באמצעות אינטראקציה עם נציג מרחוק דרך ממשק משתמש באינטרנט. מריצים את הפקודה הבאה כדי לפרוס אפליקציית 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 + click על כתובת ה-URL http://0.0.0.0:8080 בטרמינל או לוחצים על לחצן התצוגה המקדימה של האינטרנט כדי לפתוח את ממשק המשתמש של האינטרנט.
אפשר לנסות לנהל שיחה כזו :
- תראה לי תפריט של המבורגר ופיצה
- אני רוצה להזמין פיצה עם עוף ברביקיו והמבורגר קייג'ון חריף
וממשיכים את השיחה עד לסיום ההזמנה. בדקו איך האינטראקציה מתנהלת ומהי קריאת הכלי והתשובה? בתמונה הבאה אפשר לראות דוגמה לתוצאת האינטראקציה.
אנחנו רואים ששיחה עם שני נציגים שונים מובילה לשני סוגים שונים של התנהגויות, והתכונה 'סוכן לנציג' יכולה להתמודד עם זה בצורה טובה. הסוכן של מוכר הפיצה מקבל ישירות את הבקשה שלנו מסוכן הקניות, אבל הסוכן של מוכר ההמבורגרים צריך את האישור שלנו לפני שהוא ממשיך עם הבקשה. אחרי שאנחנו מאשרים, הסוכן יכול להסתמך על האישור ולעבור לסוכן של מוכר ההמבורגרים.
סיימנו להסביר את המושגים הבסיסיים של 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, שבו אמורים לקבל מידע מלא והקשרים לגבי הגישה לסוכן ולכל היכולות שלו. זה די דומה לתיעוד API מפורט באמצעות Swagger או Postman.
This is the content of our deployed burger agent agent card
{
"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 כדי לתת לנציג של הלקוח הקשר לגבי היכולות והכישורים של הנציג המרוחק שאפשר להפעיל. שדות מפורטים יותר של כרטיס הנציג הזה מופיעים במאמר הזה.
בקוד שלנו, ההטמעה של כרטיס הנציג מתבצעת באמצעות A2A python sdk. אפשר לראות את ההטמעה בקטע הקוד 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 בשלב הקודם.
Task Queue ו-Agent Executor
שרת 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())
...
בדוגמה של הקוד שלמעלה, אנחנו מטמיעים סכמת עיבוד בסיסית שבה הסוכן יופעל ישירות כשהבקשה תתקבל, וישלח אירועים של משימות שהושלמו אחרי שההפעלה תסתיים. עם זאת, לא הטמענו כאן את שיטת הביטול כי היא נחשבת לפעולה קצרת טווח.
אחרי שיוצרים את ה-executor, אפשר להשתמש ישירות ב-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 route כדי לגשת לכרטיס של הסוכן, וגם את נקודת הקצה POST כדי לתמוך בפרוטוקול A2A
סיכום
בקיצור, עד עכשיו פרסנו שרת A2A באמצעות Python SDK, שיכול לתמוך בשתי הפונקציות הבאות:
- פרסום כרטיס הסוכן במסלול
/.well-known/agent.json - טיפול בבקשת JSON-RPC באמצעות תור משימות בזיכרון
אפשר לבדוק את נקודת הכניסה להפעלת הפונקציות האלה בסקריפט __main__.py ( ב-remote_seller_agents/burger_agent או ב-remote_seller_agents/pizza_agent) .
9. 💡 [הסבר על הקוד] פריסת Agent Engine
קטע הקוד של סוכן הקנייה האישי ב-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 הוא קבוצה של שירותים שמאפשרים למפתחים לפרוס סוכני AI, לנהל אותם ולבצע להם התאמה לעומס (scaling) בסביבת ייצור. הוא מטפל בתשתית להרחבת סוכנים בסביבת ייצור, כדי שנוכל להתמקד ביצירת אפליקציות. מידע נוסף זמין במסמך הזה . אם בעבר היינו צריכים להכין קבצים כדי לפרוס את שירות הסוכן שלנו (כמו סקריפט השרת 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 ( אפשר גם לציין כאן ספריות וקבצים אחרים אם צריך) וציון משתני הסביבה הנדרשים.
10. 💡 [הסבר על הקוד] מושג הלקוח של A2A וההטמעה שלו
התמונה שלמעלה מציגה את התהליך האופייני של אינטראקציות A2A:
- הלקוח ינסה למצוא כרטיס של סוכן שפורסם בכתובת ה-URL של הסוכן המרוחק שצוינה במסלול
/.well-known/agent.json - לאחר מכן, כשיהיה צורך, הוא ישלח סוכנות הודעה עם ההודעה ופרמטרים של מטא-נתונים נדרשים ( למשל, מזהה סשן, הקשר היסטורי וכו'). השרת יתייחס להודעה הזו כמשימה שצריך להשלים.
- תהליך השרת A2A מעבד את הבקשה. אם השרת תומך בהתראות Push, הוא יוכל גם לפרסם התראות מסוימות במהלך עיבוד המשימה ( הפונקציונליות הזו לא נכללת ב-codelab הזה).
- אחרי שהפעולה מסתיימת, שרת A2A שולח את ארטיפקט התגובה בחזרה ללקוח
חלק מהאובייקטים העיקריים של האינטראקציות שלמעלה הם הפריטים הבאים (מידע נוסף זמין כאן) :
- הודעה: תור תקשורת בין לקוח לבין נציג מרוחק
- משימה: היחידה הבסיסית של עבודה שמנוהלת על ידי A2A, שמזוהה באמצעות מזהה ייחודי
- ארטיפקט: פלט (למשל, מסמך, תמונה, נתונים מובְנים) שנוצר על ידי הסוכן כתוצאה ממשימה, ומורכב מחלקים
- חלק: היחידה הקטנה ביותר של תוכן בהודעה או בארטיפקט. החלק יכול להיות טקסט, תמונה, סרטון, קובץ וכו'.
Card Discovery
כשמפעילים את שירות הלקוח A2A, התהליך הרגיל הוא לנסות לקבל את פרטי כרטיס הנציג ולאחסן אותם כדי שיהיה קל לגשת אליהם כשצריך. ב-codelab הזה אנחנו מטמיעים אותו ב-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)
...
בשלב הזה, אנחנו מנסים לגשת לכל כרטיסי הנציגים הזמינים באמצעות מודול הלקוח המובנה של A2A A2ACardResolver, ואז אוספים את הנתונים שנדרשים כדי לשלוח הודעה לנציג. לאחר מכן, אנחנו צריכים גם לפרט את כל הנציגים הזמינים ואת המפרטים שלהם בהנחיה, כדי שהנציג שלנו יידע שהוא יכול לתקשר עם הנציגים האלה.
הכלי 'הנחיה ושליחת משימה'
זו ההנחיה והכלי שאנחנו מספקים לסוכן 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.
פרוטוקולי התקשורת
ההגדרה Task היא דומיין שנמצא בבעלות של שרת A2A. עם זאת, מנקודת המבט של לקוח A2A, הוא רואה את זה כהודעה שנשלחת לשרת. השרת הוא זה שמגדיר איזו משימה מתבצעת כשמתקבלות הודעות מהלקוח, והאם השלמת המשימה דורשת אינטראקציה מהלקוח. אפשר לקרוא פרטים נוספים על מחזור החיים של משימה במסמכי התיעוד האלה. הנה תרשים שממחיש את הרעיון הכללי:
ההחלפה הזו של הודעה -> משימה מיושמת באמצעות פורמט המטען הייעודי (payload) על בסיס תקן 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 בענן בעצמך? הגיע הזמן לקחת חלק באתגר!
12. 🧹 ניקוי
כדי לא לצבור חיובים לחשבון Google Cloud על המשאבים שבהם השתמשתם ב-Code Lab הזה:
- במסוף Google Cloud, עוברים לדף Manage resources.
- ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על Delete.
- כדי למחוק את הפרויקט, כותבים את מזהה הפרויקט בתיבת הדו-שיח ולוחצים על Shut down.
- לחלופין, אפשר לעבור אל Cloud Run ואל Agent Engine במסוף, לבחור את השירות שפרסתם ולמחוק אותו.