Questa pagina è stata tradotta dall'API Cloud Translation.

Guida introduttiva al protocollo Agent-to-Agent (A2A): interazioni tra un concierge per gli acquisti e un agente venditore remoto con Gemini su Cloud Run e Agent Engine

1. Introduzione

Il protocollo da agente ad agente (A2A) è progettato per standardizzare la comunicazione tra gli agenti AI, in particolare per quelli implementati in sistemi esterni. In precedenza, questi protocolli erano stati stabiliti per gli strumenti chiamati Model Context Protocol (MCP), uno standard emergente per connettere gli LLM a dati e risorse. A2A cerca di integrare MCP, in quanto si concentra su un problema diverso. Mentre MCP si concentra sulla riduzione della complessità per connettere gli agenti a strumenti e dati, A2A si concentra su come consentire agli agenti di collaborare nelle loro modalità naturali. Consente agli agenti di comunicare come agenti (o come utenti) anziché come strumenti; ad esempio, consente la comunicazione bidirezionale quando vuoi ordinare qualcosa.

A2A è posizionata in modo da integrare MCP. Nella documentazione ufficiale è consigliato che le applicazioni utilizzino MCP per gli strumenti e A2A per gli agenti, rappresentati da AgentCard ( ne parleremo più avanti). I framework possono quindi utilizzare A2A per comunicare con l'utente, gli agenti remoti e altri agenti.

In questa demo, inizieremo con l'implementazione di A2A utilizzando il relativo SDK Python. Esploreremo un caso d'uso in cui abbiamo un concierge per gli acquisti personali che può aiutarci a comunicare con gli agenti di vendita di hamburger e pizza per gestire il nostro ordine.

A2A utilizza il principio client-server. Ecco il flusso A2A tipico che ci aspettiamo in questa demo

Il client A2A esegue innanzitutto il rilevamento di tutte le schede dell'agente del server A2A accessibili e utilizza le relative informazioni per creare un client di connessione.
Se necessario, A2A Client invierà un messaggio ad A2A Server, che lo valuterà come un'attività da completare. Se l'URL del destinatario delle notifiche push è configurato sul client A2A e supportato dal server A2A, il server sarà anche in grado di pubblicare lo stato di avanzamento dell'attività nell'endpoint di ricezione sul client
Al termine dell'attività, il server A2A invierà l'artefatto di risposta al client A2A

Nel codelab, seguirai un approccio passo passo come segue:

Prepara il progetto Google Cloud e abilita tutte le API richieste
Configurare lo spazio di lavoro per l'ambiente di programmazione
Prepara le variabili di ambiente per i servizi di agenti di hamburger e pizza e prova localmente
Esegui il deployment dell'agente per hamburger e pizza in Cloud Run
Esamina i dettagli su come è stato stabilito il server A2A
Prepara le variabili di ambiente per l'assistente agli acquisti e provale in locale
Esegui il deployment del concierge per gli acquisti in Agent Engine
Connettiti al motore dell'agente tramite l'interfaccia locale
Esamina i dettagli su come è stato stabilito il client A2A e la relativa modellazione dei dati
Ispezionare il payload e l'interazione tra il client A2A e il server

Panoramica dell'architettura

Verrà eseguito il deployment della seguente architettura di servizio

Implementeremo due servizi che fungeranno da server A2A: l'agente Burger ( basato sul framework dell'agente CrewAI) e l'agente Pizza ( basato sul framework dell'agente Langgraph). L'utente interagirà direttamente solo con il concierge degli acquisti, che verrà eseguito utilizzando il framework Agent Development Kit (ADK) che fungerà da client A2A.

Ciascuno di questi agenti avrà il proprio ambiente e il proprio deployment.

Prerequisiti

Avere familiarità con Python
Conoscenza dell'architettura full-stack di base che utilizza il servizio HTTP

Cosa imparerai a fare

Struttura principale di A2A Server
Struttura principale del client A2A
Deployment del servizio agente in Cloud Run
Deployment del servizio agente in Agent Engine
Come si connette il client A2A al server A2A
Struttura di richiesta e risposta su connessione non in streaming

Che cosa ti serve

Browser web Chrome
Un account Gmail
Un progetto cloud con fatturazione abilitata

