Agent Development Kit (ADK) di Google: guida con progetto demo

Panoramica del progetto: pianificatore di viaggi con IA con ADK e A2A

Questo progetto crea un pianificatore di viaggi che:

• Prende in input destinazione, date e budget.
• Chiama tre agent separati:
• flight_agent:  Consiglia opzioni di volo.
• stay_agent: Trova hotel entro il budget.
• activities_agent: Suggerisce attività coinvolgenti.
• Poi usa un host_agent centrale per orchestrare tutte le richieste.
• Infine, usa una UI Streamlit per l’interazione con l’utente.

Tutti gli agent sono ospitati come server FastAPI separati ed espongono un endpoint /run. La comunicazione avviene tramite il client compatibile A2A condiviso.

Nota: questo progetto gira interamente sulla tua macchina locale per semplicità, ma puoi distribuire facilmente ciascun agent e la UI su piattaforme cloud come Render, Railway o Google Cloud Run per un accesso scalabile.

Step 1: Prerequisiti

Iniziamo installando le seguenti librerie:

pip install google-adk litellm fastapi uvicorn httpx pydantic openai streamlit

Poi configura la tua chiave API di OpenAI—puoi usare anche un altro provider di modelli. Per imparare a configurare la chiave API di OpenAI, ti consiglio questo tutorial introduttivo sulla GPT-4o API.

export OPENAI_API_KEY="your_key_here"

Step 2: Schema e utility condivise

Prima di poter costruire agent intelligenti, dobbiamo definire un linguaggio comune per farli comunicare tra loro. Nel nostro setup, questo è fatto usando:

• Uno schema condiviso per l’input (definito via Pydantic)
• Un’utility REST client per chiamare gli agent
• Un wrapper REST server per standardizzare l’endpoint /run su tutti gli agent

Questi sono posizionati nelle cartelle shared/ e common/ per mantenere il codice modulare. Vediamoli uno per uno.

Creazione di un file shared/schemas.py

Definiamo uno schema TravelRequest usando Pydantic. Questo garantisce che tutti gli agent concordino sulla struttura delle richieste in arrivo, che include destinazione, date del viaggio e budget dell’utente.

from pydantic import BaseModel
class TravelRequest(BaseModel):
    destination: str
    start_date: str
    end_date: str
    budget: float

Questa classe aiuta a:

• Mantenere l’input coerente per tutti gli agent.
• Aggiungere validazione automatica con FastAPI.
• Semplificare il riuso del codice.

Creazione di un file common/a2a_client.py

Questa leggera utility async permette a qualsiasi agent (soprattutto l’host) di invocare un altro agent usando il protocollo A2A chiamando l’endpoint /run.

import httpx
async def call_agent(url, payload):
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, timeout=60.0)
        response.raise_for_status()
        return response.json()

Questa utility invia in modo asincrono una richiesta POST all’endpoint /run di un altro agent usando httpx.  Restituisce la risposta JSON parsata e solleva un errore se la richiesta fallisce.

Useremo questa utility nel nostro host agent per chiamare flight_agent, stay_agent e activities_agent.

Creazione di un file common/a2a_server.py

Invece di scrivere una route FastAPI personalizzata per ogni agent, la generalizziamo usando la funzione create_app(agent), che gestisce:

• Servire l’agent su /run
• Ricevere una richiesta di viaggio
• Restituire una risposta strutturata

from fastapi import FastAPI
import uvicorn
def create_app(agent):
    app = FastAPI()
    @app.post("/run")
    async def run(payload: dict):
        return await agent.execute(payload)
    return app

Questa utility crea un’app FastAPI con una route standard /run che delega l’esecuzione all’agent fornito.  Garantisce un’interfaccia agent‑to‑agent (A2A) coerente per tutti i servizi usando input JSON strutturato.

Insieme, questi componenti condivisi rendono il nostro sistema multi‑agent più manutenibile, riutilizzabile e allineato con la filosofia A2A di Google per una messaggistica semplice e strutturata tra agent.

Step 3: Costruire il sistema multi‑agent con ADK e A2A

Ora che abbiamo contratti e utility condivise, iniziamo a costruire i singoli agent. Per trasformarlo in un sistema davvero modulare e multi‑agent, useremo il protocollo A2A di Google—una semplice interfaccia basata su HTTP che consente agli agent di comunicare in modo coerente e interoperabile.

