Princípios fundamentais do protocolo A2A

1. Comunicação via JSON-RPC 2.0: define o formato de mensagens de requisição/resposta em JSON, transportado tipicamente sobre HTTP.
2. Agent Card obrigatório: todo agente expõe um documento descrevendo suas capacidades.
3. Ciclo de vida de tarefas: as tarefas possuem transições de estado bem definidas.
4. I/O estruturado: entradas e saídas são sempre JSON estruturado.
5. Autenticação padronizada: API Key ou OAuth 2.0.

O fluxo de funcionamento segue o padrão: o agente cliente descobre o Agent Card, envia uma tarefa e aguarda o resultado. O agente servidor publica seu Card, recebe requisições, executa e devolve a resposta.

Agent Card (Cartão de Agente)

Cada agente A2A disponibiliza publicamente o arquivo /.well-known/agent.json, funcionando como um "documento de identidade" de suas capacidades. Contém: descrição das habilidades, endpoint de acesso, método de autenticação e modalidades de interação suportadas (texto, formulários, mídia).

Padrões de mensagem e comunicação

• Estrutura: baseada em JSON-RPC 2.0, com suporte a partes multimodais (texto, arquivos, dados estruturados) acompanhadas de metadados.
• Modos de comunicação:
  • Síncrono: requisição/resposta para tarefas curtas.
  • Assíncrono: Webhook ou SSE para tarefas longas e atualizações de status.
  • Streaming: SSE para progresso em tempo real.

Ciclo de vida das tarefas

A colaboração entre agentes ocorre por meio de objetos de tarefa com transição de estados: submetida → aceita → em_processamento → concluída / falhou. O resulatdo final é entregue como um Artifact, que pode ser texto, tabela, imagem ou JSON estruturado.

Exemplo prático de implementação

A seguir, apresentamos um servidor A2A com capacidades matemáticas e um cliente correspondente.

Servidor A2A

from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.llms import Ollama

app = FastAPI(title="A2A Math Service", version="1.0")

# Configuração do modelo de linguagem
modelo = Ollama(model="llama3")

template_prompt = ChatPromptTemplate.from_messages([
    ("system", "Você é um tutor especializado em matemática, respondendo apenas questões relacionadas ao tema."),
    ("user", "{pergunta}")
])
cadeia_resposta = template_prompt | modelo | StrOutputParser()

# Modelos de dados para JSON-RPC 2.0
class RequisicaoRPC(BaseModel):
    jsonrpc: str = "2.0"
    identificador: str
    metodo: str
    parametros: dict

class RespostaRPC(BaseModel):
    jsonrpc: str = "2.0"
    identificador: str
    resultado: dict
    erro: dict | None = None

# Armazenamento em memória para estados de tarefa
registro_tarefas = {}

@app.get("/.well-known/agent.json")
async def obter_cartao_agente():
    """Retorna o cartão de capacidades do agente A2A."""
    return {
        "nome": "Assistente-Matematico",
        "descricao": "Agente A2A para operações matemáticas e respostas a perguntas",
        "versao": "1.0",
        "capacidades": ["calc.soma", "calc.produto", "ia.pergunta"],
        "endpoints": {
            "tarefa": "/rpc/tarefa",
            "consulta": "/rpc/consulta/{tid}"
        },
        "autenticacao": {
            "tipo": "api_key",
            "local": "header",
            "campo": "X-A2A-Token"
        }
    }

