Alisson's Notes

Pesquisar

SearchSearch

Uso do A2A

01 de jul. de 2025Leitura de 8 min

O protocolo Agent 2 Agent (A2A) tem o foco em basicamente como os agentes se comunicam. Diferentemente do Model Context Protocol (MCP), onde o foco é no uso de ferramentas por um agente.

Imagem ilustrando onde entra cada protocolo: (ref.: https://a2aproject.github.io/A2A/topics/a2a-and-mcp/#how-a2a-and-mcp-complement-each-other)

Arquitetura

Agente remoto (server)

Um agente que vai receber, remotamente, requisições para executar tarefas (busca de produto, avaliação de um texto, etc.). Idealmente esse agente deve ser o mais especializado possível para melhorar a sua assertividade na tarefa (basicamente fazer uma única coisa, bem feita).

AgentCard

A forma de um agente expor para o mundo quem ele é e o que ele faz é por meio de um AgentCard. É nele que é exposto, de forma padronizada, o que faz e as “habilidades” que ele tem disponíveis. Essa exposição é feita por meio de um AgentCard (json padrão) na URL https://{server_domain}/.well-known/agent.json onde o agente está sendo executado.

Um AgentCard possui duas versões, uma “pública” (usuários não autenticados) e outra “privada” (usuários autenticados). Para saber se o agente possui um card extendido, basta olharmos a flag supportsAuthenticatedExtendedCard que o agente possui um card público.

Exemplo de resposta ao acessar o path /.well-known/agent.json (AgentCard público):

{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text"
  ],
  "defaultOutputModes": [
    "text"
  ],
  "description": "Just a hello world agent",
  "name": "Hello World Agent",
  "skills": [
    {
      "description": "just returns hello world",
      "examples": [
        "hi",
        "hello world"
      ],
      "id": "hello_world",
      "name": "Returns hello world",
      "tags": [
        "hello world"
      ]
    }
  ],
  "supportsAuthenticatedExtendedCard": true,
  "url": "http://localhost:9999/",
  "version": "1.0.0"
}

Podemos ver que o agente possui um card extendido para quem tem autenticação supportsAuthenticatedExtendedCard. Para obtê-lo, acessamos o endpoint {AgentCard.url}/../agent/authenticatedExtendedCard (em relação ao URL base especificado no AgentCard público), no nosso exemplo: http://localhost:9999/agent/authenticatedExtendedCard (Nota: não sei se por motivos de especificação do .well-known, mas poderia ser no mesmo diretório do outro AgentCard, não?)

Resposta:

{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text"
  ],
  "defaultOutputModes": [
    "text"
  ],
  "description": "The full-featured hello world agent for authenticated users.",
  "name": "Hello World Agent - Extended Edition",
  "skills": [
    {
      "description": "just returns hello world",
      "examples": [
        "hi",
        "hello world"
      ],
      "id": "hello_world",
      "name": "Returns hello world",
      "tags": [
        "hello world"
      ]
    },
    {
      "description": "A more enthusiastic greeting, only for authenticated users.",
      "examples": [
        "super hi",
        "give me a super hello"
      ],
      "id": "super_hello_world",
      "name": "Returns a SUPER Hello World",
      "tags": [
        "hello world",
        "super",
        "extended"
      ]
    }
  ],
  "supportsAuthenticatedExtendedCard": true,
  "url": "http://localhost:9999/",
  "version": "1.0.1"
}

AgentExecutor

É a porta de entrada do Agente, usado para execução de tarefas solicitas via requisição remota ou local. Trata-se de uma classe que possui a seguinte interface:

  async def execute(self, context: RequestContext, event_queue: EventQueue):
    pass
    
  async def cancel(self, context: RequestContext, event_queue: EventQueue):
    pass

O método execute lida com solicitações recebidas que esperam uma resposta ou um fluxo de eventos. Processa a entrada do usuário (disponível através do contexto RequestContext) e utiliza a fila de eventos (EventQueue) para enviar de volta objetos Message, Task, TaskStatusUpdateEvent ou TaskArtifactUpdateEvent.

