Agente A2A

Introdução

Adicione o pacote NuGet necessário ao seu projeto:

dotnet add package Microsoft.Agents.AI.A2A --prerelease

Descoberta de Agentes

Antes de comunicar com um agente A2A remoto, precisa de o descobrir e criar uma AIAgent instância. O protocolo A2A define três estratégias de descoberta, cada uma suportada pelo Agent Framework.

Well-Known URI

Os agentes A2A podem tornar o seu Cartão de Agente detetável num caminho padronizado: https://{domain}/.well-known/agent-card.json. Use o A2ACardResolver para buscar o cartão e criar um agente numa única chamada:

using A2A;
using Microsoft.Agents.AI;

// Initialize a resolver pointing at the remote agent's host.
A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));

// Resolve the agent card and create an AIAgent in one step.
AIAgent agent = await resolver.GetAIAgentAsync();

// Use the agent.
Console.WriteLine(await agent.RunAsync("Hello!"));

Catalog-Based Descoberta

Em ambientes empresariais ou mercados públicos, os Cartões de Agente são frequentemente geridos por um registo central. Se já tiver um AgentCard obtido de tal registo, converta-o diretamente para um AIAgent:

using A2A;
using Microsoft.Agents.AI;

// Assume agentCard was retrieved from a registry or catalog.
AgentCard agentCard = await GetAgentCardFromRegistryAsync("travel-planner");

AIAgent agent = agentCard.AsAIAgent();

Console.WriteLine(await agent.RunAsync("Plan a trip to Paris."));

Configuração Direta

Para sistemas fortemente acoplados ou cenários de desenvolvimento onde o endpoint do agente é conhecido antecipadamente, crie um A2AClient direto e converta-o para um AIAgent:

using A2A;
using Microsoft.Agents.AI;

// Create a client pointing at the known agent endpoint.
A2AClient a2aClient = new(new Uri("https://a2a-agent.example.com"));

AIAgent agent = a2aClient.AsAIAgent(name: "my-agent", description: "A helpful assistant.");

Console.WriteLine(await agent.RunAsync("What can you help me with?"));

Seleção de Protocolo

Os agentes A2A podem expor múltiplas ligações de protocolo, como HTTP+JSON e JSON-RPC. Por defeito, HTTP+JSON é preferido ao JSON-RPC. Use A2AClientOptions.PreferredBindings para controlar explicitamente qual a ligação de protocolo utilizada:

using A2A;
using Microsoft.Agents.AI;

A2ACardResolver agentCardResolver = new(new Uri("https://a2a-agent.example.com"));

AgentCard agentCard = await agentCardResolver.GetAgentCardAsync();

// Prefer HTTP+JSON protocol binding. For JSON-RPC, set PreferredBindings = [ProtocolBindingNames.JsonRpc]
A2AClientOptions options = new()
{
    PreferredBindings = [ProtocolBindingNames.HttpJson]
};

AIAgent agent = agentCard.AsAIAgent(options: options);

Console.WriteLine(await agent.RunAsync("Tell me a joke about a pirate."));

Streaming

O A2A suporta respostas em streaming através da Server-Sent Events. Use RunStreamingAsync para receber atualizações em tempo real enquanto o agente remoto processa o pedido:

using A2A;
using Microsoft.Agents.AI;

A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));
AIAgent agent = await resolver.GetAIAgentAsync();

await foreach (var update in agent.RunStreamingAsync("Write a short story about a robot."))
{
    if (!string.IsNullOrEmpty(update.Text))
    {
        Console.Write(update.Text);
    }
}

Respostas de Contexto

Os agentes A2A suportam respostas em segundo plano para lidar com operações de longa duração. Quando um agente A2A remoto devolve uma tarefa em vez de uma mensagem imediata, o Agent Framework fornece um token de continuação que pode usar para consultar resultados ou reconectar-se a fluxos interrompidos.

Sondagem para Conclusão de Tarefa

Para cenários que não são streaming, use AllowBackgroundResponses para receber um token de continuação e uma sondagem até que a tarefa seja concluída:

