Kit de desenvolvimento de agentes (ADK) do Google: Um guia com projeto de demonstração

Neste blog, mostrarei a você como criar um assistente de viagem com vários agentes e com tecnologia de IA usando o novo código aberto do Google, o Kit de desenvolvimento de agentes (ADK) de código aberto do Google e o A2A (agente para agente) (agente para agente).

Usaremos vários agentes para lidar com recomendações de voos, hotéis e atividades, comunicando-nos por meio de APIs REST e servidores FastAPI. No final, vamos encerrar tudo com um front-end limpo do Streamlit para proporcionar uma experiência de usuário intuitiva.

Mantemos nossos leitores atualizados sobre as últimas novidades em IA enviando o The Median, nosso boletim informativo gratuito de sexta-feira que detalha as principais histórias da semana. Inscreva-se e fique atento em apenas alguns minutos por semana:

O que é o ADK (Agent Development Kit) do Google?

O ADK do Google é uma estrutura modular, pronta para produção, para a criação de agentes acionados por LLM. É o mesmo kit de ferramentas que capacita os agentes dentro dos produtos do Google, como o Agentspace e o Customer Engagement Suite. Agora de código abertoele ajuda os desenvolvedores a criar aplicativos multiagentes avançados, flexíveis e interoperáveis.

Por que usar o Agent Development Kit (ADK)?

O ADK oferece a flexibilidade do Python com estruturas integradas para gerenciamento de estado, retornos de chamada, streaming e entrada/saída estruturada. Vamos dar uma olhada em seus principais recursos:

• Multiagente por design: O ADK pode compor agentes em fluxos de trabalho paralelos, sequenciais ou hierárquicos.
• Model-agnostic: Ele funciona com Gemini, GPT-4o, Claude, Mistral e outros via LiteLlm.
• Modular e dimensionável: O usuário pode definir agentes especializados e delegar de forma inteligente usando a orquestração integrada.
• Pronto para streaming: Ele oferece suporte à interação em tempo real, incluindo áudio/vídeo bidirecional.
• Ferramentas integradas: Ele oferece suporte à CLI local e à interface do usuário da Web para depuração e avaliação.
• Oferece suporte à implementação: O ADK facilmente coloca em contêineres e implementa agentes em vários ambientes.

O que é o protocolo Agent2Agent (A2A) do Google?

O protocolo Agent2Agent (A2A) é um padrão aberto e independente de fornecedor desenvolvido pelo Google para facilitar a comunicação e a colaboração entre agentes de IA em diversas plataformas e estruturas.

Os agentes do ADK expõem um endpoint HTTP padrão /run e metadados por meio de .well-known/agent.json. Isso permite a descoberta de agentes e a fácil comunicação entre eles (ou até mesmo com orquestradores externos, como LangGraph ou CrewAI).

Embora seja opcional, a adição do arquivo de metadados A2A torna seus agentes interoperáveis com o ecossistema mais amplo de ferramentas de agentes e orquestradores.

Visão geral do projeto: Planejador de viagens com tecnologia de IA com ADK e A2A

Este projeto cria um planejador de viagens que:

• Recebe como entrada o destino, as datas e o orçamento.
• Chama três agentes separados:
• flight_agent:  Recomenda opções de voo.
• stay_agent: Encontra hotéis dentro do orçamento.
• activities_agent: Sugere atividades envolventes.
• Em seguida, usa um site central host_agent para orquestrar todas as solicitações.
• Por fim, usa uma interface de usuário Streamlit para interação com o usuário.

Todos os agentes são hospedados como servidores FastAPI separados e expõem um ponto de extremidade /run. A comunicação é feita por meio do cliente compartilhado compatível com A2A.

Observação: Este projeto é executado inteiramente em sua máquina local para simplificar, mas você pode facilmente implantar cada agente e a interface do usuário em plataformas de nuvem como Render, Railway ou Google Cloud Run para obter acesso dimensionável.

Etapa 1: Pré-requisitos

Vamos começar instalando as seguintes bibliotecas:

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

Em seguida, configure sua chave de API da OpenAI - você pode usar outro provedor de modelos. Para saber como configurar sua chave de API da OpenAI, recomendo que você leia este tutorial introdutório sobre a API GPT-4o.

export OPENAI_API_KEY="your_key_here"

