Requisitos do Sistema e Instalação
Antes de continuar, verifique:
- Python 3.8 ou superior
- pip instalado
- Acesso à internet (para pacotes e servidor A2A)
1. Clonar o Repositório
git clone https://gitcode.com/gh_mirrors/a2/a2a-python
cd a2a-python
2. Instalar Dependências
O projeto utiliza uv como gerenciador:
pip install uv
uv install
Conceitos Essenciais: Componente Cliente
O núcleo do SDK é o cliente, localizado em src/a2a/client/client.py. Ele gerencia conexões e troca de mensagens.
ClientConfig controla parâmetros como:
streaming: ativar fluxo contínuopolling: usar verificação periódicahttpx_client: instância HTTP personalizadagrpc_channel_factory: fábrica de canais gRPCsupported_transports: transportes aceitos (ex.: grpc, jsonrpc, rest)
Interface Client define operações fundamentais:
send_message: enviar mensagemget_task: consultar estado de tarefacancel_task: interromper tarefa
Primeiros Passos: Enviar sua Mensagem Inicial
1. Configurar o Cliente
from a2a.client.client import ClientConfig
cfg = ClientConfig(
streaming=True,
supported_transports=["grpc", "jsonrpc", "rest"]
)
2. Inicializar o Cliente
from a2a.client.client_factory import ClientFactory
fab = ClientFactory(cfg)
# Cartão do agente alvo
agent_card = {
"agent_id": "meu-agente-123",
"transport": {"type": "grpc", "url": "http://localhost:50051"}
}
cliente = fab.create_client(agent_card)
3. Construir e Enviar Mensagem
from a2a.types import Message
msg_obj = Message(
content="Olá do SDK Python A2A!",
sender="meu-agente-123",
recipient="agente-destino"
)
async def enviar_primeira():
async for evt in cliente.send_message(msg_obj):
print(f"Evento recebido: {evt}")
import asyncio
asyncio.run(enviar_primeira())
Transportes Suportados
O SDK oferece múltiplos protocolos, configuráveis por ordem de preferência:
- gRPC: implementado em
src/a2a/client/transports/grpc.py, comunicação binária eficiente. - JSON-RPC: em
src/a2a/client/transports/jsonrpc.py, baseado em JSON sobre HTTP. - REST: em
src/a2a/client/transports/rest.py, ideal para cenários simples.
Funcionalidades Avançadas: Interceptadores e Consumidores
Interceptadores (middleware) estendem o cliente, por exemplo, com logs ou autenticação:
from a2a.client.middleware import ClientCallInterceptor
class LogInterceptor(ClientCallInterceptor):
async def intercept(self, method, request, context, prox):
print(f"Enviando: {request}")
resp = await prox(method, request, context)
print(f"Recebido: {resp}")
return resp
cliente = fab.create_client(agent_card, middleware=[LogInterceptor()])
Consumidores de eventos tratam atualizações de tarefas:
async def meu_consumidor(evento, cartao):
print(f"Evento de {cartao['agent_id']}: {evento}")
cliente = fab.create_client(agent_card, consumers=[meu_consumidor])
Recursos para Aprofundamento
- Testes: diretório
tests/client/contém exemplos de uso. - Definição do protocolo:
src/a2a/grpc/possui arquivos proto. - Ferramentas auxiliares:
scripts/contém geradores e formatadores.