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

Habilitando no root_agent¶

Essas tools só ficam disponíveis para o root_agent (e portanto para o LLM decidir usá-las) se ENABLE_A2A_CLIENT=true — ver ifriend_agent/agent_builder.py e ifriend_agent/config/feature_flags.py. Desabilitada por padrão até haver um agente externo real configurado via A2A_EXTERNAL_AGENTS.

Para testar localmente sem depender do agente externo real (ex: antes do Agent Card do Agentforce estar disponível), suba um agente A2A de teste independente e valide o roundtrip completo:

# Terminal 1 — sobe um agente A2A de teste (fora da iFriend)
python examples/a2a_echo_agent.py --port 8766

# Terminal 2 — valida chamada direta à tool + (opcional) via root_agent/LLM
python examples/a2a_outbound_roundtrip_test.py
ENABLE_A2A_CLIENT=true python examples/a2a_outbound_roundtrip_test.py --via-agent

Parte 2: Servidor A2A (iFriend como Provedor)¶

Arquitetura¶

flowchart TD
    REQ["Request HTTP"] --> MW["A2AJWTAuthMiddleware (JWT Bearer + role ROLE_A2A_USER)"]
    MW --> H["DefaultRequestHandler (a2a-sdk)"]
    H --> EX["IfriendA2aAgentExecutor"]
    EX --> ADK["ADK Agent (root_agent)"]

Não há API Gateway/Apigee/OAuth2 na frente do Cloud Run em produção — o endpoint A2A é exposto diretamente, protegido só pelo A2AJWTAuthMiddleware. Existe um caminho OAuth2 escrito em código (create_a2a_app_oauth2, runtime/a2a/auth/) mas ele nunca é montado em unified_bot.py — é um rascunho para uma eventual migração futura, não o que está no ar hoje (ver seção "Parte 3" abaixo).

Endpoints¶

O servidor A2A usa JSON-RPC 2.0 sobre HTTP. Todos os métodos são POST para o mesmo endpoint:

Path	Método	Descrição
`/.well-known/agent-card.json`	GET	Agent Card (público, sem auth)
`/`	POST	JSON-RPC: `message/send`, `message/stream`, `tasks/get`, `tasks/cancel`

Métodos JSON-RPC suportados:

Método	Descrição
`message/send`	Envia mensagem e inicia/continua task (multi-turn)
`message/stream`	Envia mensagem com streaming SSE (se `ENABLE_A2A_STREAMING=true`)
`tasks/get`	Consulta resultado de task existente
`tasks/cancel`	Cancela task em andamento

Handler¶

A integração é feita no unified_bot.py:

from runtime.a2a.server import create_a2a_app

if get_flag("ENABLE_A2A"):
    a2a_app = create_a2a_app(
        runner=runner,
        jwt_manager=jwt_manager,
        host="0.0.0.0",
        port=int(os.environ.get("PORT", 8080)),
        enable_task_persistence=True,
    )
    app.mount("/a2a", a2a_app)

Nota: O create_a2a_app usa A2AStarletteApplication.add_routes_to_app() diretamente (não o to_a2a() do ADK, que registra rotas via startup event — incompatível com app.mount()).

Contexto do chamador (extensão `caller-context/v1`)¶

O contrato para parceiros está em Para Parceiros → Contexto do chamador. A implementação tem três peças:

Peça	Arquivo	O que faz
Agent Card	`runtime/a2a/server.py` (`_caller_context_extension`)	Declara a extensão (`AgentExtension`, `required=False`, schema em `params`) e `application/json` nos input modes
Normalizador de body	`runtime/a2a/body_normalizer.py` (`A2ABodyNormalizerMiddleware`, ASGI puro)	Converte payload estilo protobuf/v1 para v0.3 antes da validação pydantic da a2a-sdk. O v0.3 passa intacto. Depois de entregar o body, delega o `receive` original (senão quebra o SSE de `message/stream`)
Request converter	`runtime/a2a/context_converter.py` (`ifriend_request_converter`, via `A2aAgentExecutorConfig`)	Chama o conversor padrão do ADK, extrai o contexto (DataPart > `message.metadata` > `params.metadata`), grava `state_delta = {a2a_inbound_context, message_metadata: {source: "a2a"}}` e troca o DataPart por uma linha de resumo para o LLM

No agente:

before_agent_callback_combined popula {a2a_inbound_hint?} no prompt do orquestrador (ifriend_agent/config/a2a_inbound.py), com a regra "sem handoff, não peça de novo os dados já enviados".
escalar_agentforce devolve status="origem_a2a" quando message_metadata.source == "a2a".