Etapa 2: Esquema e utilitários compartilhados

Antes de podermos criar agentes inteligentes, precisamos definir uma linguagem comum para que eles conversem entre si. Em nossa configuração, isso é feito usando:

• Um esquema compartilhado para entrada (definido via Pydantic)
• Um utilitário de cliente REST para chamar agentes
• Um wrapper de servidor REST para padronizar o ponto de extremidade /run em todos os agentes

Eles são colocados nas pastas shared/ e common/ para manter o código modular. Vamos dar uma olhada em cada um deles.

Criando um shared/schemas.py arquivo

Definimos um esquema TravelRequest usando Pydantic. Isso garante que todos os agentes concordem com a estrutura das solicitações recebidas, que inclui o destino, as datas da viagem e o orçamento do usuário.

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

Esta aula ajuda você a:

• Manter a entrada consistente para todos os agentes.
• Adição de validação automática com FastAPI.
• Simplificando a reutilização do código.

Criando um common/a2a_client.py arquivo

Esse utilitário assíncrono leve permite que qualquer agente (especialmente o host) invoque outro agente usando o protocolo A2A chamando o ponto de extremidade /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()

Esse utilitário envia de forma assíncrona uma solicitação POST para o ponto de extremidade /run de outro agente usando httpx.  Ele retorna a resposta JSON analisada e gera um erro se a solicitação falhar.

Usaremos esse utilitário em nosso agente host para chamar flight_agent, stay_agent e activities_agent.

Criando um common/a2a_server.py arquivo

Em vez de escrever uma rota FastAPI personalizada para cada agente, nós a generalizamos usando a função create_app(agent), que manipula:

• Servindo o agente em /run
• Recebimento de uma solicitação de viagem
• Retornando uma resposta estruturada

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

Este utilitário cria um aplicativo FastAPI com uma rota /run padrão que delega a execução ao agente fornecido.  Ele garante uma interface agente a agente (A2A) consistente para todos os serviços que usam entrada JSON estruturada.

Juntos, esses componentes compartilhados tornam nosso sistema multiagente mais sustentável, reutilizável e alinhado com a filosofia A2A do Google de mensagens simples e estruturadas entre agentes.

Etapa 3: Criação do sistema multiagente com ADK e A2A

Agora que já temos contratos e utilitários compartilhados, vamos começar a criar os agentes individuais. Para transformar isso em um sistema verdadeiramente modular e multiagente, usaremos o protocolo A2A do Google, uma interface simples baseada em HTTP que permite que os agentes se comuniquem de forma consistente e interoperável.

O A2A (Agent-to-Agent) permite a coordenação plug-and-play entre agentes, sejam eles funções Python locais ou hospedados em redes. Cada agente expõe um ponto de extremidade /run com um esquema comum e atua como um serviço.

Em nossa demonstração, temos quatro agentes:

• host_agent: Orquestra todos os outros agentes
• flight_agent: Encontrar voos adequados
• stay_agent: Sugere acomodações
• activities_agent: Recomenda que você participe de atividades locais

Todos os agentes são estruturados de forma semelhante, com 3 arquivos e uma subpasta:

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/

Cada agente usa google.adk.agents.Agent, um wrapper de modelo LiteLlm e Runner para execução. Comece criando os seguintes arquivos na pasta activities_agent e repita o mesmo para flight_agent e stay_agent.

Criando um agent.py arquivo

Vamos agora definir a lógica do nosso site activities_agent, que será responsável por gerar experiências locais envolventes com base no itinerário de viagem do usuário. 

Etapa 1: Importações

Começamos importando os módulos essenciais para configurar e executar nosso agente.

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

Esse agente usa os componentes do Google ADK, como Agent, ' Runner, ' e LiteLlm, e lida com o gerenciamento de estado usando InMemorySessionService.. A biblioteca Types é usada para criar prompts estruturados.

Etapa 2: Agente de atividades 

Agora, instanciamos o próprio agente usando a classe Agent do 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."
    )
)

O parâmetro instruction define o prompt do sistema que orienta o comportamento do LLM. Embora este exemplo use inglês simples, você pode ajustar a instrução para retornar JSON estruturado para facilitar a análise.

Etapa 3: Gerenciamento de sessões