A2A (Agent‑to‑Agent) abilita il coordinamento plug‑and‑play tra agent, che siano funzioni Python locali o ospitati su reti. Ogni agent espone un endpoint /run con uno schema comune e agisce come un servizio.

Nella nostra demo abbiamo quattro agent:

• host_agent: orchestra tutti gli altri agent
• flight_agent: trova voli adatti
• stay_agent: suggerisce alloggi
• activities_agent: raccomanda attività locali coinvolgenti

Tutti gli agent sono strutturati in modo simile, con 3 file e una sottocartella:

agents/
├── host_agent/
│   │   ├── agent.py              # Optional if host logic is minimal
│   │   ├── task_manager.py       # Calls other agents and aggregates responses
│   │   ├── __main__.py           # Starts FastAPI app via common/a2a_server.py
│   │   └── .well-known/
│   │       └── agent.json        # A2A Agent Card metadata
├── flight_agent/
├── stay_agent/
└── activities_agent/

Ogni agent usa google.adk.agents.Agent, un wrapper di modello LiteLlm e Runner per l’esecuzione. Inizia creando i seguenti file nella cartella  activities_agent e ripeti lo stesso per flight_agent e stay_agent.

Creazione di un file agent.py

Definiamo ora la logica per il nostro activities_agent, che sarà responsabile di generare esperienze locali coinvolgenti in base all’itinerario di viaggio dell’utente. 

Step 1: Import

Iniziamo importando i moduli essenziali per configurare ed eseguire il nostro agent.

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import json

Questo agent usa componenti di Google ADK come Agent, ' Runner, ' e LiteLlm e gestisce lo stato usando InMemorySessionService. La libreria Types è usata per costruire prompt strutturati.

Step 2: Activities agent 

Ora istanziamo l’agent vero e proprio usando la classe Agent di ADK.

activities_agent = Agent(
    name="activities_agent",
    model=LiteLlm("openai/gpt-4o"),
    description="Suggests interesting activities for the user at a destination.",
    instruction=(
        "Given a destination, dates, and budget, suggest 2-3 engaging tourist or cultural activities. "
        "For each activity, provide a name, a short description, price estimate, and duration in hours. "
        "Respond in plain English. Keep it concise and well-formatted."
    )
)

Il parametro instruction definisce il system prompt che guida il comportamento dell’LLM. Anche se questo esempio usa l’inglese, puoi adattare l’istruzione per restituire JSON strutturato per un parsing più semplice.

Step 3: Gestione della sessione

Successivamente, per tenere traccia delle interazioni dell’utente, configuriamo un Runner insieme alle informazioni di sessione.

session_service = InMemorySessionService()
runner = Runner(
    agent=activities_agent,
    app_name="activities_app",
    session_service=session_service
)
USER_ID = "user_activities"
SESSION_ID = "session_activities"

Il Runner gestisce l’esecuzione dell’agent per una specifica sessione dell’app. La classe InMemorySessionService memorizza il contesto in memoria. Poi definiamo ID utente e di sessione. In produzione, questi potrebbero essere dinamici o specifici per utente. Questo garantisce l’esistenza di una sessione ADK nuova prima di inviare prompt all’LLM.

Step 4: Esecuzione della logica dell’agent

La funzione execute() gestisce le richieste in arrivo, costruisce un prompt, invoca il modello e analizza l’output.

async def execute(request):
    session_service.create_session(
        app_name="activities_app",
        user_id=USER_ID,
        session_id=SESSION_ID
    )
    prompt = (
        f"User is flying to {request['destination']} from {request['start_date']} to {request['end_date']}, "
        f"with a budget of {request['budget']}. Suggest 2-3 activities, each with name, description, price estimate, and duration. "
        f"Respond in JSON format using the key 'activities' with a list of activity objects."
    )
    message = types.Content(role="user", parts=[types.Part(text=prompt)])
    async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
        if event.is_final_response():
            response_text = event.content.parts[0].text
            try:
                parsed = json.loads(response_text)
                if "activities" in parsed and isinstance(parsed["activities"], list):
                    return {"activities": parsed["activities"]}
                else:
                    print("'activities' key missing or not a list in response JSON")
                    return {"activities": response_text}  # fallback to raw text
            except json.JSONDecodeError as e:
                print("JSON parsing failed:", e)
                print("Response content:", response_text)
                return {"activities": response_text}  # fallback to raw text