using A2A;
using Microsoft.Agents.AI;

A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));
AIAgent agent = await resolver.GetAIAgentAsync();

AgentSession session = await agent.CreateSessionAsync();

// AllowBackgroundResponses must be true so the server returns immediately with a continuation token
// instead of blocking until the task is complete.
AgentRunOptions options = new() { AllowBackgroundResponses = true };

// Start the initial run with a long-running task.
AgentResponse response = await agent.RunAsync(
    "Conduct a comprehensive analysis of quantum computing applications in cryptography.",
    session,
    options: options);

// Poll until the response is complete.
while (response.ContinuationToken is { } token)
{
    // Wait before polling again.
    await Task.Delay(TimeSpan.FromSeconds(2));

    // Continue with the token.
    response = await agent.RunAsync(session, options: new AgentRunOptions { ContinuationToken = token });
}

Console.WriteLine(response);

Reconexão do Curso de Água

Em cenários de streaming, cada atualização pode incluir um token de continuação. Se o fluxo for interrompido, use o token para reconectar e obter o fluxo de resposta desde o início:

using A2A;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));
AIAgent agent = await resolver.GetAIAgentAsync();

AgentSession session = await agent.CreateSessionAsync();

ResponseContinuationToken? continuationToken = null;

await foreach (var update in agent.RunStreamingAsync(
    "Conduct a comprehensive analysis of quantum computing applications in cryptography.",
    session))
{
    // Save the continuation token to reconnect later if the stream is interrupted.
    // Continuation tokens are only returned for long-running tasks. If the A2A agent
    // returns a message instead of a task, the continuation token will not be initialized.
    if (update.ContinuationToken is { } token)
    {
        continuationToken = token;
    }
}

// If the stream was interrupted and a continuation token was captured,
// reconnect to the response stream using the saved continuation token.
if (continuationToken is not null)
{
    await foreach (var update in agent.RunStreamingAsync(
        session,
        options: new() { ContinuationToken = continuationToken }))
    {
        if (!string.IsNullOrEmpty(update.Text))
        {
            Console.WriteLine(update.Text);
        }
    }
}

Tools

A2AAgent é um wrapper ao nível de transporte em torno de um agente A2A remoto. As ferramentas que o agente remoto usa estão do lado remoto e são invisíveis ao teu código. Os tipos de ferramentas do Agent Framework (ferramentas funcionais, interpretador de código, pesquisa de ficheiros, MCP alojado/local, etc.) não estão configurados A2AAgent diretamente — para estender as capacidades do agente remoto, alterar a configuração do agente remoto.

Introdução

Instale o pacote A2A:

pip install agent-framework-a2a --pre

Inicialização

A2AAgent Pode ser inicializado de três formas, dependendo do quanto sabe sobre o agente remoto antecipadamente.

URL direta

Para desenvolvimento ou sistemas fortemente acoplados onde o ponto final é conhecido:

from agent_framework.a2a import A2AAgent

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    response = await agent.run("Hello!")
    print(response.messages[0].text)

Quando apenas uma URL é fornecida, A2AAgent cria-se internamente um cartão de agente mínimo e liga-se usando JSON-RPC.

Cartão de agente

Se tiveres um AgentCard registo ou catálogo, passa-o diretamente:

from agent_framework.a2a import A2AAgent

async with A2AAgent(agent_card=agent_card) as agent:
    response = await agent.run("Plan a trip to Paris.")
    print(response.messages[0].text)

Quando um AgentCard é fornecido, A2AAgent incumprimento name e description do cartão. Negocia o transporte usando os cartões supported_interfaces.

Well-Known URI (A2ACardResolver)

Use A2ACardResolver a partir do a2a-sdk para descobrir o agente remoto no caminho padrão bem conhecido (/.well-known/agent.json):

import httpx
from a2a.client import A2ACardResolver
from agent_framework.a2a import A2AAgent

