Agent Development Kit (ADK) dari Google: Panduan dengan Proyek Demo

Gambaran Proyek: Perencana Perjalanan Berbasis AI dengan ADK dan A2A

Proyek ini membangun perencana perjalanan yang:

• Menerima tujuan, tanggal, dan anggaran sebagai masukan.
• Memanggil tiga agent terpisah:
• flight_agent:  Merekomendasikan opsi penerbangan.
• stay_agent: Menemukan hotel sesuai anggaran.
• activities_agent: Menyarankan aktivitas yang menarik.
• Kemudian, menggunakan host_agent pusat untuk mengorkestrasi semua permintaan.
• Terakhir, menggunakan UI Streamlit untuk interaksi pengguna.

Semua agent dihosting sebagai server FastAPI terpisah dan mengekspos endpoint /run. Komunikasi dilakukan melalui klien bersama yang kompatibel dengan A2A.

Catatan: Proyek ini berjalan sepenuhnya di mesin lokal Anda demi kesederhanaan, tetapi Anda dapat dengan mudah men-deploy setiap agent dan UI ke platform cloud seperti Render, Railway, atau Google Cloud Run untuk akses yang skalabel.

Langkah 1: Prasyarat

Mari mulai dengan memasang pustaka berikut:

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

Lalu, atur kunci API OpenAI Anda—silakan gunakan penyedia model lainnya. Untuk mempelajari cara menyiapkan kunci API OpenAI Anda, saya merekomendasikan tutorial pengantar ini tentang GPT-4o API.

export OPENAI_API_KEY="your_key_here"

Langkah 2: Skema dan Utilitas Bersama

Sebelum kita dapat membangun agent cerdas, kita perlu mendefinisikan bahasa umum agar mereka dapat saling berbicara. Dalam pengaturan kita, ini dilakukan menggunakan:

• Skema bersama untuk input (didefinisikan melalui Pydantic)
• Utilitas klien REST untuk memanggil agent
• Pembungkus server REST untuk menstandarkan endpoint /run di semua agent

Komponen ini ditempatkan di folder shared/ dan common/ untuk menjaga kode tetap modular. Mari lihat satu per satu.

Membuat berkas shared/schemas.py

Kita mendefinisikan skema TravelRequest menggunakan Pydantic. Ini memastikan semua agent sepakat atas struktur permintaan yang masuk, yang mencakup tujuan, tanggal perjalanan, dan anggaran pengguna.

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

Kelas ini membantu dalam:

• Menjaga input konsisten untuk semua agent.
• Menambahkan validasi otomatis dengan FastAPI.
• Menyederhanakan penggunaan ulang kode.

Membuat berkas common/a2a_client.py

Utilitas async ringan ini memungkinkan agent mana pun (terutama host) memanggil agent lain menggunakan protokol A2A dengan memanggil 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()

Utilitas ini mengirim permintaan POST secara asynchronous ke endpoint /run milik agent lain menggunakan httpx. Ia mengembalikan respons JSON yang telah diuraikan dan menimbulkan error jika permintaan gagal.

Kita akan menggunakan utilitas ini di host agent untuk memanggil flight_agent, stay_agent, dan activities_agent.

Membuat berkas common/a2a_server.py

Alih-alih menulis rute FastAPI kustom untuk tiap agent, kita menggeneralisasikannya menggunakan fungsi create_app(agent), yang menangani:

• Menyajikan agent pada /run
• Menerima permintaan perjalanan
• Mengembalikan respons terstruktur

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

Utilitas ini membuat aplikasi FastAPI dengan rute standar /run yang mendelegasikan eksekusi ke agent yang diberikan.  Ini memastikan antarmuka agent-to-agent (A2A) yang konsisten untuk semua layanan dengan input JSON terstruktur.

Bersama-sama, komponen bersama ini membuat sistem multi-agent kita lebih mudah dirawat, dapat digunakan ulang, dan selaras dengan filosofi A2A dari Google tentang perpesanan antar agent yang sederhana dan terstruktur.

