Plano de Implementação — Protocolo A2A (Agent-to-Agent)¶

Visão Geral¶

Este documento detalha o plano para implementar e evoluir o protocolo A2A no projeto iFriend, permitindo que agentes externos conversem com os agentes iFriend.

Estado Atual (v1.0)¶

✅ Funcionalidades Implementadas¶

Componente	Arquivo	Status
Módulo A2A Core	`runtime/a2a/server.py`	✅ Pronto
Middleware JWT Auth	`A2AJWTAuthMiddleware`	✅ Pronto
Agent Card (8 skills)	`runtime/a2a/server.py:_build_agent_card`	✅ Pronto
Feature Flag	`ENABLE_A2A` em `feature_flags.py`	✅ Pronto
Integração FastAPI	`unified_bot.py:254-271`	✅ Pronto
Testes Unitários	`ifriend_agent/tests/test_a2a.py`	✅ Pronto
Testes de Integração (rotas JSON-RPC reais)	`ifriend_agent/tests/test_a2a_integration.py`	✅ Pronto
Task Persistence (CloudSQL, conectada ao TaskStore real)	`runtime/a2a/task_persistence.py` (`CloudSQLTaskStore`)	✅ Pronto
Streaming (SSE, `message/stream`)	`google.adk.a2a.executor.A2aAgentExecutor.execute` (vendor)	✅ Pronto — funcional ponta a ponta, verificado via a2a-inspector e teste de integração
Cancelamento de task (`tasks/cancel`)	`runtime/a2a/executor.py` (`IfriendA2aAgentExecutor.cancel`)	✅ Pronto (cooperativo/best-effort — não interrompe uma geração já em andamento, mas sempre retorna estado consistente em vez de erro)
Exemplos de Uso	`examples/a2a_client_example.py`, `examples/a2a_test_cli.py`	✅ Pronto
A2A Client (chamar agentes externos, ex: Agentforce)	`runtime/a2a/client.py` + `ifriend_agent/tools/a2a_client_tools.py` (via flag `ENABLE_A2A_CLIENT`)	✅ Pronto — bugs de conformidade de schema corrigidos e validados ponta a ponta com `examples/a2a_echo_agent.py` + `examples/a2a_outbound_roundtrip_test.py`

Correção (2026-09): Task Store real, não apenas advertised¶

Antes desta correção, create_a2a_app() usava InMemoryTaskStore() mesmo com enable_task_persistence=True — a tabela CloudSQL era criada mas nunca lida/escrita. Isso quebrava conversas multi-turn quando a segunda mensagem (reusando taskId) caía em outra instância do Cloud Run (produção roda com max-instances=10), ou após qualquer restart. Agora create_a2a_app() usa CloudSQLTaskStore, que persiste via a2a_task_persistence de fato.

tasks/cancel também era deterministicamente quebrado (A2aAgentExecutor.cancel() do google-adk vendor lança NotImplementedError). IfriendA2aAgentExecutor corrige isso com cancelamento cooperativo: a chamada nunca falha e a task passa a TaskState.canceled para chamadas subsequentes de tasks/get — mas não interrompe uma geração LLM já em andamento no meio (o Runner ADK não expõe esse hook hoje).

Funcionalidades Planejadas¶

Phase 1 — Core A2A (v1.0) — ✅ COMPLETO¶

[x] Exposição do root_agent via JSON-RPC 2.0
[x] Autenticação JWT com role verification
[x] Agent Card público (discovery)
[x] Feature flag ENABLE_A2A
[x] Task persistence (CloudSQL)
[x] Exemplos de cliente

Phase 2 — Streaming & Enhancement (v1.1) — ✅ COMPLETO¶

[x] Streaming capability no AgentCard
[x] SSE streaming endpoint (message/stream, verificado ponta a ponta)
[x] Task cancellation support (cooperativo — ver IfriendA2aAgentExecutor)
[ ] Push notifications setup

Phase 3 — Enterprise Features (v1.2) — 🔄 EM PROGRESSO¶

[ ] Agent Registry (discovery de outros agentes)
[x] A2A Client (chamar agentes externos) — runtime/a2a/client.py + a2a_client_tools, via ENABLE_A2A_CLIENT
[ ] Multi-agent orchestration
[ ] Metrics & observability

