Agente A2A

Introdução

Adicione o pacote NuGet necessário ao seu projeto:

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

Descoberta de agente

Antes de se comunicar com um agente A2A remoto, você precisa descobri-lo e criar uma AIAgent instância. O protocolo A2A define três estratégias de descoberta, cada uma com suporte do Agent Framework.

URI de Well-Known

Os agentes A2A podem tornar o Cartão do Agente detectável em um caminho padronizado: https://{domain}/.well-known/agent-card.json. Use o A2ACardResolver para buscar o cartão e criar um agente em uma ú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!"));

Descoberta de Catalog-Based

Em ambientes corporativos ou marketplaces públicos, os Cartões de Agente geralmente são gerenciados por um registro central. Se você já tiver obtido AgentCard um registro desse tipo, converta-o diretamente em 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 em que o ponto de extremidade do agente é conhecido antecipadamente, crie um A2AClient diretamente e converta-o em 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 várias associações de protocolo, como HTTP+JSON e JSON-RPC. Por padrão, HTTP+JSON é preferencial em vez de JSON-RPC. Use A2AClientOptions.PreferredBindings para controlar explicitamente qual associação de protocolo é usada:

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 dá suporte a respostas de streaming por meio de eventos Server-Sent. Use RunStreamingAsync para receber atualizações em tempo real à medida que o agente remoto processa a solicitação:

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 em segundo plano

Os agentes A2A dão suporte a respostas em segundo plano para lidar com operações de execução longa. Quando um agente A2A remoto retorna uma tarefa em vez de uma mensagem imediata, o Agent Framework fornece um token de continuação que você pode usar para sondar resultados ou reconectar a fluxos interrompidos.

Sondagem para conclusão da tarefa

Para cenários que não são de streaming, use AllowBackgroundResponses para receber um token de continuação e sondar 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 de fluxo

Em cenários de streaming, cada atualização pode incluir um token de continuação. Se o fluxo for interrompido, use o token para se 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 de nível de transporte em torno de um agente A2A remoto. Quaisquer que sejam as ferramentas que o agente remoto usa vivem no lado remoto e são invisíveis para seu código. Os tipos de ferramentas do Agent Framework (ferramentas de função, interpretador de código, pesquisa de arquivos, MCP hospedado/local etc.) não são configurados por A2AAgent si só – para estender os recursos 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 maneiras, dependendo do quanto você sabe sobre o agente remoto com antecedência.

URL Direta

Para desenvolvimento ou sistemas firmemente acoplados em que o ponto de extremidade é 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 um cartão de agente mínimo internamente e se conecta usando JSON-RPC.

Cartão do Agente

Se você tiver um AgentCard registro ou catálogo, passe-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 o padrão name e description do cartão. Ele negocia o transporte usando o cartão.supported_interfaces

Well-Known URI (A2ACardResolver)

Use A2ACardResolver do a2a-sdk para descobrir o agente remoto no caminho padrão 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 à medida que o agente remoto processa a solicitação:

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 padrão, A2AAgent aguarda o agente remoto concluir suas operações antes de retornar. Para tarefas de execução prolongada, defina background=True para exibir um token de continuação que você pode usar para sondar ou assinar 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)

Você também pode assinar novamente o fluxo SSE em vez de sondar:

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

Identidade da conversa (context_id)

Quando chama A2AAgent.run() com um AgentSession, o agente deriva automaticamente o A2A context_id da mensagem de saída de session.service_session_id se ainda não houver um. Isso permite manter a continuidade da conversa em 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 context_id explícito em seu additional_properties, esse valor terá precedência sobre o fallback derivado da sessão.

Autenticação

Use um AuthInterceptor para pontos de extremidade A2A protegidos.

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 tempos limite de solicitação:

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 nenhum tempo limite é especificado, os padrões são: 10s connect, 60s read, 10s write, 5s pool.

Tools

A2AAgent é um wrapper de nível de transporte em torno de um agente A2A remoto. Quaisquer que sejam as ferramentas que o agente remoto usa vivem no lado remoto e são invisíveis para seu código. Os tipos de ferramentas do Agent Framework (ferramentas de função, interpretador de código, pesquisa de arquivos, MCP hospedado/local etc.) não são configurados por A2AAgent si só – para estender os recursos do agente remoto, alterar a configuração do agente remoto.

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