Questo codelab, progettato per sviluppatori di tutti i livelli (inclusi i principianti), utilizza Python nella sua applicazione di esempio. Tuttavia, la conoscenza di Python non è necessaria per comprendere i concetti presentati.

2. Prima di iniziare

Seleziona il progetto attivo in Cloud Console

Questo codelab presuppone che tu disponga già di un progetto Google Cloud con la fatturazione abilitata. Se non l'hai ancora fatto, puoi seguire le istruzioni riportate di seguito per iniziare.

Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud.
Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata per un progetto.

Configura il progetto cloud nel terminale Cloud Shell

Utilizzerai Cloud Shell, un ambiente a riga di comando in esecuzione in Google Cloud precaricato con bq. Fai clic su Attiva Cloud Shell nella parte superiore della console Google Cloud. Se ti viene chiesto di autorizzare, fai clic su Autorizza.

Una volta eseguita la connessione a Cloud Shell, verifica di essere già autenticato e che il progetto sia impostato sul tuo ID progetto utilizzando il seguente comando:

gcloud auth list

Esegui questo comando in Cloud Shell per verificare che il comando gcloud conosca il tuo progetto.

gcloud config list project

Se il progetto non è impostato, utilizza il seguente comando per impostarlo:

gcloud config set project <YOUR_PROJECT_ID>

In alternativa, puoi anche visualizzare l'ID PROJECT_ID nella console.

Fai clic e vedrai tutti i tuoi progetti e l'ID progetto sul lato destro.

Abilita le API richieste tramite il comando mostrato di seguito. L'operazione potrebbe richiedere alcuni minuti.

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

Se il comando viene eseguito correttamente, dovresti visualizzare un messaggio simile a quello mostrato di seguito:

Operation "operations/..." finished successfully.

L'alternativa al comando gcloud è tramite la console, cercando ogni prodotto o utilizzando questo link.

Se manca un'API, puoi sempre abilitarla durante l'implementazione.

Consulta la documentazione per i comandi e l'utilizzo di gcloud.

Vai all'editor di Cloud Shell e configura la directory di lavoro dell'applicazione

Ora possiamo configurare l'editor di codice per svolgere alcune attività di programmazione. Per questo utilizzeremo l'editor di Cloud Shell

Fai clic sul pulsante Apri editor per aprire un editor di Cloud Shell in cui puoi scrivere il codice
Assicurati che il progetto Cloud Code sia impostato nell'angolo in basso a sinistra (barra di stato) dell'editor di Cloud Shell, come evidenziato nell'immagine seguente, e che sia impostato sul progetto Google Cloud attivo in cui hai attivato la fatturazione. Autorizza, se richiesto. Se segui già il comando precedente, il pulsante potrebbe anche puntare direttamente al progetto attivato anziché al pulsante di accesso.

Successivamente, cloniamo la directory di lavoro del modello per questo codelab da GitHub eseguendo il seguente comando. Verrà creata la directory di lavoro nella directory purchasing-concierge-a2a

git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a

Dopodiché, vai alla sezione superiore dell'editor di Cloud Shell e fai clic su File->Apri cartella, trova la directory username e la directory purchasing-concierge-a2a, poi fai clic sul pulsante OK. In questo modo, la directory scelta diventerà la directory di lavoro principale. In questo esempio, il nome utente è alvinprayuda, quindi il percorso della directory è mostrato di seguito

Ora, l'editor di Cloud Shell dovrebbe avere il seguente aspetto

Configurazione dell'ambiente

Il passaggio successivo consiste nel preparare l'ambiente di sviluppo. Il terminale attivo corrente deve trovarsi all'interno della directory di lavoro purchasing-concierge-a2a. In questo codelab utilizzeremo Python 3.12 e uv python project manager per semplificare la necessità di creare e gestire la versione di Python e l'ambiente virtuale.

Se non hai ancora aperto il terminale, aprilo facendo clic su Terminale -> Nuovo terminale o utilizza Ctrl + Maiusc + C, che aprirà una finestra del terminale nella parte inferiore del browser.

Ora inizializziamo l'ambiente virtuale del concierge degli acquisti utilizzando uv (già preinstallato sul terminale cloud). Esegui questo comando

uv sync --frozen