O método cancel simplesmente faz o cancelamento de tarefas sendo executadas, útil para tarefas de longa duração (onde o client do Agente precisa fazer pooling de tempos em tempos ou receber uma notificação de conclusão via algum outro canal).

Observações:

  • O objeto RequestContext (A2A - RequestContext) possui uma estrutura simples, apenas com a mensagem “raw” do usuário, não há qualquer indicador sobre qual tarefa (skill) deve ser usada. Isso tudo deve ser feito “manualmente” pelo executor.

Juntando tudo

import uvicorn
 
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import (
    HelloWorldAgentExecutor,
)
 
 
if __name__ == '__main__':
    skill = AgentSkill(
        id='hello_world',
        name='Returns hello world',
        description='just returns hello world',
        tags=['hello world'],
        examples=['hi', 'hello world'],
    )
 
    extended_skill = AgentSkill(
        id='super_hello_world',
        name='Returns a SUPER Hello World',
        description='A more enthusiastic greeting, only for authenticated users.',
        tags=['hello world', 'super', 'extended'],
        examples=['super hi', 'give me a super hello'],
    )
 
    public_agent_card = AgentCard(
        name='Hello World Agent',
        description='Just a hello world agent',
        url='http://localhost:9999/',
        version='1.0.0',
        defaultInputModes=['text'],
        defaultOutputModes=['text'],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],  # Only the basic skill for the public card
        supportsAuthenticatedExtendedCard=True,
    )
 
    specific_extended_agent_card = public_agent_card.model_copy(
        update={
            'name': 'Hello World Agent - Extended Edition',  # Different name for clarity
            'description': 'The full-featured hello world agent for authenticated users.',
            'version': '1.0.1',  # Could even be a different version
            # Capabilities and other fields like url, defaultInputModes, defaultOutputModes,
            # supportsAuthenticatedExtendedCard are inherited from public_agent_card unless specified here.
            'skills': [
                skill,
                extended_skill,
            ],  # Both skills for the extended card
        }
    )
 
    request_handler = DefaultRequestHandler(
        agent_executor=HelloWorldAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )
 
    server = A2AStarletteApplication(
        agent_card=public_agent_card,
        http_handler=request_handler,
        extended_agent_card=specific_extended_agent_card,
    )
 
    uvicorn.run(server.build(), host='0.0.0.0', port=9999)

E o AgentExecutor:

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message
 
class HelloWorldAgent:
    """Hello World Agent."""
 
    async def invoke(self) -> str:
        return 'Hello World'
 
 
class HelloWorldAgentExecutor(AgentExecutor):
    """Test AgentProxy Implementation."""
 
    def __init__(self):
        self.agent = HelloWorldAgent()
 
    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        result = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(result))
 
    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        raise Exception('cancel not supported')

Agente local (client)

Com um agente remoto e AgentCards’s expostos. Podemos consumi-los por outros agentes na rede. Isso pode ser feito simplesmente cadastrando as urls base onde os AgentCards do agente estão expostos via cliente http convencional e incluíndo isso no prompt de instrução (abaixo).

**Role:** You are an expert Routing Delegator. Your primary function is to accurately delegate user inquiries regarding weather or accommodations to the appropriate specialized remote agents.
 
**Core Directives:**
 
* **Task Delegation:** Utilize the `send_message` function to assign actionable tasks to remote agents.
* **Contextual Awareness for Remote Agents:** If a remote agent repeatedly requests user confirmation, assume it lacks access to the full conversation history. In such cases, enrich the task description with all necessary contextual information relevant to that specific agent.
* **Autonomous Agent Engagement:** Never seek user permission before engaging with remote agents. If multiple agents are required to fulfill a request, connect with them directly without requesting user preference or confirmation.
* **Transparent Communication:** Always present the complete and detailed response from the remote agent to the user.
* **User Confirmation Relay:** If a remote agent asks for confirmation, and the user has not already provided it, relay this confirmation request to the user.
* **Focused Information Sharing:** Provide remote agents with only relevant contextual information. Avoid extraneous details.
* **No Redundant Confirmations:** Do not ask remote agents for confirmation of information or actions.
* **Tool Reliance:** Strictly rely on available tools to address user requests. Do not generate responses based on assumptions. If information is insufficient, request clarification from the user.
* **Prioritize Recent Interaction:** Focus primarily on the most recent parts of the conversation when processing requests.
* **Active Agent Prioritization:** If an active agent is already engaged, route subsequent related requests to that agent using the appropriate task update tool.
 