async with httpx.AsyncClient(timeout=60.0) as http_client:
    resolver = A2ACardResolver(httpx_client=http_client, base_url="https://a2a-agent.example.com")
    agent_card = await resolver.get_agent_card()

async with A2AAgent(agent_card=agent_card) as agent:
    response = await agent.run("What can you help me with?")
    print(response.messages[0].text)

Streaming

Use stream=True para receber atualizações em tempo real enquanto o agente remoto processa o pedido:

from agent_framework.a2a import A2AAgent

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    async with agent.run("Write a short story about a robot.", stream=True) as stream:
        async for update in stream:
            for content in update.contents:
                if content.text:
                    print(content.text, end="", flush=True)

        final = await stream.get_final_response()
        print(f"\n({len(final.messages)} message(s))")

Tarefas de Longa Duração

Por defeito, A2AAgent espera que o agente remoto termine antes de regressar. Para tarefas de longa duração, defina background=True para mostrar um token de continuação que pode usar para inquirir ou subscrever mais tarde:

from agent_framework.a2a import A2AAgent

async with A2AAgent(name="worker", url="https://a2a-agent.example.com") as agent:
    # Start a long-running task
    response = await agent.run("Process this large dataset", background=True)

    if response.continuation_token:
        # Poll for completion later
        result = await agent.poll_task(response.continuation_token)
        print(result)

Também pode subscrever novamente o fluxo SSE em vez de fazer polling:

# Resubscribe to the task's event stream
response = await agent.run(continuation_token=response.continuation_token)

Identidade de Conversa (context_id)

Quando chama A2AAgent.run() com um AgentSession, o agente determina automaticamente o A2A context_id de session.service_session_id se a mensagem que está a ser enviada ainda não incluir um. Isto permite-lhe manter a continuidade da conversa ao longo de várias chamadas A2A:

from agent_framework import AgentSession
from agent_framework.a2a import A2AAgent

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    session = AgentSession(service_session_id="my-conversation-1")

    # context_id is automatically set to "my-conversation-1"
    response = await agent.run("Hello!", session=session)

    # Subsequent calls with the same session continue the conversation
    response = await agent.run("Follow-up question", session=session)

Se uma mensagem tiver um valor explícito context_id no seu additional_properties, esse valor tem precedência sobre o valor padrão derivado da sessão.

Authentication

Use um AuthInterceptor para endpoints A2A seguros:

from a2a.client.auth.interceptor import AuthInterceptor
from agent_framework.a2a import A2AAgent

class BearerAuth(AuthInterceptor):
    def __init__(self, token: str):
        self.token = token

    async def intercept(self, request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        return request

async with A2AAgent(
    name="secure-agent",
    url="https://secure-a2a-agent.example.com",
    auth_interceptor=BearerAuth("your-token"),
) as agent:
    response = await agent.run("Hello!")

Configuração de Tempo Limite

A2AAgent aceita um timeout parâmetro para controlar os timeouts dos pedidos:

import httpx
from agent_framework.a2a import A2AAgent

# Simple timeout (applies to all components)
async with A2AAgent(name="remote", url="https://example.com", timeout=120.0) as agent:
    ...

# Fine-grained timeout
async with A2AAgent(
    name="remote",
    url="https://example.com",
    timeout=httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=5.0),
) as agent:
    ...

Quando não é especificado o timeout, os valores predefinidos são: 10s conecta, 60s de leitura, 10s de escrita, 5s pool.

Tools

A2AAgent é um wrapper ao nível de transporte em torno de um agente A2A remoto. As ferramentas que o agente remoto usa estão do lado remoto e são invisíveis ao teu código. Os tipos de ferramentas do Agent Framework (ferramentas funcionais, interpretador de código, pesquisa de ficheiros, MCP alojado/local, etc.) não estão configurados A2AAgent diretamente — para estender as capacidades do agente remoto, alterar a configuração do agente remoto.

Se quiser que um agente da Foundry chame um agente A2A como ferramenta, consulte a get_a2a_tool fábrica em FoundryChatClient.