Protocolo A2A

JSON — Agent Card Completo

{
  "name": "Analista Financeiro IA",
  "description": "Especialista em análise de demonstrativos financeiros, KPIs e forecasting",
  "version": "2.1.0",
  "url": "https://agents.minhaempresa.com/financial-analyst",
  "documentationUrl": "https://docs.minhaempresa.com/financial-agent",

  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },

  "authentication": {
    "schemes": ["Bearer"],
    "oauthConfig": {
      "tokenUrl": "https://auth.minhaempresa.com/token",
      "scopes": ["agent:financial:read", "agent:financial:write"]
    }
  },

  "defaultInputModes": ["text/plain", "application/json"],
  "defaultOutputModes": ["application/json", "text/plain"],

  "skills": [
    {
      "id": "analyze_balance_sheet",
      "name": "Análise de Balanço Patrimonial",
      "description": "Analisa balanços e extrai índices financeiros (liquidez, endividamento, rentabilidade)",
      "tags": ["financial", "balance-sheet", "kpi"],
      "examples": [
        "Analise o balanço do Q3 e calcule o índice de liquidez corrente",
        "Compare endividamento dos últimos 4 trimestres"
      ],
      "inputModes": ["application/pdf", "application/json"],
      "outputModes": ["application/json"]
    },
    {
      "id": "forecast_revenue",
      "name": "Previsão de Receita",
      "description": "Gera forecast de receita para próximos N períodos com intervalos de confiança",
      "tags": ["forecast", "revenue", "ml"],
      "examples": [
        "Projete receita para os próximos 6 meses com base no histórico"
      ],
      "inputModes": ["application/json"],
      "outputModes": ["application/json", "text/csv"]
    }
  ]
}
  

JSON — Resposta de Task Concluída

{
  "id": "task-uuid-456",
  "sessionId": "session-789",
  "status": {
    "state": "completed",
    "timestamp": "2026-03-03T14:23:45Z"
  },
  "artifacts": [
    {
      "name": "kpi_analysis",
      "description": "KPIs extraídos do relatório Q4",
      "mimeType": "application/json",
      "parts": [
        {
          "type": "data",
          "data": {
            "receita_liquida": 4250000,
            "margem_ebitda": 0.284,
            "crescimento_yoy": 0.187,
            "nps": 72,
            "churn_rate": 0.031
          }
        }
      ]
    },
    {
      "name": "executive_summary",
      "description": "Resumo executivo em texto",
      "mimeType": "text/plain",
      "parts": [
        {
          "type": "text",
          "text": "Q4 demonstrou crescimento de 18.7% YoY com melhora de margem..."
        }
      ]
    }
  ],
  "history": [
    {"role": "user", "parts": [{"type": "text", "text": "Analise o relatório Q4..."}]},
    {"role": "agent", "parts": [{"type": "text", "text": "Processando PDF..."}]}
  ]
}
  

Python — Servidor A2A (Agente Remoto)

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import uuid, json, asyncio
from pydantic import BaseModel

app = FastAPI()

# Agent Card
AGENT_CARD = {
    "name": "Analista de Texto IA",
    "version": "1.0.0",
    "url": "https://meu-agente.com",
    "capabilities": {"streaming": True},
    "skills": [
        {
            "id": "summarize",
            "name": "Resumir Texto",
            "description": "Gera resumo conciso de textos longos",
            "tags": ["nlp", "summarization"]
        }
    ]
}

# Registry de tasks ativas
tasks: dict = {}

@app.get("/.well-known/agent.json")
async def get_agent_card():
    return AGENT_CARD

@app.post("/tasks/send")
async def send_task(request: dict):
    task_id = request.get("id") or str(uuid.uuid4())
    message = request["params"]["message"]

    # Cria task
    tasks[task_id] = {"state": "submitted", "result": None}

    # Executa (em background)
    asyncio.create_task(process_task(task_id, message))

    # Retorna imediatamente com estado submitted
    return {
        "jsonrpc": "2.0",
        "id": request.get("id"),
        "result": {
            "id": task_id,
            "status": {"state": "working"}
        }
    }

@app.get("/tasks/{task_id}/status")
async def get_task_status(task_id: str):
    if task_id not in tasks:
        raise HTTPException(404, "Task não encontrada")
    return tasks[task_id]

async def process_task(task_id: str, message: dict):
    tasks[task_id]["state"] = "working"
    text = message["parts"][0]["text"]

    # Chama LLM
    summary = await summarize_with_llm(text)

    tasks[task_id] = {
        "state": "completed",
        "artifacts": [{
            "name": "summary",
            "mimeType": "text/plain",
            "parts": [{"type": "text", "text": summary}]
        }]
    }