O problema: sistemas REST em um mundo agêntico
As arquiteturas corporativas foram construídas ao longo de anos sobre APIs REST e microsserviços. Esses sistemas são estáveis, bem testados e estão profundamente enraizados em ambientes produtivos. O problema é que eles não foram projetados para o modelo de comunicação Agent-to-Agent (A2A) — o padrão emergente para agentes autônomos que colaboram, raciocinam e se coordenam por meio de mensagens estruturadas.
Com a consolidação do A2A como padrão, as empresas se veem diante de um desafio duplo: não só precisam integrar serviços REST tradicionais a esse novo modelo, como também precisam trazer agentes que já existem — mas foram construídos como APIs REST — para dentro do ecossistema A2A. Reconstruir tudo do zero não é viável. É aí que entra o conceito de agentic overlay.
REST vs. A2A: paradigmas diferentes
Para entender o problema, vale diferenciar os dois modelos. Uma API REST é projetada para integração determinística cliente-servidor: o cliente chama um endpoint bem definido, passa parâmetros e recebe uma resposta previsível, geralmente em um fluxo stateless baseado em HTTP. Isso torna o REST excelente para expor capacidades de negócio com contratos claros e simplicidade operacional.
Já o A2A foi projetado para interoperabilidade entre agentes autônomos. Os agentes se descobrem por meio de metadados (como um “agent card”), negociam capacidades e trocam mensagens estruturadas — frequentemente via JSON-RPC (Notação de Objeto JavaScript para Chamada de Procedimento Remoto) — para coordenar tarefas em múltiplos passos. Enquanto o REST otimiza para interfaces de serviço estáveis e execução direta, o A2A otimiza para coordenação orientada a raciocínio, mensagens orientadas a tarefas e colaboração entre agentes.
A solução: agentic overlays
Um agentic overlay é uma camada fina de encapsulamento que permite que serviços baseados em REST participem de comunicações A2A. Essa camada realiza duas funções principais:
- Transforma mensagens agênticas em payloads REST, e vice-versa.
- Expõe endpoints REST como tarefas ou ferramentas de agente, compatíveis também com o Model Context Protocol (MCP — Protocolo de Contexto de Modelo).
O ponto central é que o A2A não é uma nova API — é uma nova interface para uma API que já existe. O serviço REST subjacente permanece intocado.
Abordagens alternativas (e por que são problemáticas)
Antes de detalhar o overlay, a AWS compara essa abordagem com outras alternativas comuns:
Manter stacks REST e A2A separados
Essa abordagem significa manter dois conjuntos de endpoints (/api/v2/... e /a2a/...), duas implementações de autenticação e validação, dois pipelines de deploy e duplo trabalho de observabilidade. O risco de inconsistência entre os dois caminhos é alto, e o custo operacional cresce com o tempo.
Stacks separados com lógica de negócio compartilhada
Aqui, a ideia é refatorar os endpoints existentes para que a lógica de negócio possa ser reutilizada por uma nova interface A2A. Embora pareça mais limpa, essa abordagem pode introduzir regressões, desvios de comportamento e uma carga de testes considerável — mesmo que os caminhos REST externos permaneçam os mesmos.
Implementando um overlay dentro da aplicação
Nessa abordagem, a aplicação passa a ter dois conjuntos de endpoints (/api/v2/... e /a2a/...), mas mantém um único pipeline de build, teste e deploy. Endpoints REST tradicionais são transformados em endpoints agênticos sem reescrever a lógica central. Novas rotas são adicionadas no mesmo host e porta. As próprias skills do agente podem ser usadas para roteamento interno, sem necessidade de um servidor MCP separado.
A AWS apresenta um exemplo de prova de conceito usando um serviço de calculadora legado em Flask, convertido para o padrão A2A com um overlay. O fluxo de tradução de mensagens A2A funciona assim:
- Recebe requisições JSON-RPC 2.0
- Mapeia tarefas A2A para endpoints REST
- Repassa cabeçalhos de autenticação
- Chama endpoints REST internamente
- Traduz respostas REST para o formato JSON-RPC
Comparação de formatos: REST vs. A2A
Para ilustrar a diferença, veja como a mesma operação de soma é representada nos dois protocolos.
Requisição de entrada — REST vs. A2A:
REST:
{ "operation": "add", "operands": [5, 3] }
A2A:
{
"jsonrpc": "2.0",
"method": "SendMessage",
"params": {
"message": {
"role": "user",
"parts": [
{
"kind": "data",
"data": { "operation": "add", "operands": [5, 3] }
}
]
}
},
"id": 1
}Resposta de saída — REST vs. A2A:
REST:
{"result": 8}
A2A:
{
"jsonrpc": "2.0",
"result": {
"messageId": "uuid",
"contextId": "uuid",
"role": "agent",
"parts": [{"kind": "data", "data": {"result": 8}}],
"kind": "message",
"metadata": {}
},
"id": 1
}Estrutura do overlay em Flask
O overlay é composto por componentes bem definidos. O agent card é construído dinamicamente e expõe as capacidades e skills do agente:
A2A_API_URL = "http://localhost:5000/a2a"
EXECUTE_TIMEOUT_SECONDS = 30
_SKILLS_FILE = Path(__file__).parent / "skills.json"
_SKILLS_CACHE: Optional[List[Dict[str, Any]]] = None
def _load_skills() -> List[Dict[str, Any]]:
global _SKILLS_CACHE
if _SKILLS_CACHE is not None:
return _SKILLS_CACHE
try:
with open(_SKILLS_FILE) as f:
_SKILLS_CACHE = json.load(f)
return _SKILLS_CACHE
except FileNotFoundError:
logger.error(f"Skills file not found: {_SKILLS_FILE}")
return []
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON in skills file: {e}")
return []
def build_agent_card(api_url: Optional[str] = None) -> Dict[str, Any]:
if api_url is None:
api_url = A2A_API_URL
return {
"name": "Calculator Agent",
"description": "Simple calculator supporting basic arithmetic operations",
"supportedInterfaces": [
{"url": api_url, "protocolBinding": "JSONRPC", "protocolVersion": "0.3"},
],
"provider": {"organization": "Example Organization", "url": ""},
"version": "1.0.0",
"capabilities": {
"streaming": False,
"pushNotifications": False,
"extendedAgentCard": False,
},
"defaultInputModes": ["text/plain", "application/json"],
"defaultOutputModes": ["text/plain", "application/json"],
"skills": _load_skills(),
}
AGENT_CARD = build_agent_card()O chamador REST interno repassa autenticação e trata erros de rede:
def invoke_rest_endpoint(
endpoint: str,
json_data: Optional[Dict] = None,
http_method: str = "POST"
) -> Tuple[Optional[Dict], int]:
try:
base_url = request.host_url.rstrip("/")
url = f"{base_url}{endpoint}"
headers = {"Content-Type": "application/json"}
auth_header = request.headers.get("Authorization")
if auth_header:
headers["Authorization"] = auth_header
logger.info(f"Adapter: Delegating to REST {http_method} {url}")
if http_method.upper() == "POST":
response = http_requests.post(url, json=json_data, headers=headers, timeout=EXECUTE_TIMEOUT_SECONDS)
elif http_method.upper() == "GET":
response = http_requests.get(url, headers=headers, timeout=EXECUTE_TIMEOUT_SECONDS)
else:
response = http_requests.request(http_method, url, json=json_data, headers=headers, timeout=EXECUTE_TIMEOUT_SECONDS)
logger.info(f"Adapter: REST returned {response.status_code}")
return response.json(), response.status_code
except http_requests.RequestException as e:
logger.error(f"Adapter: Error calling REST endpoint: {e}", exc_info=True)
return {"error": "Internal server error"}, 500A extração do payload da mensagem A2A e a construção da resposta seguem o formato da Spec 0.3:
def extract_message_payload(message: Dict) -> Optional[Dict]:
try:
parts = message.get("parts", [])
for part in parts:
if isinstance(part, dict) and part.get("kind") == "data":
return part.get("data")
return None
except Exception as e:
logger.error(f"Error extracting message payload: {e}")
return None
def build_a2a_message(message_id: str, context_id: str, content: Any) -> Dict:
if isinstance(content, dict):
parts = [{"kind": "data", "data": content}]
else:
parts = [{"kind": "text", "text": str(content)}]
return {
"messageId": message_id,
"contextId": context_id,
"role": "agent",
"parts": parts,
"kind": "message",
"metadata": {}
}Os construtores de resposta JSON-RPC seguem a especificação 2.0:
class JsonRpcError:
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
def jsonrpc_error(code: int, message: str, data: Any = None, request_id: Any = None) -> Dict:
response = {
"jsonrpc": "2.0",
"error": {"code": code, "message": message},
"id": request_id
}
if data is not None:
response["error"]["data"] = data
return response
def jsonrpc_success(result: Any, request_id: Any = None) -> Dict:
return {
"jsonrpc": "2.0",
"result": result,
"id": request_id
}O handler do método SendMessage delega diretamente para o endpoint REST:
def handle_send_message(data: Dict) -> Tuple[Any, int]:
request_id = data.get("id")
params = data.get("params", {})
message = params.get("message", {})
context_id = message.get("contextId") or generate_id()
message_id = generate_id()
payload = extract_message_payload(message)
if not payload:
return jsonify(jsonrpc_error(
JsonRpcError.INVALID_PARAMS,
"Invalid params: No data found in message.parts.",
request_id=request_id
)), 400
rest_response, status = invoke_rest_endpoint(
endpoint="/api/v1/calculate",
json_data=payload,
http_method="POST"
)
if 200 <= status < 300:
a2a_message = build_a2a_message(message_id, context_id, rest_response)
return jsonify(jsonrpc_success(a2a_message, request_id)), 200
else:
error_message = "Operation failed"
if isinstance(rest_response, dict):
error_message = (rest_response.get("error") or rest_response.get("details") or "Operation failed")
error_code = (JsonRpcError.INVALID_PARAMS if 400 <= status < 500 else JsonRpcError.INTERNAL_ERROR)
return jsonify(jsonrpc_error(
error_code,
error_message,
data=rest_response,
request_id=request_id
)), status
def generate_id() -> str:
return str(uuid.uuid4())As rotas A2A são registradas na aplicação Flask:
def setup_a2a_routes(app: Flask) -> None:
app.add_url_rule("/.well-known/agent-card.json", "get_agent_card", get_agent_card, methods=["GET"])
app.add_url_rule("/a2a/capabilities", "get_capabilities", get_capabilities, methods=["GET"])
app.add_url_rule("/a2a/health", "a2a_health", a2a_health, methods=["GET"])
app.add_url_rule("/a2a", "a2a_jsonrpc", _handle_jsonrpc, methods=["POST"])
logger.info("A2A Protocol v0.3 routes registered")E a inicialização da aplicação une as duas partes:
# app/main.py
from flask import Flask
from app.rest_api import rest_api
from app.a2a_adapter import setup_a2a_routes
def create_app():
app = Flask(__name__)
app.register_blueprint(rest_api)
setup_a2a_routes(app)
app.logger.info("A2A Protocol enabled via Request Translator Pattern")
return appPara rodar a aplicação:
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install -r requirements.txt
python -m app.mainEscala corporativa com Amazon Bedrock AgentCore Gateway
Para cenários de maior escala, a AWS apresenta o Amazon Bedrock AgentCore como alternativa para desacoplar o overlay da aplicação. O AgentCore Runtime atua como ponto de acesso único para endpoints e serviços, permitindo que um único overlay agêntico sirva múltiplos serviços — não apenas um.
O AgentCore Gateway suporta até 10 targets por gateway, com integração nativa a serviços AWS e suporte a endpoints OpenAPI. Isso é especialmente útil em aplicações corporativas que orquestram múltiplos serviços para tarefas complexas: em vez de adicionar um overlay a cada serviço individualmente, é possível unir vários serviços em um único overlay que orquestra as chamadas conforme necessário.
Além do Gateway, o ecossistema AgentCore oferece:
- AgentCore Identity: gerencia autenticação para componentes de agente e gateway, com suporte a provedores OAuth 2.0 e integrações gerenciadas para Okta, GitHub e Slack.
- AgentCore Observability: monitora a performance dos agentes por meio de métricas, logs e visualizações de span. Permite visualizar dados de alto nível como chamadas de ferramentas e latência, ou inspecionar caminhos de execução granulares entre componentes, com integração nativa ao Amazon CloudWatch.
- AgentCore Runtime: realiza o deploy de modelos via imagem de container — sejam open-source, customizados ou Modelos de Linguagem de Grande Escala (LLMs) do Amazon Bedrock como Nova e Anthropic Claude — sem exigir que o time gerencie a infraestrutura de LLM.
Próximos passos recomendados
A AWS sugere três ações práticas para quem quer explorar esse padrão:
- Avalie sua arquitetura: faça um levantamento dos serviços REST candidatos à habilitação A2A. Overlays dentro da aplicação são indicados para agentes de serviço único; o AgentCore Gateway é mais adequado para fluxos multi-serviço.
- Revise a implementação de referência: o exemplo da calculadora em Flask demonstra o padrão de tradução com configuração do agent card, extração de mensagens, invocação REST e construção de respostas.
- Explore o Amazon Bedrock AgentCore: Gateway, Identity e Observability oferecem infraestrutura para overlays agênticos em produção. A especificação e a documentação do SDK do protocolo A2A estão disponíveis em a2a-protocol.org, com bibliotecas para Flask, FastAPI e Starlette.
Conclusão
O padrão de agentic overlay oferece às empresas um caminho pragmático para adotar comunicação Agent-to-Agent sem abandonar os investimentos em APIs REST. A ideia central é simples e poderosa: o A2A não é uma nova API — é uma nova interface para a API que já existe. Essa perspectiva muda o desafio de reconstruir tudo para uma modernização incremental, permitindo que as organizações capturem valor de IA mais rapidamente enquanto gerenciam riscos de forma eficaz.
Fonte
Retrofit, don’t rebuild: Agentic overlays for transforming legacy enterprise services (https://aws.amazon.com/blogs/machine-learning/retrofit-dont-rebuild-agentic-overlays-for-transforming-legacy-enterprise-services/)
Leave a Reply