Serviço de agente A2A

Permite A2AAgent que seu aplicativo se conecte a agentes remotos expostos por meio do protocolo Agente para Agente (A2A). Ele encapsula qualquer ponto de extremidade compatível com A2A como um padrão AIAgent, para que você possa usar métodos familiares como RunAsync e RunStreamingAsync interagir com agentes remotos, independentemente de qual estrutura ou tecnologia eles foram criados.

Para expor um agente do Agent Framework como um servidor A2A, consulte agentes host com 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!"));

Dica

GetAIAgentAsync também aceita um parâmetro opcional A2AClientOptions para seleção de protocolo.

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:

Note

O agente A2A remoto deve estar disponível em um ponto de extremidade que dê suporte à associação de protocolo selecionada.

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);
        }
    }
}

Note

Os agentes A2A dão suporte à reconexão de fluxo (obtendo o mesmo fluxo de resposta desde o início), não à retomada do fluxo de um ponto específico no fluxo.

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:
    stream = agent.run("Write a short story about a robot.", stream=True)
    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)

A2AAgent armazena o estado do protocolo durável como AgentSession.service_session_id um A2AServiceSessionId mapeamento:

Campo	Tipo	Purpose
context_id	str	Identifica a conversa A2A.
task_id	str \| None	Controla a tarefa remota mais recente, quando a resposta criou uma.
task_state	TaskState \| None	Registra o estado da tarefa mais recente para que a próxima solicitação possa continuar uma tarefa necessária de entrada ou fazer referência a uma tarefa concluída.

Crie uma sessão com estado estruturado quando seu aplicativo já conhece o contexto A2A:

from agent_framework import AgentSession
from agent_framework.a2a import A2AAgent, A2AServiceSessionId

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

    # The A2A message uses context_id="my-conversation-1".
    response = await agent.run("Hello!", session=session)

    # A2AAgent updates task_id and task_state from the response.
    response = await agent.run("Follow-up question", session=session)

Você também pode começar com AgentSession() e deixar A2AAgent preencher o mapeamento estruturado da primeira resposta. Persista a sessão regular com session.to_dict() e restaure-a com AgentSession.from_dict(...); o contexto A2A, a ID da tarefa e o estado da tarefa permanecem juntos.

Para uma tarefa em TASK_STATE_INPUT_REQUIRED, a próxima mensagem define que task_id para continuar a mesma tarefa. Para outros estados de tarefa, a ID da tarefa anterior é enviada para reference_task_ids que o agente remoto possa refinar ou continuar a partir do resultado anterior.

Authentication

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.

O Go dá suporte a agentes A2A remotos por meio do provider/a2aprovider pacote.

Instale o Agent Framework e os pacotes A2A:

go get github.com/microsoft/agent-framework-go
go get github.com/a2aproject/a2a-go/v2

Conectar-se a um agente A2A remoto

Resolva o cartão do agente remoto, crie um cliente A2A a partir dele e encapsule o cliente como um agente padrão do Agent Framework:

import (
    "context"

    "github.com/a2aproject/a2a-go/v2/a2aclient"
    "github.com/a2aproject/a2a-go/v2/a2aclient/agentcard"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/provider/a2aprovider"
)

ctx := context.Background()

card, err := agentcard.DefaultResolver.Resolve(ctx, "http://localhost:5000")
if err != nil {
    panic(err)
}

client, err := a2aclient.NewFromCard(ctx, card)
if err != nil {
    panic(err)
}

a := a2aprovider.NewAgent(
    client,
    a2aprovider.AgentConfig{
        Config: agent.Config{
            Name:        card.Name,
            Description: card.Description,
        },
    },
)

resp, err := a.RunText(ctx, "Hello!").Collect()

O provedor armazena o A2A context_id e as IDs de tarefa na sessão do Agent Framework para que as mensagens de acompanhamento possam preservar a continuidade da conversa.

Seleção de protocolo

Se um agente remoto anunciar várias associações de transporte, configure o transporte preferencial ao criar o cliente A2A:

client, err := a2aclient.NewFromCard(
    ctx,
    card,
    a2aclient.WithConfig(a2aclient.Config{
        PreferredTransports: []a2a.TransportProtocol{a2a.TransportProtocolHTTPJSON},
    }),
)

Use a2a.TransportProtocolJSONRPC quando desejar preferir JSON-RPC.

Tarefas de longa duração

As tarefas A2A são expostas por meio de tokens de continuação do Agent Framework. Inicie a execução com uma sessão explícita e agent.AllowBackgroundResponses(true); depois, consulte chamando Run sem novas mensagens e com o token de continuação:

session, err := a.CreateSession(ctx)
if err != nil {
    panic(err)
}

resp, err := a.RunText(
    ctx,
    "Process this large dataset.",
    agent.WithSession(session),
    agent.AllowBackgroundResponses(true),
).Collect()
if err != nil {
    panic(err)
}

for resp.ContinuationToken != "" {
    resp, err = a.Run(
        ctx,
        nil,
        agent.WithSession(session),
        agent.WithContinuationToken(resp.ContinuationToken),
    ).Collect()
    if err != nil {
        panic(err)
    }
}

Para execuções de streaming interrompidas, capture update.ContinuationToken da última atualização recebida e passe-a para uma execução de streaming posterior com agent.WithContinuationToken(token) e agent.Stream(true).

Usar agentes A2A remotos como ferramentas

Resolva cada agente remoto, embrulhe-o e a2aprovider.NewAgentconverta-o em uma ferramenta com agenttool.New.

tools := make([]tool.Tool, 0, len(agentURLs))

for _, agentURL := range agentURLs {
    card, err := agentcard.DefaultResolver.Resolve(ctx, agentURL)
    if err != nil {
        panic(err)
    }

    client, err := a2aclient.NewFromCard(ctx, card)
    if err != nil {
        panic(err)
    }

    remoteAgent := a2aprovider.NewAgent(client, a2aprovider.AgentConfig{
        Config: agent.Config{
            Name:        card.Name,
            Description: card.Description,
        },
    })

    tools = append(tools, agenttool.New(remoteAgent, agenttool.Config{}))
}

Dica

Consulte o exemplo do provedor A2A e os agentes A2A como exemplo de ferramentas para obter exemplos executáveis completos.

Próximas Etapas 

Agentes de host com A2A

Vá mais fundo:

• Agentes personalizados
• especificação do protocolo A2A