Verranno create la directory .venv e installate le dipendenze. Una rapida occhiata al file pyproject.toml ti fornirà informazioni sulle dipendenze visualizzate in questo modo

dependencies = [
    "a2a-sdk>=0.2.16",
    "google-adk>=1.8.0",
    "gradio>=5.38.2",
]

Per testare l'ambiente virtuale, crea un nuovo file main.py e copia il seguente codice

def main():
   print("Hello from purchasing-concierge-a2a!")

if __name__ == "__main__":
   main()

Quindi, esegui questo comando

uv run main.py

Otterrai un output simile a quello mostrato di seguito.

Using CPython 3.12
Creating virtual environment at: .venv
Hello from purchasing-concierge-a2a!

Ciò dimostra che il progetto Python è in fase di configurazione corretta.

Ora possiamo passare al passaggio successivo, ovvero configurare e implementare l'agente venditore remoto.

3. Deployment di Remote Seller Agent - A2A Server su Cloud Run

In questo passaggio, eseguiremo il deployment di questi due agenti venditori remoti contrassegnati dalla casella rossa. L'agente per gli hamburger sarà basato sul framework di agenti CrewAI e l'agente per la pizza sarà basato sull'agente Langgraph. Entrambi sono basati sul modello Gemini Flash 2.0.

Deployment dell'agente Remote Burger

Il codice sorgente dell'agente burger si trova nella directory remote_seller_agents/burger_agent. L'inizializzazione dell'agente può essere esaminata nello script agent.py. Ecco lo snippet di codice dell'agente inizializzato

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

...

Tutti i file presenti nella directory remote_seller_agents/burger_agent sono già sufficienti per eseguire il deployment del nostro agente in Cloud Run in modo che sia accessibile come servizio. Ne parleremo più avanti. Esegui questo comando per eseguire il deployment

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}

Se ti viene chiesto se creare un repository di container per il deployment dall'origine, rispondi Y. Ciò si verificava solo se non avevi mai eseguito il deployment su Cloud Run da un'origine in precedenza. Una volta eseguito il deployment, verrà visualizzato un log simile a questo.

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

La parte xxxx qui sarà un identificatore univoco quando implementeremo il servizio. Ora proviamo a seguire il percorso https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json di questi servizi di agenti di ristoranti di cui è stato eseguito il deployment tramite browser. Questo è l'URL per accedere alla scheda dell'agente server A2A di cui è stato eseguito il deployment.

Se il deployment è stato eseguito correttamente, quando accedi a https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json nel browser, vedrai la risposta mostrata di seguito :

Queste sono le informazioni della scheda dell'agente del burger che devono essere accessibili a scopo di rilevamento. Ne parleremo più avanti. Tieni presente che il valore url è ancora impostato su http://0.0.0.0:8080/. Questo valore url dovrebbe essere l'informazione principale per il client A2A per inviare messaggi dal mondo esterno e non è configurato correttamente. Per questa demo, dovremo aggiornare questo valore all'URL del nostro servizio di agenti di hamburger aggiungendo una variabile di ambiente aggiuntiva HOST_OVERRIDE.

Aggiornamento del valore dell'URL dell'agente Burger sulla scheda dell'agente tramite la variabile di ambiente

Per aggiungere HOST_OVERRIDE al servizio di agenti di burger, segui questi passaggi

Cerca Cloud Run nella barra di ricerca nella parte superiore della console cloud

Fai clic sul servizio Cloud Run burger-agent di cui è stato eseguito il deployment in precedenza.

Copia l'URL del servizio burger, quindi fai clic su Modifica ed esegui il deployment di una nuova revisione.

Poi, fai clic sulla sezione Variabili e secret.

Dopodiché, fai clic su Aggiungi variabile e imposta HOST_OVERRIDE come valore dell'URL del servizio ( quello con il pattern https://burger-agent-xxxxxxxxx.us-central1.run.app).

Infine, fai clic sul pulsante Esegui il deployment per eseguire nuovamente il deployment del servizio.

Ora, quando accedi di nuovo alla scheda dell'agente burger-agent nel browser https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json , il valore url sarà già configurato correttamente.

Deployment dell'agente per la pizza da remoto

