Guia Prático A2A Python SDK: Preparação e Primeira Mensagem

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ínuo
  • polling: usar verificação periódica
  • httpx_client: instância HTTP personalizada
  • grpc_channel_factory: fábrica de canais gRPC
  • supported_transports: transportes aceitos (ex.: grpc, jsonrpc, rest)

Interface Client define operações fundamentais:

  • send_message: enviar mensagem
  • get_task: consultar estado de tarefa
  • cancel_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.

Tags: A2A Python SDK Agent-to-Agent gRPC JSON-RPC

Publicado em 7-14 21:00