La funzione execute() costruisce dinamicamente un prompt usando i parametri della richiesta in arrivo come destinazione, date e budget. Ecco cosa succede dietro le quinte:

• Il prompt istruisce il modello a restituire un oggetto JSON strutturato contenente una lista di attività.
• Un oggetto Content viene costruito e passato al Runner di ADK, che attende in modo asincrono la risposta finale del modello usando un generatore in streaming.
• Una volta ricevuta la risposta finale, l’agent estrae l’output di testo grezzo e prova a eseguirne il parsing come JSON.
• Se il parsing riesce e la chiave prevista activities esiste, vengono restituiti i dati strutturati.
• Se la chiave manca o è malformata, il fallback è restituire il testo grezzo in modo che la UI abbia comunque un output utilizzabile.
• Questo doppio approccio garantisce una degradazione graduale quando l’LLM restituisce testo semplice invece di JSON strutturato.

Questa strategia migliora la robustezza e l’esperienza utente, specialmente quando gli output del modello variano leggermente a causa della temperatura o dell’interpretazione del prompt.

Creazione di un file task_manager.py

Dopo aver definito la logica execute() in agent.py, la colleghiamo alla configurazione del server compatibile ADK usando task_manager.py.

from .agent import execute
async def run(payload):
    return await execute(payload)

Questo file funge da sottile wrapper intorno alla funzione execute() definita prima. Rende il metodo run() disponibile ai moduli esterni, in particolare allo script del server in __main__.py.

Creazione di un file __main__.py

Il file __main__.py avvia un server FastAPI sulla porta 8003, servendo l’agent all’endpoint /run.

from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, port=8003)

Ecco cosa succede:

• La create_app() (da common/a2a_server.py) incapsula il nostro agent in un’interfaccia FastAPI standard compatibile A2A.
• Costruiamo dinamicamente un oggetto con un metodo execute() così che ADK possa invocare correttamente run().
• Questa separazione ci consente di mantenere interfacce API stateless riutilizzando la logica core dell’agent.

Creazione di un file .well-known/agent.json

Usiamo questo file JSON per descrivere l’identità e lo scopo dell’agent secondo il protocollo A2A (Agent‑to‑Agent).

{
    "name": "activity_agent",
    "description": "Agent providing activity details."
  }

Nota: Sebbene il file .well-known/agent.json non sia usato direttamente dai nostri agent in questo progetto, aderisce alla specifica A2A ed è importante per discovery, introspezione e futura compatibilità con orchestratori come LangGraph, CrewAI o il registry di agent di Google.

Logiche simili sono usate anche per flight_agent e stay_agent.

Step 4: Coordinamento con host_agent

L’host_agent funge da pianificatore centrale per la demo. L’host_agent esemplifica il pattern del controller nei sistemi multi‑agent. Separa decisione ed esecuzione, permettendo a ciascun agent a valle di concentrarsi sulla propria nicchia mentre centralizza la logica di coordinamento. Questo semplifica test e scalabilità e rispecchia l’architettura a microservizi nel mondo reale nei sistemi distribuiti.

Invia lo stesso payload a tutti e tre gli agent usando le loro API /run esposte e unisce i risultati. Aggiungiamo i seguenti file alla cartella host_agent.

Creazione di un file agent.py

Partiamo dagli import di base.

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

Questo blocco importa tutti i mattoni necessari per definire ed eseguire un agent basato su LLM usando Google ADK: la classe Agent, il wrapper LLM leggero, il Runner per la gestione dell’esecuzione e la gestione della sessione in memoria.

host_agent = Agent(
    name="host_agent",
    model=LiteLlm("openai/gpt-4o"),
    description="Coordinates travel planning by calling flight, stay, and activity agents.",
    instruction="You are the host agent responsible for orchestrating trip planning tasks. "
                "You call external agents to gather flights, stays, and activities, then return a final result."
)
session_service = InMemorySessionService()
runner = Runner(
    agent=host_agent,
    app_name="host_app",
    session_service=session_service
)
USER_ID = "user_host"
SESSION_ID = "session_host"