@app.post("/rpc/tarefa", response_model=RespostaRPC)
async def processar_tarefa(
    req: RequisicaoRPC,
    x_a2a_token: str = Header(None)
):
    # Validação de credencial
    if x_a2a_token != "chave-secreta-123":
        raise HTTPException(status_code=401, detail="Token inválido")

    if req.jsonrpc != "2.0":
        return RespostaRPC(
            identificador=req.identificador,
            resultado={},
            erro={"code": -32600, "message": "Formato JSON-RPC inválido"}
        )

    tid = req.parametros.get("task_id")
    acao = req.parametros.get("acao")
    dados = req.parametros.get("dados", {})

    registro_tarefas[tid] = {"estado": "aceita", "acao": acao}

    saida = {}
    try:
        if acao == "calc.soma":
            x = dados.get("x", 0)
            y = dados.get("y", 0)
            saida = {"valor": x + y, "estado": "concluida"}

        elif acao == "calc.produto":
            x = dados.get("x", 0)
            y = dados.get("y", 0)
            saida = {"valor": x * y, "estado": "concluida"}

        elif acao == "ia.pergunta":
            pergunta = dados.get("pergunta", "")
            resposta_ia = cadeia_resposta.invoke({"pergunta": pergunta})
            saida = {"texto_resposta": resposta_ia, "estado": "concluida"}

        else:
            saida = {"estado": "falhou", "motivo": "Ação não suportada"}

        registro_tarefas[tid].update(saida)

        return RespostaRPC(
            identificador=req.identificador,
            resultado={"task_id": tid, "estado": saida["estado"], "dados": saida}
        )

    except Exception as e:
        registro_tarefas[tid] = {"estado": "falhou", "motivo": str(e)}
        return RespostaRPC(
            identificador=req.identificador,
            resultado={},
            erro={"code": -32000, "message": f"Erro: {str(e)}"}
        )

@app.get("/rpc/consulta/{tid}")
async def consultar_estado(tid: str):
    if tid not in registro_tarefas:
        raise HTTPException(status_code=404, detail="Tarefa não encontrada")
    return {"jsonrpc": "2.0", "result": {"task_id": tid, **registro_tarefas[tid]}}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Cliente A2A

import requests
import json

URL_BASE = "http://localhost:8000"
TOKEN = "chave-secreta-123"

class ClienteA2A:
    def __init__(self, url_base, token):
        self.url_base = url_base
        self.cabecalhos = {
            "Content-Type": "application/json",
            "X-A2A-Token": token
        }

    def descobrir_agente(self):
        """Obtém o Agent Card do servidor."""
        endpoint = f"{self.url_base}/.well-known/agent.json"
        resp = requests.get(endpoint)
        if resp.status_code == 200:
            info = resp.json()
            print(f"Agente encontrado: {info['nome']}")
            print(f"Capacidades: {info['capacidades']}")
            return info
        print(f"Falha na descoberta: {resp.status_code}")
        return None

    def enviar_tarefa(self, tid, acao, dados):
        """Submete uma tarefa via JSON-RPC 2.0."""
        endpoint = f"{self.url_base}/rpc/tarefa"
        corpo = {
            "jsonrpc": "2.0",
            "identificador": tid,
            "metodo": "a2a.task.submit",
            "parametros": {"task_id": tid, "acao": acao, "dados": dados}
        }
        resp = requests.post(endpoint, headers=self.cabecalhos, json=corpo)
        if resp.status_code == 200:
            resultado = resp.json()
            print(json.dumps(resultado, indent=2, ensure_ascii=False))
            return resultado
        print(f"Erro ao enviar tarefa: {resp.status_code}")
        return None

    def consultar(self, tid):
        """Verifica o estado de uma tarefa."""
        endpoint = f"{self.url_base}/rpc/consulta/{tid}"
        resp = requests.get(endpoint, headers=self.cabecalhos)
        if resp.status_code == 200:
            resultado = resp.json()
            print(json.dumps(resultado, indent=2, ensure_ascii=False))
            return resultado

if __name__ == "__main__":
    cliente = ClienteA2A(URL_BASE, TOKEN)

    # Descoberta do agente
    agente = cliente.descobrir_agente()
    if not agente:
        exit(1)

    # Teste de soma
    cliente.enviar_tarefa("t-100", "calc.soma", {"x": 15, "y": 25})

    # Teste de pergunta com IA
    cliente.enviar_tarefa("t-101", "ia.pergunta", {"pergunta": "Explique o teorema de Pitágoras com um exemplo"})

    # Consulta de estados
    cliente.consultar("t-100")
    cliente.consultar("t-101")

O exemplo demonstra os elementos essenciais do A2A: descoberta via Agent Card, comunicação estruturada em JSON-RPC 2.0, ciclo de vida de tarefas com rastreamento de estado e suporte a múltiplas capacidades em um único agente servidor.