Variáveis de Ambiente¶

A2A Protocol¶

Variável	Default	Descrição
`ENABLE_A2A`	`false`	Habilita endpoint A2A
`ENABLE_A2A_STREAMING`	`false`	Habilita streaming SSE
`A2A_BASE_URL`	—	URL pública do endpoint (para AgentCard)
`A2A_REQUIRED_ROLE`	`ROLE_A2A_USER`	Role JWT obrigatória
`A2A_USE_HTTPS`	`false`	Forçar HTTPS no AgentCard
`A2A_TASK_TTL_HOURS`	`24`	TTL de tasks em horas

Arquitetura¶

                    ┌─────────────────┐
                    │  Agente Externo │
                    │   (client A2A)  │
                    └────────┬────────┘
                             │
                             │ HTTPS + JWT (Bearer)
                             │ Authorization: Bearer <token>
                             ▼
┌─────────────────────────────────────────────────────────────┐
│                    FastAPI (unified_bot.py)                │
│                    app.mount("/a2a", a2a_app)              │
└─────────────────────────────────────────────────────────────┘
                             │
              ┌──────────────┴──────────────┐
              │                             │
              ▼                             ▼
    ┌─────────────────┐          ┌─────────────────┐
    │ Agent Card     │          │  JSON-RPC 2.0   │
    │ /.well-known/   │          │  /a2a/          │
    │ agent-card.json│          │                 │
    │ ( público )    │          │  - message/send │
    └─────────────────┘          │  - tasks/get   │
                                 │  - tasks/cancel│
                                 └────────┬────────┘
                                          │
                                          ▼
                                ┌─────────────────┐
                                │  A2AJWTAuth    │
                                │  Middleware    │
                                │  - JWT verify  │
                                │  - Role check  │
                                └────────┬────────┘
                                         │
                                         ▼
                                ┌─────────────────┐
                                │ IfriendA2aAgent │
                                │ Executor (ADK)  │
                                └────────┬────────┘
                                         │
                                         ▼
                                ┌─────────────────┐
                                │  root_agent     │
                                │  (iFriend)      │
                                └─────────────────┘

    ┌─────────────────────────────────────────────┐
    │  Task Persistence (CloudSQL)                │
    │  a2a_tasks table (auto-created)              │
    └─────────────────────────────────────────────┘

Integração no Código¶

Startup (unified_bot.py)¶

if get_flag("ENABLE_A2A"):
    from runtime.a2a.server import create_a2a_app

    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)

Cliente Externo¶

import requests

payload = {
    "jsonrpc": "2.0",
    "id": "1",
    "method": "message/send",
    "params": {
        "message": {
            "role": "user",
            "messageId": "msg-001",
            "parts": [{"kind": "text", "text": "Quero experiências em SP"}]
        }
    }
}

headers = {"Authorization": "Bearer <JWT_WITH_ROLE_A2A_USER>"}

resp = requests.post(
    "https://agents.theifriend.com/trip/a2a/",
    json=payload,
    headers=headers
)

Testes¶

Testes Unitários e de Integração¶

cd ifriend_agent
pytest tests/test_a2a.py -v                # auth middleware + agent card
pytest tests/test_a2a_integration.py -v     # rotas JSON-RPC reais (create_a2a_app via ASGI)
pytest tests/test_a2a_client.py -v          # A2A client outbound (conformidade de schema)
pytest tests/test_a2a_client_tools.py -v    # tools ADK do client outbound

Teste Manual¶

export A2A_JWT_TOKEN="eyJ..."
export A2A_BASE_URL="http://localhost:8080/a2a"
python examples/a2a_client_example.py

Checklist de Deploy¶

[ ] Habilitar ENABLE_A2A=true no environment
[ ] Configurar A2A_BASE_URL (URL pública do agente)
[ ] Garantir que CloudSQL está configurado (para task persistence)
[ ] Verificar que JWT provider emite role ROLE_A2A_USER
[ ] Configurar CORS se agentes externos forem de outros domínios

Próximos Passos¶

Criar Agent Registry: Para discovery de outros agentes
Push notifications: Agent Card já declara pushNotifications: false — implementar se necessário
Adicionar métricas: Monitoring de requests A2A
Multi-agent orchestration: Coordenar múltiplos agentes externos numa mesma conversa