Il codice sopra definisce un agent ADK di livello superiore responsabile di coordinare l’intero piano di viaggio. Anche se in questa implementazione non invochiamo sub‑agent dall’LLM, il system prompt imposta il ruolo per una futura estensione in cui l’LLM potrebbe gestire tool use e meta‑reasoning.

async def execute(request):
    # Ensure session exists
    session_service.create_session(
        app_name="host_app",
        user_id=USER_ID,
        session_id=SESSION_ID
    )
    prompt = (
        f"Plan a trip to {request['destination']} from {request['start_date']} to {request['end_date']} "
        f"within a total budget of {request['budget']}. Call the flights, stays, and activities agents for results."
    )
    message = types.Content(role="user", parts=[types.Part(text=prompt)])
    async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=message):
        if event.is_final_response():
            return {"summary": event.content.parts[0].text}

Questa funzione execute() funge da punto di ingresso principale per l’LLM dell’host agent. Fa quanto segue:

• Inizializza una sessione (per supportare la memoria se necessario)
• Costruisce dinamicamente un prompt utente
• Lo invia al modello usando il metodo runner.run_async() di ADK
• Infine, attende ed estrae la risposta finale

Creazione di un file task_manager.py

Il task manager esegue la logica di orchestrazione chiamando agent remoti e gestendo l’intero workflow di pianificazione del viaggio. Per l’implementazione pratica, definiamo gli URL di servizio per ogni agent figlio. Questi endpoint rispettano il protocollo A2A /run e si aspettano uno schema JSON condiviso TravelRequest`.

from common.a2a_client import call_agent
FLIGHT_URL = "http://localhost:8001/run"
STAY_URL = "http://localhost:8002/run"
ACTIVITIES_URL = "http://localhost:8003/run"

Ora definiamo il payload.

async def run(payload):
    #Print what the host agent is sending
    print("Incoming payload:", payload)
    flights = await call_agent(FLIGHT_URL, payload)
    stay = await call_agent(STAY_URL, payload)
    activities = await call_agent(ACTIVITIES_URL, payload)
    # Log outputs
    print("flights:", flights)
    print("stay:", stay)
    print("activities:", activities)
    # Ensure all are dicts before access
    flights = flights if isinstance(flights, dict) else {}
    stay = stay if isinstance(stay, dict) else {}
    activities = activities if isinstance(activities, dict) else {}
    return {
        "flights": flights.get("flights", "No flights returned."),
        "stay": stay.get("stays", "No stay options returned."),
        "activities": activities.get("activities", "No activities found.")
    }

Questa funzione usa l’helper call_agent() per inviare il payload a ciascun servizio a valle e registra input e output per visibilità durante lo sviluppo. Questo file è sostanzialmente dove vive la vera logica di orchestrazione.

Creazione di un file __main__.py

Il file __main__.py funge da entry point per il server FastAPI che incapsula l’host agent. 

from common.a2a_server import create_app
from .task_manager import run
app = create_app(agent=type("Agent", (), {"execute": run}))
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, port=8000)

Il file principale fa quanto segue:

• Usa create_app() da common/a2a_server.py per generare un’app FastAPI con un endpoint /run standardizzato.
• Poi passa un semplice oggetto tipo agent con un metodo execute() che delega internamente alla funzione task_manager.run().
• Infine, avvia il server FastAPI usando uvicorn su una porta specificata (di solito 8000).

Questo allinea l’interfaccia dell’host agent con gli altri agent a valle, mantenendo la coerenza in tutto il sistema.

Creazione di un file .well-known/agent.json

Questo file rappresenta un classico pattern multi‑agent in cui un nodo centrale delega e compone i task. 

{
  "name": "host_agent",
  "description": "Coordinates travel planning among specialized agents."
}

Sebbene opzionale, è una buona pratica includerlo in tutte le directory degli agent, come spiegato in precedenza.

Step 5: Costruire la UI con Streamlit

Infine, costruiamo una semplice app in cui gli utenti possono inserire le proprie preferenze e ricevere un itinerario strutturato. Crea un file travel_ui.py nella root e aggiungi il seguente codice.

import streamlit as st
import requests

Importiamo librerie di base come Streamlit e requests per il supporto UI.

st.set_page_config(page_title="ADK-Powered Travel Planner", page_icon="✈️")
st.title("🌍 ADK-Powered Travel Planner")
origin = st.text_input("Where are you flying from?", placeholder="e.g., New York")
destination = st.text_input("Destination", placeholder="e.g., Paris")
start_date = st.date_input("Start Date")
end_date = st.date_input("End Date")
budget = st.number_input("Budget (in USD)", min_value=100, step=50)
if st.button("Plan My Trip ✨"):
    if not all([origin, destination, start_date, end_date, budget]):
        st.warning("Please fill in all the details.")
    else:
        payload = {
            "origin": origin,
            "destination": destination,
            "start_date": str(start_date),
            "end_date": str(end_date),
            "budget": budget
        }
        response = requests.post("http://localhost:8000/run", json=payload)
        if response.ok:
            data = response.json()
            st.subheader("✈️ Flights")
            st.markdown(data["flights"])
            st.subheader("🏨 Stays")
            st.markdown(data["stay"])
            st.subheader("🗺️ Activities")
            st.markdown(data["activities"])
        else:
            st.error("Failed to fetch travel plan. Please try again.")

L’app Streamlit fornisce una UI user‑friendly per interagire con il pianificatore di viaggi multi‑agent costruito con ADK. Ecco alcuni aspetti toccati nel codice sopra.

• Usa text_input, date_input e number_input per raccogliere origine, destinazione, date e budget.
• Al clic su “Plan My Trip”, valida l’input per assicurarsi che nessun campo sia vuoto.
• Se valido, costruisce un payload JSON e invia una richiesta POST all’host_agent su http://localhost:8000/run.
• L’host_agent invoca tutti gli agent figli (volo, soggiorno, attività), aggrega le loro risposte e restituisce un piano di viaggio unificato.
• La risposta è parsata e mostrata usando il metodo st.markdown() in intestazioni separate per voli, soggiorni e attività.
• Se il backend fallisce, viene mostrato un messaggio di errore di fallback usando st.error().

Ora esegui localmente i seguenti comandi nel terminale:

uvicorn agents.host_agent.__main__:app --port 8000 &
uvicorn agents.flight_agent.__main__:app --port 8001 &
uvicorn agents.stay_agent.__main__:app --port 8002 &      
uvicorn agents.activities_agent.__main__:app --port 8003 &
streamlit run travel_ui.py

Quando un utente clicca “Plan My Trip”, l’host agent prende il controllo, attiva gli agent e mostra i risultati nella UI:

La struttura complessiva dei file sarà simile a questa:

ADK_demo/
├── agents/
│   ├── host_agent/
│   │   ├── agent.py              
│   │   ├── task_manager.py      
│   │   ├── __main__.py         
│   │   └── .well-known/
│   │       └── agent.json   
│   │
│   ├── flight_agent/
│   │   ├── agent.py        
│   │   ├── task_manager.py    
│   │   ├── __main__.py         
│   │   └── .well-known/
│   │       └── agent.json        
│   │
│   ├── stay_agent/
│   │   ├── agent.py              
│   │   ├── task_manager.py       
│   │   ├── __main__.py           
│   │   └── .well-known/
│   │       └── agent.json     
│   │
│   └── activities_agent/
│       ├── agent.py              
│       ├── task_manager.py       
│       ├── __main__.py           
│       └── .well-known/
│           └── agent.json       
│
├── common/
│   ├── a2a_client.py             # Utility to send requests to other agents
│   └── a2a_server.py             # Shared FastAPI A2A-compatible server template
│
├── shared/
│   └── schemas.py                # Shared Pydantic schema
│
├── streamlit_app.py             # Frontend UI for user input and response rendering
├── requirements.txt           
└── README.md                    

E questo è tutto! Ho messo insieme tutto ciò che abbiamo costruito in questo progetto GitHub.

Conclusione

Con poche app FastAPI e agent ADK, abbiamo costruito un pianificatore di viaggi collaborativo che:

• Comunica tramite il protocollo A2A
• Usa agent LLM per voli, soggiorni e attività
• Aggrega e mostra i risultati in una UI Streamlit pulita

Anche se tutti gli agent usano lo stesso modello sotto il cofano, questo sistema è multi‑agent nel comportamento: gli agent hanno ruoli distinti, responsabilità isolate e comunicazione strutturata.

Ecco alcune risorse per iniziare:

• A2A GitHub
• Crea un bot meteo con ADK
• Documentazione tecnica A2A