**Agent Roster:**
 
* Available Agents: `{"name": "Airbnb Agent", "description": "Helps with searching accommodation"} {"name": "Weather Agent", "description": "Helps with weather"}`
* Currently Active Seller Agent: `None`

fonte: ^airbnb-planner-multiagent

Nesse exemplo, podemos ver no final que o item Available Agents é um json simples que apenas junta o nome e descrição dos agentes descobertos. Esses dados são obtidos dos AgentCards.

Dessa forma, o LLM apenas usa uma tool (send_message) para determinar qual agente vai responder aquela mensagem.

Um ponto importante que ajuda no fluxo de conversa é informar qual é o agente respondendo atualmente a pessoa. Dessa forma, em caso de dúvidas, o LLM apenas envia para o agente atual e assim a conversa é continuada.

Question

No exemplo que usei aqui (airbnb multiagent) o agente local não envia outras mensagens para o agente remoto, pode ocorrer do agente remoto perder o contexto por conta disso.

Exemplo de código

O código a seguir é um exemplo de um agente local usado no código anterior de hello world. A idéia aqui é mostrar as utilidades que o framework A2A já provê para conexão e leitura de AgentCards.

import logging
 
from typing import Any
from uuid import uuid4
 
import httpx
 
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
)
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)  # Get a logger instance
 
async def main() -> None:
    # Used here only for logging proposes
    PUBLIC_AGENT_CARD_PATH = '/.well-known/agent.json'
    EXTENDED_AGENT_CARD_PATH = '/agent/authenticatedExtendedCard'
 
    base_url = 'http://localhost:9999'
 
    async with httpx.AsyncClient() as httpx_client:
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=base_url,
        )
 
        # Fetch Public Agent Card and Initialize Client
        final_agent_card_to_use: AgentCard | None = None
 
        # check públic
        logger.info(
            f'Attempting to fetch public agent card from: {base_url}{PUBLIC_AGENT_CARD_PATH}'
        )
        _public_card = await resolver.get_agent_card()
        logger.info('Successfully fetched public agent card')
 
        final_agent_card_to_use = _public_card # for now
 
        if _public_card.supportsAuthenticatedExtendedCard:
            # Attempt to fetch the extended card (only for authenticated users)
            auth_headers_dict = {
                'Authorization': 'Bearer dummy-token-for-extended-card'
            }
            _extended_card = await resolver.get_agent_card(
                relative_card_path=EXTENDED_AGENT_CARD_PATH,
                http_kwargs={'headers': auth_headers_dict},
            )
 
            final_agent_card_to_use = _extended_card
 
        client = A2AClient(
            httpx_client=httpx_client, agent_card=final_agent_card_to_use
        )
        logger.info('A2AClient initialized.')
 
        send_message_payload: dict[str, Any] = {
            'message': {
                'role': 'user',
                'parts': [
                    {'kind': 'text', 'text': 'how much is 10 USD in INR?'}
                ],
                'messageId': uuid4().hex,
            },
        }
        request = SendMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )
 
        response = await client.send_message(request)
        print(response.model_dump(mode='json', exclude_none=True))
 
 
if __name__ == '__main__':
    import asyncio
 
    asyncio.run(main())

Referências

airbnb planner multiagent: https://github.com/a2aproject/a2a-samples/tree/main/samples/python/agents/airbnb_planner_multiagent

Visão de gráfico

Backlinks

  • Sem backlinks encontrados