Langkah 3: Membangun Sistem Multi-Agent dengan ADK dan A2A

Sekarang kontrak dan utilitas bersama telah siap, mari mulai membangun agent individual. Untuk menjadikannya sistem yang benar-benar modular dan multi-agent, kita akan menggunakan protokol A2A dari Google—antarmuka berbasis HTTP sederhana yang memungkinkan agent berkomunikasi secara konsisten dan dapat saling beroperasi.

A2A (Agent-to-Agent) memungkinkan koordinasi plug-and-play di antara agent, baik itu fungsi Python lokal maupun yang dihosting di jaringan. Setiap agent mengekspos endpoint /run dengan skema umum dan bertindak sebagai layanan.

Dalam demo kita, ada empat agent:

• host_agent: Mengorkestrasi semua agent lainnya
• flight_agent: Menemukan penerbangan yang sesuai
• stay_agent: Menyarankan akomodasi
• activities_agent: Merekomendasikan aktivitas lokal yang menarik

Semua agent disusun serupa, dengan 3 berkas dan sebuah subfolder:

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/

Setiap agent menggunakan google.adk.agents.Agent, pembungkus model LiteLlm, dan Runner untuk eksekusi. Mulailah dengan membuat berkas berikut dalam folder activities_agent lalu ulangi hal yang sama untuk flight_agent dan stay_agent.

Membuat berkas agent.py

Sekarang mari kita definisikan logika untuk activities_agent, yang bertanggung jawab menghasilkan pengalaman lokal yang menarik berdasarkan rencana perjalanan pengguna. 

Langkah 1: Import

Kita mulai dengan mengimpor modul penting untuk mengonfigurasi dan menjalankan 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

Agent ini menggunakan komponen Google ADK seperti Agent, ' Runner, ' dan LiteLlm serta menangani manajemen state menggunakan InMemorySessionService. Pustaka Types digunakan untuk menyusun prompt terstruktur.

Langkah 2: Agent aktivitas 

Sekarang, kita membuat instance agent itu sendiri menggunakan kelas Agent dari 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."
    )
)

Parameter instruction mendefinisikan system prompt yang memandu perilaku LLM. Meskipun contoh ini menggunakan bahasa Inggris biasa, Anda dapat menyesuaikan instruksi untuk mengembalikan JSON terstruktur agar lebih mudah diurai.

Langkah 3: Manajemen sesi

Berikutnya, untuk melacak interaksi pengguna, kita mengonfigurasi Runner beserta informasi sesi.

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

Runner mengelola eksekusi agent untuk sesi aplikasi tertentu. Sementara kelas InMemorySessionService menyimpan konteks di memori. Lalu, kita menentukan ID pengguna dan sesi. Namun, di produksi, ini bisa bersifat dinamis atau khusus pengguna. Ini memastikan sesi ADK baru tersedia sebelum mengirim prompt apa pun ke agent LLM.

Langkah 4: Menjalankan logika agent

Fungsi execute() menangani permintaan masuk, membangun prompt, memanggil model, dan mengurai keluarannya.

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

Fungsi execute() secara dinamis membangun prompt menggunakan parameter permintaan yang masuk seperti tujuan, tanggal, dan anggaran. Berikut yang terjadi di balik layar:

• Prompt menginstruksikan model untuk mengembalikan objek JSON terstruktur berisi daftar aktivitas.
• Objek Content dibuat dan dikirim ke ADK Runner, yang menunggu secara asynchronous respons final model menggunakan generator streaming.
• Setelah respons final diterima, agent mengekstrak keluaran teks mentah dan mencoba mengurainya sebagai JSON.
• Jika penguraian berhasil dan kunci activities yang diharapkan ada, data terstruktur dikembalikan.
• Jika kunci hilang atau salah format, fallback-nya adalah mengembalikan teks mentah agar UI tetap memiliki output yang dapat digunakan.
• Pendekatan penanganan ganda ini memastikan degradasi yang anggun ketika LLM mengembalikan teks biasa alih-alih JSON terstruktur.