Allo stesso modo, il codice sorgente dell'agente per la pizza si trova nella directory remote_seller_agents/pizza_agent. L'inizializzazione dell'agente può essere esaminata nello script agent.py. Ecco lo snippet di codice dell'agente inizializzato

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,
)

...

Analogamente al passaggio di deployment dell'agente burger precedente, tutti i file presenti nella directory remote_seller_agents/pizza_agent sono già sufficienti per eseguire il deployment dell'agente in Cloud Run in modo che sia accessibile come servizio. Esegui questo comando per eseguire il deployment

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}

Una volta eseguito il deployment, verrà visualizzato un log simile a questo.

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

La parte xxxx qui sarà un identificatore univoco quando implementeremo il servizio. Lo stesso vale per l'agente per gli hamburger: quando provi ad accedere alla route https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json di questi servizi dell'agente per la pizza tramite browser per accedere alla scheda dell'agente del server A2A, il valore dell'agente per la pizza url nella relativa scheda non è ancora configurato correttamente. Dobbiamo anche aggiungere HOST_OVERRIDE alla variabile di ambiente

Aggiornamento del valore dell'URL dell'agente per la pizza nella scheda dell'agente tramite la variabile di ambiente

Per aggiungere HOST_OVERRIDE al servizio dell'agente per la pizza, segui questi passaggi

Cerca Cloud Run nella barra di ricerca nella parte superiore della console cloud

Fai clic sul servizio Cloud Run pizza-agent di cui è stato eseguito il deployment in precedenza.

Fai clic su Modifica ed esegui il deployment di una nuova revisione.

Copia l'URL del servizio di consegna di pizze, poi fai clic sulla sezione Variabili e secret.

Dopodiché, fai clic su Aggiungi variabile e imposta HOST_OVERRIDE come valore dell'URL del servizio ( quello con il pattern https://pizza-agent-xxxxxxxxx.us-central1.run.app).

Infine, fai clic sul pulsante Esegui il deployment per eseguire nuovamente il deployment del servizio.

Ora, quando accedi di nuovo alla scheda dell'agente pizza-agent nel browser https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json, il valore url sarà già configurato correttamente.

A questo punto, abbiamo già eseguito il deployment dei servizi di hamburger e pizza su Cloud Run. Ora parliamo dei componenti principali del server A2A

4. Componenti principali del server A2A

Ora parliamo del concetto e dei componenti principali del server A2A.

Scheda dell'agente

Ogni server A2A deve avere una scheda agente accessibile nella risorsa /.well-known/agent.json. per supportare la fase di scoperta sul client A2A, che dovrebbe fornire informazioni e contesti completi su come accedere all'agente e conoscerne tutte le funzionalità. È un po' come avere una documentazione API ben documentata che utilizza Swagger o Postman.

Questi sono i contenuti della scheda dell'agente burger che abbiamo implementato

{
  "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"
}

Queste schede dell'agente mettono in evidenza molti componenti importanti, come le competenze dell'agente, le funzionalità di streaming, le modalità supportate, la versione del protocollo e altro ancora.

Tutte queste informazioni possono essere utilizzate per sviluppare un meccanismo di comunicazione adeguato in modo che il client A2A possa comunicare correttamente. La modalità e il meccanismo di autenticazione supportati garantiscono che la comunicazione possa essere stabilita correttamente e che le informazioni dell'agente skills possano essere incorporate nel prompt del sistema client A2A per fornire all'agente del cliente il contesto delle capacità e delle competenze dell'agente remoto da richiamare. I campi più dettagliati per questa scheda dell'agente sono disponibili in questa documentazione.

Nel nostro codice, l'implementazione della scheda dell'agente viene stabilita utilizzando l'SDK Python A2A. Per l'implementazione, consulta lo snippet remote_seller_agents/burger_agent/main.py riportato di seguito.

...

        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],
        )

...

Possiamo vedere diversi campi, ad esempio:

AgentCapabilities : dichiarazione di funzioni opzionali aggiuntive supportate dal servizio di agente,come la possibilità di streaming e/o il supporto delle notifiche push
AgentSkill : strumenti o funzioni supportati dall'agente
Input/OutputModes : modalità di tipo Input/Output supportata
Url : indirizzo per comunicare con l'agente

