A2A — Para Desenvolvedores¶

Visão Geral¶

Este documento cobre duas áreas:

como chamar agentes externos (iFriend como cliente A2A)
como o servidor A2A funciona (iFriend como provedor A2A)

Parte 1: Chamando Agentes Externos (iFriend → Parceiro)¶

A iFriend pode chamar agentes externos usando o A2AClientManager.

Via Environment Variable¶

# JSON com agentes registrados
export A2A_EXTERNAL_AGENTS='{
  "turismobot": {
    "url": "https://turismobot.example.com/a2a/",
    "auth_token": "token-do-parceiro",
    "name": "TurismoBot",
    "description": "Agente de pacotes turísticos"
  }
}'

Via Código¶

from ifriend_agent.a2a_client import get_a2a_client_manager

manager = get_a2a_client_manager()
manager.register_agent(
    agent_id="turismobot",
    url="https://turismobot.example.com/a2a/",
    auth_token="token-do-parceiro",
    name="TurismoBot",
)

Usando nas Tools¶

from ifriend_agent.tools.a2a_client_tools import (
    listar_agentes_a2a,
    chamar_agente_a2a,
    consultar_task_a2a,
)

# Listar agentes disponíveis
agents = await listar_agentes_a2a(tool_context)

# Chamar agente externo
result = await chamar_agente_a2a(
    agent_id="turismobot",
    mensagem="Quais pacotes para Buenos Aires?",
    session_id="sess-123",
)

Parte 2: Servidor A2A (iFriend como Provedor)¶

Arquitetura¶

Request HTTP
    |
    v
API Gateway (ESPv2/Apigee)
    |
    v (OAuth2/JWT validate)
Middleware Auth
    |
    v
A2A Handler
    |
    v
ADK Agent

Endpoints¶

Path	Método	Descrição
`/.well-known/agent-card.json`	GET	Agent Card (público)
`/invoke`	POST	message/send
`/stream`	POST	message/stream
`/task/{id}`	GET	tasks/get
`/task/{id}`	DELETE	tasks/cancel

Handler¶

from ifriend_agent.a2a import create_a2a_app

app = create_a2a_app(
    runner=runner,
    jwt_manager=jwt_manager,
    host="0.0.0.0",
    port=8080,
)

# No unified_bot.py:
if get_flag("ENABLE_A2A"):
    app.mount("/a2a", create_a2a_app(...))

Configurações¶

Variável	Default	Descrição
`ENABLE_A2A`	false	Habilita endpoint A2A
`ENABLE_A2A_STREAMING`	false	Habilita streaming SSE
`A2A_BASE_URL`	—	URL pública para Agent Card
`A2A_REQUIRED_ROLE`	ROLE_A2A_USER	Role mínima requerida
`A2A_TASK_TTL_HOURS`	24	TTL de tasks

Parte 3: Autenticação (Escalabilidade)¶

Para Parceiros Individuais (Client Credentials)¶

# Auth server simples para A2A
async def validate_client_credentials(client_id: str, client_secret: str) -> Optional[dict]:
    # Consultar no banco de clientes
    client = await db.clients.find_one({"client_id": client_id})
    if not client or client["secret"] != hash(client_secret):
        return None
    return client

Para Empresas com Auth Próprio (JWT)¶

# Validar JWT do parceiro
async def validate_partner_jwt(token: str) -> Optional[dict]:
    # Decodificar e validar
    payload = jwt.decode(
        token,
        public_key,  # Chave pública do parceiro
        algorithms=["RS256"],
        audience="ifriend-a2a",
    )
    return payload

Estrutura do Módulo¶

ifriend_agent/a2a.py
├── create_a2a_app          # Factory de app FastAPI
├── A2ARouter            # Rotas A2A
└── task_store          # Persistência de tasks

ifriend_agent/a2a_client.py
├── A2AClientManager     # Gerencia múltiplos agentes
├── A2AClient          # Cliente para agente específico
├── RegisteredAgent     # Dataclass de agente
└── A2AResponse       # Dataclass de resposta

ifriend_agent/tools/a2a_tools.py
├── listar_agentes_a2a_tool
├── chamar_agente_a2a_tool
└── consultar_task_a2a_tool

Testes¶

cd ifriend_agent
pytest tests/test_a2a.py -v
pytest tests/test_a2a_client.py -v

Configurações de Environment¶

Variável	Default	Descrição
`A2A_CLIENT_TIMEOUT`	60	Timeout em segundos
`A2A_EXTERNAL_AGENTS`	{}	JSON de agentes
`ENABLE_A2A`	false	Habilita servidor A2A
`ENABLE_A2A_STREAMING`	false	Habilita streaming