Em seguida, para manter o controle das interações do usuário, configuramos um programa Runner junto com as informações da sessão.

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

O Runner gerencia a execução do agente para uma determinada sessão do aplicativo. Já a classe InMemorySessionService armazena o contexto na memória. Em seguida, definimos IDs de usuário e de sessão. No entanto, na produção, eles podem ser dinâmicos ou específicos do usuário. Isso garante a existência de uma nova sessão do ADK antes de enviar qualquer prompt para o agente LLM.

Etapa 4: Execução da lógica do agente

A função execute() lida com as solicitações recebidas, cria um prompt, invoca o modelo e analisa a saída.

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

A função execute() cria dinamicamente um prompt usando os parâmetros da solicitação de entrada, como destino, datas e orçamento. Aqui está o que acontece sob o capô:

• O prompt instrui o modelo a retornar um objeto JSON estruturado contendo uma lista de atividades.
• Um objeto Content é construído e passado para o ADK Runner, que aguarda de forma assíncrona a resposta final do modelo usando um gerador de streaming.
• Quando a resposta final é recebida, o agente extrai a saída de texto bruto e tenta analisá-la como JSON.
• Se a análise for bem-sucedida e a chave de atividades esperada existir, os dados estruturados serão retornados.
• Se a chave estiver ausente ou malformada, a alternativa é retornar a resposta de texto bruto para que a interface do usuário ainda tenha uma saída utilizável.
• Essa abordagem de manuseio duplo garante uma degradação graciosa quando o LLM retorna texto simples em vez de JSON estruturado.

Essa estratégia melhora a robustez e a experiência do usuário, especialmente quando os resultados do modelo variam ligeiramente devido à temperatura ou à interpretação imediata.

Criando um task_manager.py arquivo

Depois de definir a lógica do execute() dentro do agent.py, agora vamos conectá-lo à configuração do servidor compatível com o ADK usando o task_manager.py.

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

Esse arquivo funciona como um invólucro fino em torno da função execute() definida anteriormente. Ele torna o método run() disponível para módulos externos, especialmente o script do servidor em __main__.py.

Criando um __main__.py arquivo

O arquivo __main__.py inicia um servidor FastAPI na porta 8003, servindo o agente no ponto de extremidade /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)

Aqui está o que está acontecendo:

• O create_app() (de common/a2a_server.py) envolve nosso agente em uma interface FastAPI compatível com A2A padrão.
• Construímos dinamicamente um objeto com um método execute() para que o ADK possa invocar run() corretamente.
• Essa separação nos permite manter interfaces de API sem estado e reutilizar a lógica do agente principal.

Criando um .well-known/agent.json arquivo

Usamos esse arquivo JSON para descrever a identidade e a finalidade do agente de acordo com o protocolo A2A (Agent-to-Agent).

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

Observação: Embora o arquivo .well-known/agent.json não seja usado diretamente pelos nossos agentes neste projeto, ele segue a especificação A2A e é importante para descoberta, introspecção e compatibilidade futura com orquestradores como LangGraph, CrewAI ou o registro de agentes do Google.

Uma lógica semelhante é usada para flight_agent e stay_agent também.

Etapa 4: Coordenação com host_agent

O host_agent atua como um planejador central para a demonstração. O site host_agent exemplifica o padrão de controle em sistemas multiagentes. Ele separa a tomada de decisões e a execução, permitindo que cada agente downstream se concentre em seu nicho enquanto centraliza a lógica de coordenação. Isso não apenas simplifica os testes e o dimensionamento, mas também reflete a arquitetura de microsserviço do mundo real em sistemas distribuídos.

Ele envia a mesma carga útil para todos os três agentes usando suas APIs /run expostas e mescla os resultados. Vamos adicionar os seguintes arquivos à pasta host_agent.

Criando um agent.py arquivo

Vamos começar com as importações básicas.

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

Esse bloco de importação traz todos os principais componentes necessários para você definir e executar um agente baseado em LLM usando o Google ADK: classe Agent, wrapper LLM leve, Runner para lidar com a execução e gerenciamento de sessões na memória.

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"

O código acima define um agente ADK de nível superior responsável por coordenar o plano de viagem completo. Embora não invoquemos subagentes do LLM nesta implementação, o prompt do sistema define a função para uma extensão futura em que o LLM poderia lidar com o uso de ferramentas e o meta-raciocínio.

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}