In questa configurazione forniamo una creazione dinamica dell'URL host dell'agente, in modo che sia più facile passare dai test locali al deployment sul cloud. Per questo motivo, nel passaggio precedente è necessario aggiungere la variabile HOST_OVERRIDE.

Coda di attività e Agent Executor

Il server A2A potrebbe gestire richieste di diversi agenti o utenti ed essere in grado di isolare perfettamente ogni attività. Per visualizzare meglio i contesti di questi, puoi esaminare l'immagine seguente.

Pertanto, ogni server A2A deve essere in grado di monitorare le attività in entrata e archiviare le informazioni corrette. L'SDK A2A fornisce moduli per affrontare questa sfida nel server A2A. Innanzitutto, possiamo istanziare la logica su come vogliamo gestire la richiesta in arrivo. Ereditando la classe astratta AgentExecutor, possiamo controllare come gestire l'esecuzione e l'annullamento delle attività. Questa implementazione di esempio può essere esaminata nel modulo remote_seller_agents/burger_agent/agent_executor.py ( percorso simile per il caso del venditore di pizza).

...

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())

...

Nel codice riportato sopra, implementiamo uno schema di elaborazione di base in cui l'agente viene richiamato direttamente quando la richiesta è in arrivo e invia gli eventi di attività completata dopo aver terminato la chiamata. Tuttavia, non abbiamo implementato il metodo di annullamento in questo caso perché è stato considerato un'operazione di breve durata.

Dopo aver creato l'executor, possiamo utilizzare direttamente DefaultRequestHandler, InMemoryTaskStore e A2AStarletteApplication integrati per avviare il server HTTP. Questa implementazione può essere esaminata in 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)

...

Questo modulo ti fornirà l'implementazione della route /.well-known/agent.json per accedere alla scheda dell'agente e all'endpoint POST per supportare il protocollo A2A

Riepilogo

In breve, finora il nostro server A2A di cui è stato eseguito il deployment utilizza l'SDK Python in grado di supportare le due funzionalità riportate di seguito:

Pubblicazione della scheda dell'agente sull'itinerario /.well-known/agent.json
Gestire la richiesta JSON-RPC con la gestione in coda delle attività in memoria

Il punto di ingresso per l'avvio di queste funzionalità può essere esaminato nello script __main__.py ( su remote_seller_agents/burger_agent o remote_seller_agents/pizza_agent) .

5. Deployment di Purchasing Concierge - A2A Client to Agent Engine

In questo passaggio, eseguiremo il deployment dell'agente concierge per gli acquisti. Questo è l'agente con cui interagiremo.

Il codice sorgente del nostro agente concierge per gli acquisti si trova nella directory purchasing_concierge. L'inizializzazione dell'agente può essere esaminata nello script purchasing_agent.py. Ecco lo snippet di codice dell'agente inizializzato.

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,
            ],
        )

...

Eseguiremo il deployment di questo agente nel motore degli agenti. Vertex AI Agent Engine è un insieme di servizi che consente agli sviluppatori di distribuire, gestire e scalare gli agenti AI in produzione. Gestisce l'infrastruttura per scalare gli agenti in produzione, così possiamo concentrarci sulla creazione di applicazioni. Per saperne di più, consulta questo documento . Se in precedenza dovevamo preparare i file necessari per il deployment del nostro servizio di agenti (come lo script del server main e Dockerfile), in questo caso possiamo eseguire il deployment del nostro agente direttamente dallo script Python senza la necessità di sviluppare il nostro servizio di backend utilizzando una combinazione di ADK e Agent Engine. Per implementarlo, segui questi passaggi :

Innanzitutto, dobbiamo creare l'archiviazione di staging in Cloud Storage

gcloud storage buckets create gs://purchasing-concierge-{your-project-id} --location=us-central1

Ora dobbiamo preparare prima la variabile .env, quindi copiamo .env.example nel file .env

cp .env.example .env

Ora apri il file .env e vedrai i seguenti contenuti

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}

Questo agente comunicherà con l'agente di hamburger e pizza, pertanto dobbiamo fornire le credenziali appropriate per entrambi. Dovremo aggiornare PIZZA_SELLER_AGENT_URL e BURGER_SELLER_AGENT_URL con l'URL Cloud Run dei passaggi precedenti.