Strategi ini meningkatkan ketahanan dan pengalaman pengguna, terutama saat keluaran model sedikit bervariasi karena suhu atau interpretasi prompt.

Membuat berkas task_manager.py

Setelah mendefinisikan logika execute() di agent.py, kini kita menghubungkannya ke pengaturan server yang kompatibel dengan ADK menggunakan task_manager.py.

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

Berkas ini bertindak sebagai pembungkus tipis di sekitar fungsi execute() yang didefinisikan sebelumnya. Ini membuat metode run() tersedia untuk modul eksternal, terutama skrip server di __main__.py.

Membuat berkas __main__.py

Berkas __main__.py meluncurkan server FastAPI pada port 8003, menyajikan agent di 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)

Berikut yang terjadi:

• Fungsi create_app() (dari common/a2a_server.py) membungkus agent kita dalam antarmuka FastAPI standar yang kompatibel dengan A2A.
• Kita secara dinamis menyusun objek dengan metode execute() agar ADK dapat memanggil run() dengan benar.
• Pemisahan ini memungkinkan kita mempertahankan antarmuka API stateless sambil menggunakan ulang logika agent inti.

Membuat berkas .well-known/agent.json

Kami menggunakan berkas JSON ini untuk mendeskripsikan identitas dan tujuan agent sesuai protokol A2A (Agent-to-Agent).

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

Catatan: Meskipun berkas .well-known/agent.json tidak digunakan langsung oleh agent kita di proyek ini, berkas tersebut mematuhi spesifikasi A2A dan penting untuk penemuan, introspeksi, serta kompatibilitas di masa depan dengan orkestrator seperti LangGraph, CrewAI, atau registri agent Google.

Logika serupa juga digunakan untuk flight_agent dan stay_agent.

Langkah 4: Berkoordinasi dengan host_agent

host_agent bertindak sebagai perencana pusat untuk demo. host_agent mencontohkan pola controller dalam sistem multi-agent. Ia memisahkan pengambilan keputusan dan eksekusi, memungkinkan setiap agent hilir fokus pada bagiannya sementara logika koordinasi dipusatkan. Ini tidak hanya menyederhanakan pengujian dan penskalaan tetapi juga mencerminkan arsitektur microservice dunia nyata dalam sistem terdistribusi.

Agent ini mengirim payload yang sama ke ketiga agent menggunakan API /run yang diekspos dan menggabungkan hasilnya. Mari tambahkan berkas berikut ke folder host_agent.

Membuat berkas agent.py

Mari mulai dengan import dasar.

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

Blok import ini menghadirkan semua blok bangunan inti yang diperlukan untuk mendefinisikan dan menjalankan agent berbasis LLM menggunakan Google ADK: kelas Agent, pembungkus LLM ringan, Runner untuk menangani eksekusi, dan manajemen sesi dalam memori.

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"

Kode di atas mendefinisikan agent ADK tingkat atas yang bertanggung jawab mengoordinasikan rencana perjalanan secara menyeluruh. Meskipun kita tidak memanggil sub-agent dari LLM dalam implementasi ini, system prompt menetapkan peran untuk pengembangan di masa depan di mana LLM dapat menangani penggunaan alat dan penalaran meta.

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}

Fungsi execute() ini berfungsi sebagai titik masuk utama ke LLM milik host agent. Fungsi ini:

• Melakukan inisialisasi sesi (untuk dukungan memori bila diperlukan)
• Menyusun prompt pengguna secara dinamis
• Mengirimkannya ke model menggunakan metode runner.run_async() dari ADK
• Pada akhirnya menunggu dan mengekstrak respons final

Membuat berkas task_manager.py