Essa função execute() serve como o principal ponto de entrada para o LLM do agente host. It:

• Inicializa uma sessão (para suporte de memória, se necessário)
• Constrói dinamicamente um prompt de usuário
• Envia-o para o modelo usando runner.run_async() do ADK method
• Por fim, você aguarda e extrai a resposta final

Criando um task_manager.py arquivo

O gerenciador de tarefas executa a lógica de orquestração chamando agentes remotos e lidando com o fluxo de trabalho completo de planejamento de viagens. Para sua implementação prática, definimos as URLs de serviço para cada agente filho. Esses pontos de extremidade estão em conformidade com o esquema JSON A2A /run protocol and expect a shared 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"

Agora, definimos a carga útil.

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

Essa função usa a função auxiliar call_agent() para enviar a carga útil para cada serviço downstream e registra entradas e saídas para visibilidade durante o desenvolvimento. Esse arquivo é essencialmente onde reside a verdadeira lógica de orquestração.

Criando um __main__.py arquivo

O arquivo __main__.py serve como ponto de entrada para o servidor FastAPI que envolve o agente host. 

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)

O arquivo principal faz o seguinte:

• Ele usa create_app() de common/a2a_server.py para gerar um aplicativo FastAPI com um ponto de extremidade /run padronizado.
• Em seguida, ele passa um objeto simples do tipo agente com um método execute() que delega internamente para a função task_manager.run().
• Por fim, ele inicia o servidor FastAPI usando uvicorn em uma porta especificada (geralmente 8000).

Isso alinha a interface do agente host com outros agentes downstream, mantendo a consistência em todo o sistema.

Criando um .well-known/agent.json arquivo

Esse arquivo funciona como um padrão multiagente clássico em que um nó central delega e compõe tarefas. 

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

Embora seja opcional, é uma ótima prática incluir isso em todos os diretórios de agentes, conforme explicado anteriormente.

Etapa 5: Criando a interface do usuário com o Streamlit

Por fim, vamos criar um aplicativo simples em que os usuários possam inserir suas preferências e receber um itinerário estruturado. Comece criando um arquivo travel_ui.py no diretório raiz e adicione o seguinte código a ele.

import streamlit as st
import requests

Importamos bibliotecas básicas como Streamlit e solicitações de suporte à interface do usuário.

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

O aplicativo Streamlit oferece uma interface de usuário fácil de usar para interagir com o planejador de viagens multiagente criado com o ADK. Aqui estão alguns aspectos que abordamos no código acima.

• Ele usa text_input, date_input e number_input para coletar origem, destino, datas e orçamento.
• Ao clicar em "Plan My Trip" (Planejar minha viagem), ele valida a entrada para garantir que nenhum campo seja deixado em branco.
• Se for válido, ele constrói uma carga útil JSON e envia uma solicitação POST para o site host_agent em http://localhost:8000/run.
• O site host_agent invoca todos os agentes secundários (voo, estadia, atividade), agrega suas respostas e retorna um plano de viagem unificado.
• A resposta é analisada e exibida usando o método st.markdown() em cabeçalhos separados para voos, estadias e atividades.
• Se o backend falhar, uma mensagem de erro de fallback será exibida usando st.error().

Agora, execute o seguinte comando em seu terminal localmente:

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 um usuário clica em "Planejar minha viagem", o agente host assume o controle, ativa os agentes e exibe os resultados na interface do usuário:

Sua estrutura geral de arquivos seria semelhante a esta:

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 é isso! Eu reuni tudo o que construímos neste projeto do GitHub.

Conclusão

Com apenas alguns aplicativos FastAPI e agentes ADK, criamos um planejador de viagens colaborativo que:

• Comunica-se por meio do protocolo A2A
• Usa agentes do LLM para voos, estadias e atividades
• Agrega e exibe os resultados em uma interface de usuário Streamlit limpa

Embora todos os agentes usem o mesmo modelo, o comportamento desse sistema é multiagente, ou seja, os agentes têm funções distintas, responsabilidades isoladas e comunicação estruturada.

Aqui estão alguns recursos para você começar:

• A2A GitHub
• Crie um bot meteorológico com o ADK
• Documentação técnica A2A