Se te ne dimentichi, visita la console Cloud Run. Digita "Cloud Run" nella barra di ricerca nella parte superiore della console e fai clic con il tasto destro del mouse sull'icona di Cloud Run per aprirla in una nuova scheda.

Dovresti visualizzare i nostri precedenti servizi di agenti di vendita remota di cui è stato eseguito il deployment, come mostrato di seguito

Ora, per visualizzare l'URL pubblico di questi servizi, fai clic su uno dei servizi e verrà visualizzata la pagina Dettagli servizio. Puoi visualizzare l'URL nella parte superiore, accanto alle informazioni sulla regione.

La variabile di ambiente finale dovrebbe essere simile a questa

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}

Ora siamo pronti per eseguire il deployment del nostro agente di acquisto concierge. In questa demo verrà eseguito il deployment utilizzando lo script deploy_to_agent_engine.py i cui contenuti sono riportati di seguito

"""
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}")

Questi sono i passaggi necessari per eseguire il deployment dell'agente ADK nel motore dell'agente. Innanzitutto, dobbiamo creare un oggetto AdkApp dal nostro ADK root_agent. A questo punto possiamo eseguire il metodo agent_engines.create fornendo l'oggetto adk_app, specificando i requisiti nel campo requirements, specificando il percorso della directory dell'agente in extra_packages e fornendo le variabili di ambiente necessarie.

Possiamo eseguirne il deployment eseguendo lo script:

uv run deploy_to_agent_engine.py

Una volta eseguito il deployment, verrà visualizzato un log simile a questo. Tieni presente che xxxx è l'ID progetto e yyyy è l'ID risorsa del motore dell'agente.

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

Quando lo ispezioniamo nella dashboard di Agent Engine (cerca "Agent Engine" nella barra di ricerca), viene visualizzato il deployment precedente.

Test dell'agente di cui è stato eseguito il deployment su Agent Engine

L'interazione con il motore dell'agente può essere eseguita tramite il comando curl e l'SDK. Ad esempio, esegui il seguente comando per provare a interagire con l'agente di cui è stato eseguito il deployment.

Puoi provare a inviare questa query per verificare se l'agente è stato implementato correttamente

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",
  }
}'

In caso di esito positivo, nella console verranno visualizzati diversi eventi di risposta in streaming, come questo

{
  "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
}

Nel passaggio successivo proveremo a utilizzare la UI, ma prima vediamo quali sono i componenti principali e il flusso tipico dei client A2A

6. Componenti principali del client A2A

L'immagine mostrata sopra è il flusso tipico delle interazioni A2A:

Il client tenterà di trovare una scheda dell'agente pubblicata nell'URL dell'agente remoto fornito nella route /.well-known/agent.json
Poi, quando necessario, invierà un messaggio all'agente con il messaggio e i parametri dei metadati necessari ( ad es.ID sessione, contesto storico e così via). Il server percepirà questo messaggio come un'attività da completare.
Il server A2A elabora la richiesta. Se il server supporta le notifiche push, sarà anche in grado di pubblicare alcune notifiche durante l'elaborazione dell'attività ( questa funzionalità non rientra nell'ambito di questo codelab).
Al termine, il server A2A invierà l'artefatto di risposta al client

Alcuni degli oggetti principali per le interazioni precedenti sono questi elementi (per ulteriori dettagli, fai clic qui) :

Messaggio:un turno di comunicazione tra un cliente e un agente remoto
Attività: l'unità di lavoro fondamentale gestita da A2A, identificata da un ID univoco
Artefatto:un output (ad es. un documento, un'immagine, dati strutturati) generato dall'agente come risultato di un'attività, composto da parti
Parte:l'unità di contenuti più piccola all'interno di un messaggio o di un artefatto. Una parte può essere un testo, un'immagine, un video, un file e così via.

Card Discovery

Quando viene avviato il servizio A2A Client, la procedura tipica consiste nel tentare di ottenere i dati della scheda dell'agente e archiviarli per accedervi facilmente quando necessario. In questo codelab, lo implementiamo su before_agent_callback. Puoi vedere l'implementazione in purchasing_concierge/purchasing_agent.py. Vedi lo snippet di codice di seguito.

...

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)

...

Qui tentiamo di accedere a tutte le schede degli agenti disponibili utilizzando il modulo client A2A A2ACardResolver integrato, quindi raccogliamo la connessione necessaria per inviare messaggi all'agente. Dopodiché, dobbiamo anche elencare tutti gli agenti disponibili e le relative specifiche nel prompt in modo che il nostro agente sappia di poter comunicare con questi agenti.

Strumento Richiedi e invia attività

Questo è il prompt e lo strumento che forniamo al nostro agente ADK qui

...

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

...

Nel prompt, forniamo al nostro agente di concierge per gli acquisti il nome e la descrizione di tutti gli agenti remoti disponibili e nello strumento self.send_task forniamo un meccanismo per recuperare il cliente appropriato a cui connettersi all'agente e inviare i metadati richiesti utilizzando l'oggetto SendMessageRequest.

Protocolli di comunicazione

La definizione di Task è un dominio di proprietà del server A2A. Tuttavia, dal punto di vista del client A2A, viene visualizzato come un messaggio inviato al server. Spetta al server definire i messaggi in arrivo dal client come attività e se il completamento dell'attività richiede l'interazione del client. Puoi leggere ulteriori dettagli sul ciclo di vita dell'attività in questa documentazione. Il concetto di livello superiore può essere visualizzato di seguito:

Questo scambio di messaggio -> attività viene implementato utilizzando il formato del payload in aggiunta allo standard JSON-RPC, come mostrato nell'esempio seguente del protocollo 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?"
  }  
}

Sono disponibili vari metodi, ad esempio per supportare diversi tipi di comunicazione (ad es. sincronizzazione, streaming, asincrona) o per configurare le notifiche per lo stato dell'attività. Il server A2A può essere configurato in modo flessibile per gestire questi standard di definizione delle attività. I dettagli di questi metodi sono disponibili in questo documento.

7. Test di integrazione e ispezione del payload

Ora esaminiamo il nostro concierge per gli acquisti con l'interazione dell'agente remoto utilizzando un'interfaccia web.

Innanzitutto, dobbiamo aggiornare AGENT_ENGINE_RESOURCE_NAME in .env file. Assicurati di fornire il nome della risorsa del motore dell'agente corretto. Il file .env dovrebbe avere il seguente aspetto:

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

Dopodiché, esegui questo comando per eseguire il deployment di un'app Gradio

uv run purchasing_concierge_ui.py

Se l'operazione va a buon fine, viene visualizzato il seguente output.

* Running on local URL:  http://0.0.0.0:8080
* To create a public link, set `share=True` in `launch()`.

Poi, Ctrl + fai clic sull'URL http://0.0.0.0:8080 nel terminale o fai clic sul pulsante di anteprima web per aprire la UI web.

Prova a fare una conversazione come questa :

Mostrami il menu di hamburger e pizza
Vorrei ordinare una pizza con pollo al barbecue e un hamburger cajun piccante.

e continua la conversazione finché non completi l'ordine. Controlla come procede l'interazione e qual è la chiamata allo strumento e la risposta. L'immagine seguente è un esempio del risultato dell'interazione.

Possiamo notare che la comunicazione con due agenti diversi produce due comportamenti diversi e A2A può gestirlo bene. L'agente del venditore di pizza accetta direttamente la nostra richiesta di agente di acquisto, mentre l'agente di hamburger ha bisogno della nostra conferma prima di procedere con la nostra richiesta e, dopo la nostra conferma, può fare affidamento su di essa per l'agente di hamburger

Ora abbiamo terminato i concetti di base di A2A e vediamo come viene implementato come architettura client-server

8. La sfida

Ora puoi preparare il file necessario ed eseguire il deployment dell'app Gradio in Cloud Run autonomamente. È ora di accettare la sfida.

9. Esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo codelab, segui questi passaggi:

Nella console Google Cloud, vai alla pagina Gestisci risorse.
Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare, quindi fai clic su Elimina.
Nella finestra di dialogo, digita l'ID progetto, quindi fai clic su Chiudi per eliminare il progetto.
In alternativa, puoi andare a Cloud Run nella console, selezionare il servizio di cui hai appena eseguito il deployment ed eliminarlo.