Task manager menjalankan logika orkestrasi dengan memanggil agent jarak jauh dan menangani alur kerja perencanaan perjalanan secara penuh. Untuk implementasinya, kita mendefinisikan URL layanan untuk masing-masing agent turunan. Endpoint ini mematuhi protokol A2A /run dan mengharapkan skema JSON TravelRequest` bersama.

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"

Sekarang, kita definisikan 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.")
    }

Fungsi ini menggunakan helper call_agent() untuk mengirim payload ke setiap layanan hilir dan mencatat input serta output demi visibilitas selama pengembangan. Berkas inilah yang menjadi tempat logika orkestrasi sesungguhnya.

Membuat berkas __main__.py

Berkas __main__.py berfungsi sebagai titik masuk untuk server FastAPI yang membungkus 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)

Berkas utama melakukan hal-hal berikut:

• Menggunakan create_app() dari common/a2a_server.py untuk menghasilkan aplikasi FastAPI dengan endpoint /run yang distandarkan.
• Kemudian, meneruskan objek mirip agent sederhana dengan metode execute() yang secara internal mendelegasikan ke fungsi task_manager.run().
• Terakhir, memulai server FastAPI menggunakan uvicorn pada port yang ditentukan (biasanya 8000).

Ini menyelaraskan antarmuka host agent dengan agent hilir lainnya, menjaga konsistensi di seluruh sistem.

Membuat berkas .well-known/agent.json

Berkas ini berperan sebagai pola multi-agent klasik di mana simpul pusat mendelegasikan dan menyusun tugas. 

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

Meskipun opsional, menyertakan ini di semua direktori agent adalah praktik yang baik, seperti dijelaskan sebelumnya.

Langkah 5: Membangun UI dengan Streamlit

Terakhir, mari bangun aplikasi sederhana tempat pengguna dapat memasukkan preferensi dan menerima itinerary terstruktur. Mulailah dengan membuat berkas travel_ui.py di direktori root dan tambahkan kode berikut.

import streamlit as st
import requests

Kita mengimpor pustaka dasar seperti Streamlit dan requests untuk dukungan 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.")

Aplikasi Streamlit menyediakan UI yang ramah pengguna untuk berinteraksi dengan perencana perjalanan multi-agent yang dibangun menggunakan ADK. Berikut beberapa hal yang kita bahas dalam kode di atas.

• Menggunakan text_input, date_input, dan number_input untuk mengumpulkan asal, tujuan, tanggal, dan anggaran.
• Saat mengklik “Plan My Trip”, aplikasi memvalidasi input untuk memastikan tidak ada kolom yang kosong.
• Jika valid, aplikasi menyusun payload JSON dan mengirim permintaan POST ke host_agent di http://localhost:8000/run.
• host_agent memanggil semua agent anak (penerbangan, akomodasi, aktivitas), menggabungkan respons mereka, dan mengembalikan rencana perjalanan terpadu.
• Respons diurai dan ditampilkan menggunakan metode st.markdown() di bawah header terpisah untuk penerbangan, akomodasi, dan aktivitas.
• Jika backend gagal, pesan error fallback ditampilkan menggunakan st.error().

Sekarang, jalankan perintah berikut di terminal lokal Anda:

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

Saat pengguna mengeklik “Plan My Trip,” host agent mengambil alih, mengaktifkan agent, dan menampilkan hasil di UI:

Struktur berkas keseluruhan Anda akan tampak seperti ini:

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                    

Dan itu saja! Saya telah menggabungkan semua yang kita bangun dalam proyek GitHub ini.

Kesimpulan

Hanya dengan beberapa aplikasi FastAPI dan agent ADK, kita telah membangun perencana perjalanan kolaboratif yang:

• Berkomunikasi melalui protokol A2A
• Menggunakan agent LLM untuk penerbangan, akomodasi, dan aktivitas
• Menggabungkan dan menampilkan hasil dalam UI Streamlit yang bersih

Meskipun semua agent menggunakan model yang sama di balik layar, sistem ini bersifat multi-agent dalam perilaku, yaitu agent memiliki peran yang berbeda, tanggung jawab terisolasi, dan komunikasi terstruktur.

Berikut beberapa sumber untuk memulai:

• GitHub A2A
• Bangun bot cuaca dengan ADK
• Dokumentasi teknis A2A