1. מבוא
פרוטוקול Agent-to-agent (A2A) נועד לתקנן את התקשורת בין סוכני AI, במיוחד בין סוכנים שנפרסים במערכות חיצוניות. בעבר, פרוטוקולים כאלה הוגדרו עבור כלים שנקראים Model Context Protocol (MCP). זהו תקן חדש לחיבור של מודלים גדולים של שפה (LLM) לנתונים ולמשאבים. A2A מנסה להשלים את MCP. בעוד ש-MCP מתמקד בהפחתת המורכבות של חיבור סוכנים לכלים ולנתונים, A2A מתמקד בהפעלת שיתוף פעולה בין סוכנים בשיטות הטבעיות שלהם. הוא מאפשר לנציגים לתקשר כנציגים (או כמשתמשים) ולא ככלים. לדוגמה, הוא מאפשר תקשורת הלוך ושוב כשרוצים להזמין משהו.
התקשורת בין אפליקציות (A2A) נועדה להשלים את התקשורת בין תהליכים (MCP). במסמכים הרשמיים מומלץ להשתמש ב-MCP עבור כלים וב-A2A עבור סוכנים – שמיוצגים על ידי AgentCard (נרחיב על כך בהמשך). לאחר מכן, מסגרות העבודה יכולות להשתמש ב-A2A כדי לתקשר עם המשתמש, עם הסוכנים המרוחקים ועם סוכנים אחרים.
בהדגמה הזו נתחיל בהטמעה של A2A באמצעות python SDK. נבחן תרחיש שימוש שבו יש לנו עוזר אישי לקניות שיכול לעזור לנו לתקשר עם סוכנים של מוכרי המבורגרים ופיצות כדי לטפל בהזמנה שלנו.
ב-A2A נעשה שימוש בעקרון הלקוח-שרת. זהו התהליך האופייני של A2A שצפוי בהדגמה הזו
- לקוח A2A יבצע קודם גילוי בכל כרטיס סוכן של שרת A2A שאפשר לגשת אליו, וישתמש במידע שלו כדי ליצור לקוח חיבור.
- כשנדרש, A2A Client ישלח הודעה אל A2A Server, והשרת יעריך את ההודעה הזו כמשימה שצריך להשלים. אם כתובת ה-URL של מקבל ההתראות בדחיפה מוגדרת בלקוח A2A ונתמכת על ידי שרת A2A, השרת יוכל גם לפרסם את מצב התקדמות המשימה בנקודת הקצה המקבלת בלקוח
- אחרי שהמשימה מסתיימת, שרת A2A שולח את ארטיפקט התגובה ללקוח A2A
במהלך ה-codelab, תשתמשו בגישה שלב אחר שלב באופן הבא:
- הכנת הפרויקט ב-Google Cloud והפעלת כל ממשקי ה-API הנדרשים בו
- הגדרת סביבת עבודה לסביבת הקידוד
- הכנת משתני סביבה לשירותי סוכנים של המבורגר ופיצה וניסיון מקומי
- פריסת סוכן המבורגר ופיצה ב-Cloud Run
- בדיקת הפרטים של אופן ההגדרה של שרת A2A
- הכנת משתני סביבה לקונסיירז' לרכישה וניסיון מקומי
- פריסת Concierge לרכישות ב-Agent Engine
- התחברות למנוע הסוכן דרך ממשק מקומי
- בדיקת הפרטים לגבי האופן שבו לקוח A2A נוצר והמודלים של הנתונים שלו
- בדיקת מטען הייעודי (payload) והאינטראקציה בין לקוח A2A לשרת
סקירה כללית של הארכיטקטורה
נפרוס את ארכיטקטורת השירות הבאה
נפרוס 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. לפני שמתחילים
בחירת פרויקט פעיל ב-Cloud Console
ב-codelab הזה אנחנו יוצאים מנקודת הנחה שכבר יש לכם פרויקט ב-Google Cloud עם חיוב מופעל. אם עדיין אין לכם חשבון, אתם יכולים לפעול לפי ההוראות שבהמשך כדי להתחיל.
- ב-Google Cloud Console, בדף לבחירת הפרויקט, בוחרים או יוצרים פרויקט ב-Google Cloud.
- מוודאים שהחיוב מופעל בפרויקט ב-Cloud. כך בודקים אם החיוב מופעל בפרויקט
הגדרת פרויקט בענן בטרמינל Cloud Shell
- תשתמשו ב-Cloud Shell, סביבת שורת פקודה שפועלת ב-Google Cloud ומגיעה עם bq שנטען מראש. לוחצים על 'הפעלת Cloud Shell' בחלק העליון של מסוף Google Cloud. אם מוצגת בקשה לאישור, לוחצים על אישור.
- אחרי שמתחברים ל-Cloud Shell, בודקים שכבר בוצע אימות ושהפרויקט מוגדר למזהה הפרויקט שלכם באמצעות הפקודה הבאה:
gcloud auth list
- מריצים את הפקודה הבאה ב-Cloud Shell כדי לוודא שפקודת gcloud מכירה את הפרויקט.
gcloud config list project
- אם הפרויקט לא מוגדר, משתמשים בפקודה הבאה כדי להגדיר אותו:
gcloud config set project <YOUR_PROJECT_ID>
אפשר גם לראות את המזהה במסוף:PROJECT_ID
לוחצים עליו וכל הפרויקטים ומזהה הפרויקט יופיעו בצד שמאל.
- מפעילים את ממשקי ה-API הנדרשים באמצעות הפקודה שמוצגת למטה. זה יימשך כמה דקות, אז כדאי לחכות בסבלנות.
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com
אם הפקודה תפעל בהצלחה, תוצג הודעה שדומה לזו שמופיעה בהמשך:
Operation "operations/..." finished successfully.
אפשר גם לחפש כל מוצר במסוף או להשתמש בקישור הזה במקום בפקודת gcloud.
אם פספסתם API כלשהו, תמיד תוכלו להפעיל אותו במהלך ההטמעה.
אפשר לעיין במאמרי העזרה בנושא פקודות gcloud ושימוש בהן.
כניסה אל Cloud Shell Editor והגדרת ספריית עבודה של האפליקציה
עכשיו אפשר להגדיר את עורך הקוד כדי לבצע פעולות שקשורות לקוד. נשתמש ב-Cloud Shell Editor לצורך הזה
- לוחצים על הלחצן Open Editor (פתיחת העורך) כדי לפתוח את Cloud Shell Editor. כאן אפשר לכתוב את הקוד
- מוודאים שהפרויקט ב-Cloud Code מוגדר בפינה הימנית התחתונה (סרגל הסטטוס) של עורך Cloud Shell, כפי שמודגש בתמונה שלמטה, ושהוא מוגדר לפרויקט הפעיל ב-Google Cloud שבו מופעל החיוב. אם מתבקשים, לוחצים על אישור. אם כבר ביצעתם את הפקודה הקודמת, יכול להיות שהלחצן יפנה ישירות לפרויקט שהפעלתם במקום ללחצן הכניסה.
- בשלב הבא, משכפלים את ספריית העבודה של התבנית בשביל ה-codelab הזה מ-GitHub, ומריצים את הפקודה הבאה. היא תיצור את ספריית העבודה בספרייה purchasing-concierge-a2a
git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a
- אחרי זה, עוברים לקטע העליון של Cloud Shell Editor ולוחצים על File->Open Folder (קובץ > פתיחת תיקייה), מוצאים את ספריית שם המשתמש ואת הספרייה purchasing-concierge-a2a ואז לוחצים על הלחצן OK. הספרייה שנבחרה תוגדר כספריית העבודה הראשית. בדוגמה הזו, שם המשתמש הוא alvinprayuda, ולכן נתיב הספרייה מוצג למטה
עכשיו Cloud Shell Editor אמור להיראות כך
הגדרת הסביבה
השלב הבא הוא הכנת סביבת הפיתוח. הטרמינל הפעיל הנוכחי צריך להיות בתוך ספריית העבודה purchasing-concierge-a2a. ב-codelab הזה נשתמש ב-Python 3.12 וב-uv python project manager כדי לפשט את הצורך ביצירה ובניהול של גרסת Python וסביבה וירטואלית.
- אם עדיין לא פתחתם את הטרמינל, פותחים אותו על ידי לחיצה על Terminal (טרמינל) -> New Terminal (טרמינל חדש), או על ידי הקשה על Ctrl + Shift + C. חלון הטרמינל ייפתח בחלק התחתון של הדפדפן.
- עכשיו נאחל את הסביבה הווירטואלית של עוזר הקניות באמצעות
uv(כבר מותקן מראש בטרמינל בענן). מריצים את הפקודה הזו
uv sync --frozen
תיקיית .venv תיצור ותתקין את יחסי התלות. תצוגה מקדימה מהירה של pyproject.toml תיתן לכם מידע על התלות שמוצגת כך
dependencies = [
"a2a-sdk>=0.2.16",
"google-adk>=1.8.0",
"gradio>=5.38.2",
]
- כדי לבדוק את הסביבה הווירטואלית, יוצרים קובץ חדש בשם main.py ומעתיקים את הקוד הבא
def main():
print("Hello from purchasing-concierge-a2a!")
if __name__ == "__main__":
main()
- לאחר מכן, מריצים את הפקודה הבאה
uv run main.py
יוצג פלט כמו בדוגמה הבאה
Using CPython 3.12 Creating virtual environment at: .venv Hello from purchasing-concierge-a2a!
ההודעה הזו מציינת שהגדרת פרויקט Python מתבצעת בצורה תקינה.
עכשיו אפשר לעבור לשלב הבא, הגדרה ופריסה של סוכן המכירות מרחוק
3. פריסת סוכן מוכר מרוחק – שרת A2A ב-Cloud Run
בשלב הזה נבצע פריסה של שני סוכני המכירות מרחוק שמסומנים בתיבה האדומה. הסוכן להזמנת המבורגר יופעל על ידי מסגרת הסוכנים CrewAI, והסוכן להזמנת פיצה יופעל על ידי סוכן Langgraph. שניהם יופעלו על ידי מודל Gemini Flash 2.0
פריסת סוכן המבורגר מרוחק
קוד המקור של סוכן ההמבורגר נמצא בספרייה remote_seller_agents/burger_agent. אפשר לבדוק את האתחול של הנציג בסקריפט agent.py. זהו קטע הקוד של הסוכן שאותחל
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
...
כל הקבצים שנמצאים בספרייה 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. זה קורה רק אם אף פעם לא פרסתם ל-Cloud Run ממקור לפני כן. אחרי פריסה מוצלחת, יומן האירועים ייראה כך.
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 שנפרס.
אם הפריסה בוצעה בהצלחה, התגובה שמוצגת בדפדפן כשניגשים אל https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json תיראה כך :
אלה פרטי הכרטיס של סוכן ההמבורגרים שאמורים להיות זמינים למטרות גילוי. נרחיב על הנושא הזה בהמשך. שימו לב שהערך 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 כבר יוגדר כמו שצריך
פריסת סוכן הפיצה המרוחק
באופן דומה, קוד המקור של סוכן הפיצה נמצא בספרייה remote_seller_agents/pizza_agent. אפשר לבדוק את האתחול של הנציג בסקריפט agent.py. זהו קטע הקוד של הסוכן שאותחל
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,
)
...
בדומה לשלב הקודם של פריסת סוכן ה-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. עכשיו נדון ברכיבי הליבה של שרת A2A
4. רכיבי הליבה של שרת 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 בשלב הקודם.
תור משימות וסוכן לביצוע
שרת 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 לגישה לכרטיס הסוכן, וגם נקודת קצה של POST לתמיכה בפרוטוקול A2A
סיכום
בקיצור, עד עכשיו פרסנו שרת A2A באמצעות Python SDK, שיכול לתמוך בשתי הפונקציות הבאות:
- פרסום כרטיס הנציג במסלול
/.well-known/agent.json - טיפול בבקשת JSON-RPC באמצעות תור משימות בזיכרון
אפשר לבדוק את נקודת הכניסה להפעלת הפונקציות האלה בסקריפט __main__.py ( ב-remote_seller_agents/burger_agent או ב-remote_seller_agents/pizza_agent) .
5. פריסת Concierge לרכישה – לקוח A2A ב-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,
],
)
...
אנחנו נפעיל את הסוכן הזה במנוע הסוכנים. Vertex AI Agent Engine הוא קבוצה של שירותים שמאפשרים למפתחים לפרוס סוכני AI, לנהל אותם ולבצע להם התאמה לעומס (scaling) בסביבת ייצור. הוא מטפל בתשתית להרחבת סוכנים בסביבת ייצור, כדי שנוכל להתמקד ביצירת אפליקציות. מידע נוסף זמין במסמך הזה . אם בעבר היינו צריכים להכין קבצים שנדרשים לפריסת שירות הסוכן שלנו (כמו סקריפט השרת main ו-Dockerfile), במקרה הזה אנחנו יכולים לפרוס את הסוכן שלנו ישירות מסקריפט Python בלי לפתח שירות קצה עורפי משלנו, באמצעות שילוב של ADK ו-Agent Engine. כדי לפרוס את התוסף :
- קודם צריך ליצור את האחסון הזמני ב-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שמוצג בהמשך
"""
Copyright 2025 Google LLC
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
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 ואספקת משתני הסביבה הנדרשים.
כדי לפרוס אותו, מריצים את הסקריפט:
uv run deploy_to_agent_engine.py
אחרי פריסה מוצלחת, יומן האירועים ייראה כך. שימו לב: xxxx הוא מזהה הפרויקט ו-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
אפשר לקיים אינטראקציה עם מנוע הסוכן באמצעות פקודת curl ו-SDK. לדוגמה, מריצים את הפקודה הבאה כדי לנסות אינטראקציה עם הסוכן שנפרס.
אפשר לנסות לשלוח את השאילתה הזו כדי לבדוק אם הסוכן נפרס בהצלחה
curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://us-central1-aiplatform.googleapis.com/v1/projects/{YOUR_PROJECT_ID}/locations/us-central1/reasoningEngines/{YOUR_AGENT_ENGINE_RESOURCE_ID}:streamQuery?alt=sse -d '{
"class_method": "stream_query",
"input": {
"user_id": "user_123",
"message": "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
6. רכיבי הליבה של לקוח 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 בצורה גמישה כדי לטפל בתקנים האלה של הגדרות משימות. פרטים על השיטות האלה מופיעים במסמך הזה.
7. בדיקת שילוב ובדיקת מטען ייעודי (payload)
עכשיו נבדוק את עוזר הקנייה שלנו באמצעות אינטראקציה עם נציג מרחוק דרך ממשק משתמש באינטרנט.
קודם צריך לעדכן את 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
לאחר מכן, מריצים את הפקודה הבאה כדי לפרוס אפליקציית Gradio
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 ועכשיו נראה איך היא מיושמת כארכיטקטורת לקוח ושרת
8. האתגר
עכשיו, תוכל להכין את הקובץ הנדרש ולפרוס את אפליקציית Gradio בענן בעצמך? הגיע הזמן לקחת חלק באתגר!
9. הסרת המשאבים
כדי לא לצבור חיובים לחשבון Google Cloud על המשאבים שבהם השתמשתם ב-Code Lab הזה:
- במסוף Google Cloud, עוברים לדף Manage resources.
- ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על Delete.
- כדי למחוק את הפרויקט, כותבים את מזהה הפרויקט בתיבת הדו-שיח ולוחצים על Shut down.
- לחלופין, אפשר לעבור אל Cloud Run במסוף, לבחור את השירות שפרסתם ולמחוק אותו.