Serviço de agente A2A

Permite A2AAgent que a sua aplicação se ligue a agentes remotos expostos através do protocolo Agent-to-Agent (A2A). Envolve qualquer endpoint compatível com A2A como padrão AIAgent, para que possa usar métodos familiares como RunAsync e RunStreamingAsync interagir com agentes remotos, independentemente da estrutura ou tecnologia com que foram construídos.

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

Tip

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

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:

Observação

O agente A2A remoto deve estar disponível num endpoint que suporte a ligaçã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 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);
        }
    }
}

Observação

Os agentes A2A suportam a reconexão do fluxo (obtendo o mesmo fluxo de resposta desde o início), não a retomada do fluxo a partir de um ponto específico do fluxo.

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:
    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 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)

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

Campo	Tipo	Purpose
context_id	str	Identifica a conversa A2A.
task_id	str \| None	Regista a tarefa remota mais recente, quando a resposta criou uma.
task_state	TaskState \| None	Regista o estado mais recente da tarefa para que o próximo pedido possa continuar uma tarefa necessária ou referenciar uma tarefa concluída.

Crie uma sessão com estado estruturado quando a sua aplicação 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)

Também podes começar com AgentSession() e deixar A2AAgent preencher o mapeamento estruturado a partir da primeira resposta. Persistir a sessão regular com session.to_dict() e restaurá-la com AgentSession.from_dict(...); o contexto A2A, ID da tarefa e estado da tarefa mantêm-se juntos.

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

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.

O Go suporta agentes A2A remotos através 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

Liga-te a um agente A2A remoto

Resolve o cartão de agente remoto, cria um cliente A2A a partir dele e envolve 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 fornecedor armazena os IDs A2A context_id e de tarefa na sessão do Agent Framework, para que as mensagens de seguimento possam preservar a continuidade da conversa.

Seleção do protocolo

Se um agente remoto anunciar múltiplas associações de transporte, configure o transporte preferido ao criar o cliente A2A:

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

Usa a2a.TransportProtocolJSONRPC quando quiseres preferir JSON-RPC.

Tarefas de longa duração

As tarefas A2A são apresentadas através 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 usando 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-o numa execução de streaming posterior com agent.WithContinuationToken(token) e agent.Stream(true).

Use agentes A2A remotos como ferramentas

Resolve cada agente remoto, envolve-o com a2aprovider.NewAgent, e converte-o numa 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{}))
}

Tip

Veja a amostra do fornecedor A2A e os agentes A2A como ferramentas para exemplos completos executáveis.

Passos seguintes

Agentes anfitriões com A2A

Vai mais fundo:

• Agentes aduaneiros
• Especificação do protocolo A2A