Para capturar um payload real, dá para ligar DEBUG só no logger a2a.server.apps.jsonrpc.jsonrpc_app (ele loga Request body). Cuidado: o body contém PII.

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 — o que está em produção vs. rascunho não implantado¶

Em produção: JWT Bearer (`create_a2a_app`)¶

Todo parceiro (incluindo o Agentforce) autentica assim: recebe uma conta de serviço (email/senha) provisionada pela iFriend com a role ROLE_A2A_USER, chama POST /authentication_token na API iFriend (mesmo endpoint usado internamente pelo próprio agente — ifriend_agent/tools/booking/auth.py) para obter um JWT, e envia esse JWT como Authorization: Bearer <token>. A2AJWTAuthMiddleware (runtime/a2a/server.py) valida assinatura, expiração e a role. Sem escopos, sem client_id/client_secret.

⚠️ Rascunho não implantado: OAuth2 Client Credentials (`create_a2a_app_oauth2`)¶

runtime/a2a/auth/ (auth_server.py, middleware.py, registry.py, models.py) implementa um servidor OAuth2 completo (POST /token com client_credentials, GET /jwks, escopos, rate limit por cliente) e create_a2a_app_oauth2() em server.py monta esse fluxo — mas nenhum dos dois é chamado por unified_bot.py. É código real, testável isoladamente, mas não é o que responde em https://agents.theifriend.com/trip/a2a/ hoje. Só documentar/oferecer a parceiros externos depois que for de fato montado em produção.

Estrutura do Módulo¶

runtime/a2a/
├── server.py              # create_a2a_app (EM PRODUÇÃO) + create_a2a_app_oauth2 (rascunho, não montado)
│   ├── A2AJWTAuthMiddleware (JWT Bearer auth)
│   ├── _build_agent_card() — Agent Card com skills + security schemes
│   └── create_a2a_app() — Starlette app com add_routes_to_app()
├── executor.py            # IfriendA2aAgentExecutor — cancel() cooperativo (vendor lança NotImplementedError)
├── context_converter.py   # request_converter: contexto do chamador → session.state (extensão caller-context/v1)
├── body_normalizer.py     # middleware ASGI: payload protobuf/v1 → v0.3
├── client.py              # A2AClientManager + A2AClient (iFriend chamando agentes externos)
├── task_persistence.py    # CloudSQLTaskStore — persistência de tasks (task store real, não in-memory)
├── auth/                  # ⚠️ Não usado em produção (ver Parte 3 acima)
│   ├── auth_server.py     # OAuth2 token server (client_credentials) — nunca montado
│   └── middleware.py      # A2AOAuth2Middleware — nunca montado

ifriend_agent/tools/a2a_client_tools.py   # habilitado via ENABLE_A2A_CLIENT
├── listar_agentes_a2a
├── chamar_agente_a2a
└── consultar_task_a2a

Testes¶

cd ifriend_agent
pytest tests/test_a2a.py -v                # auth middleware + agent card (app Starlette fake)
pytest tests/test_a2a_integration.py -v     # rotas JSON-RPC reais via create_a2a_app() (ASGI in-process)
pytest tests/test_a2a_caller_context.py -v  # contexto do chamador (DataPart/metadata/protobuf) + Agent Card
pytest tests/test_a2a_client.py -v          # A2AClient (inclui conformidade de schema com o a2a-sdk real)
pytest tests/test_a2a_client_tools.py -v    # camada de tools ADK (listar/chamar/consultar)

Environment Variables¶

Variável	Default	Descrição
`ENABLE_A2A`	false	Habilita o servidor A2A como provedor
`ENABLE_A2A_STREAMING`	false	Habilita streaming SSE nas respostas
`A2A_BASE_URL`	—	URL pública para o Agent Card (ex: `https://agents.theifriend.com/a2a`)
`A2A_REQUIRED_ROLE`	ROLE_A2A_USER	Role JWT mínima para acessar os endpoints A2A
`A2A_TASK_TTL_HOURS`	24	TTL de tasks persistidas
`ENABLE_A2A_CLIENT`	false	Habilita as tools de A2A client no root_agent (chamar agentes externos)
`A2A_CLIENT_TIMEOUT`	60	Timeout do cliente A2A em segundos
`A2A_EXTERNAL_AGENTS`	{}	JSON de agentes externos registrados