Servizio agente A2A

A2AAgent consente all'applicazione di connettersi agli agenti remoti esposti tramite il protocollo da agente a agente (A2A). Esegue il wrapping di qualsiasi endpoint conforme a A2A come standard AIAgent, in modo da poter usare metodi familiari come RunAsync e RunStreamingAsync per interagire con gli agenti remoti indipendentemente dal framework o dalla tecnologia con cui sono stati creati.

Per esporre un agente di Agent Framework come server A2A, vedere Agenti host con A2A.

Getting Started

Aggiungere il pacchetto NuGet necessario al progetto:

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

Individuazione agenti

Prima di comunicare con un agente A2A remoto, è necessario individuarlo e creare un'istanza AIAgent di . Il protocollo A2A definisce tre strategie di individuazione, ognuna supportata da Agent Framework.

URI Well-Known

Gli agenti A2A possono rendere individuabile la scheda agente in un percorso standardizzato: https://{domain}/.well-known/agent-card.json. A2ACardResolver Usare per recuperare la scheda e creare un agente in una singola chiamata:

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!"));

Suggerimento

GetAIAgentAsync accetta anche un parametro facoltativo A2AClientOptions per la selezione del protocollo.

individuazione Catalog-Based

Negli ambienti aziendali o nei marketplace pubblici, le schede agente vengono spesso gestite da un registro centrale. Se è già AgentCard stato ottenuto da un registro di questo tipo, convertirlo direttamente in :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."));

Configurazione diretta

Per gli scenari di sviluppo o sistemi strettamente associati in cui l'endpoint agente è noto in anticipo, creare un oggetto A2AClient direttamente e convertirlo in :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?"));

Selezione protocollo

Gli agenti A2A possono esporre più associazioni di protocollo, ad esempio HTTP+JSON e JSON-RPC. Per impostazione predefinita, HTTP+JSON è preferibile rispetto a JSON-RPC. Usare A2AClientOptions.PreferredBindings per controllare in modo esplicito quale associazione di protocollo viene usata:

Annotazioni

L'agente A2A remoto deve essere disponibile in un endpoint che supporta l'associazione di protocollo selezionata.

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

A2A supporta lo streaming delle risposte tramite eventi di Server-Sent. Usare RunStreamingAsync per ricevere gli aggiornamenti in tempo reale quando l'agente remoto elabora la richiesta:

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

Risposte in background

Gli agenti A2A supportano le risposte in background per la gestione delle operazioni a esecuzione prolungata. Quando un agente A2A remoto restituisce un'attività anziché un messaggio immediato, Agent Framework fornisce un token di continuazione che è possibile usare per eseguire il polling dei risultati o riconnettersi a flussi interrotti.

Polling per il completamento attività

Per gli scenari non di streaming, usare AllowBackgroundResponses per ricevere un token di continuazione ed eseguire il polling fino al completamento dell'attività:

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

Riconnessione di flusso

Negli scenari di streaming, ogni aggiornamento può includere un token di continuazione. Se il flusso viene interrotto, usare il token per riconnettersi e ottenere il flusso di risposta dall'inizio:

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

Annotazioni

Gli agenti A2A supportano la riconnessione del flusso (ottenendo lo stesso flusso di risposta dall'inizio), non la ripresa del flusso da un punto specifico nel flusso.

Tools

A2AAgent è un wrapper a livello di trasporto intorno a un agente A2A remoto. Qualsiasi strumento usato dall'agente remoto sul lato remoto e sia invisibile al codice. I tipi di strumento di Agent Framework (strumenti per le funzioni, interprete del codice, ricerca di file, MCP ospitato/locale e così via) non sono configurati su A2AAgent se stesso, per estendere le funzionalità dell'agente remoto, modificare la configurazione dell'agente remoto.

Getting Started

Installare il pacchetto A2A:

pip install agent-framework-a2a --pre

Inizializzazione

A2AAgent può essere inizializzato in tre modi a seconda della quantità di informazioni sull'agente remoto in anticipo.

URL diretto

Per sistemi di sviluppo o strettamente associati in cui è noto l'endpoint:

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 viene fornito solo un URL, A2AAgent crea una scheda agente minima internamente e si connette usando JSON-RPC.

Scheda agente

Se si dispone di un oggetto AgentCard da un registro o un catalogo, passarlo direttamente:

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 viene specificato un oggetto AgentCard , A2AAgent le impostazioni predefinite name e description dalla scheda. Negozia il trasporto utilizzando la scheda .supported_interfaces

Well-Known URI (A2ACardResolver)

Usare A2ACardResolver da a2a-sdk per individuare l'agente remoto nel percorso noto standard (/.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

Usare stream=True per ricevere gli aggiornamenti in tempo reale quando l'agente remoto elabora la richiesta:

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

Attività a lungo termine

Per impostazione predefinita, A2AAgent attende che l'agente remoto termini prima di restituire. Per le attività a esecuzione prolungata, impostare per background=True visualizzare un token di continuazione che è possibile usare per eseguire il polling o la sottoscrizione in un secondo momento:

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)

È anche possibile ripetere la sottoscrizione al flusso SSE invece di eseguire il polling:

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

L'Identità della Conversazione (context_id)

A2AAgent archivia lo stato del protocollo durevole in AgentSession.service_session_id come A2AServiceSessionId mapping:

Campo	Tipo	Purpose
context_id	str	Identifica la conversazione A2A.
task_id	str \| None	Tiene traccia dell'attività remota più recente, quando la risposta ne ha creata una.
task_state	TaskState \| None	Registra lo stato dell'attività più recente in modo che la richiesta successiva possa continuare un'attività richiesta di input o fare riferimento a un'attività completata.

Creare una sessione con stato strutturato quando l'applicazione conosce già il contesto 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)

È anche possibile iniziare con AgentSession() e consentire A2AAgent di popolare il mapping strutturato dalla prima risposta. Rendere persistente la sessione regolare con session.to_dict() e ripristinarla con AgentSession.from_dict(...). Il contesto A2A, l'ID attività e lo stato dell'attività rimangono insieme.

Per un'attività in TASK_STATE_INPUT_REQUIRED, il messaggio successivo imposta che task_id per continuare la stessa attività. Per altri stati dell'attività, l'ID attività precedente viene inviato tramite reference_task_ids in modo che l'agente remoto possa perfezionare o continuare dal risultato precedente.

Authentication

Usare un AuthInterceptor per gli endpoint A2A protetti:

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!")

Configurazione del timeout

A2AAgent accetta un timeout parametro per controllare i timeout delle richieste:

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 non viene specificato alcun timeout, le impostazioni predefinite sono: 10s connect, 60s read, 10s write, 5s pool.

Tools

A2AAgent è un wrapper a livello di trasporto intorno a un agente A2A remoto. Qualsiasi strumento usato dall'agente remoto sul lato remoto e sia invisibile al codice. I tipi di strumento di Agent Framework (strumenti per le funzioni, interprete del codice, ricerca di file, MCP ospitato/locale e così via) non sono configurati su A2AAgent se stesso, per estendere le funzionalità dell'agente remoto, modificare la configurazione dell'agente remoto.

Se si vuole che un agente Foundry chiami un agente A2A come strumento, vedere la get_a2a_tool factory in FoundryChatClient.

Go supporta gli agenti A2A remoti tramite il provider/a2aprovider pacchetto.

Installare i pacchetti Agent Framework e A2A:

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

Connettersi a un agente A2A remoto

Risolvere la scheda dell'agente remoto, creare un client A2A ed eseguire il wrapping del client come agente di Agent Framework standard:

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

Il provider archivia gli ID attività e A2A context_id nella sessione di Agent Framework in modo che i messaggi di completamento possano mantenere la continuità della conversazione.

Selezione del protocollo

Se un agente remoto annuncia più associazioni di trasporto, configurare il trasporto preferito durante la creazione del client A2A:

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

Usare a2a.TransportProtocolJSONRPC quando si vuole preferire JSON-RPC.

Attività a esecuzione prolungata

Le attività A2A emergono tramite i token di continuazione di Agent Framework. Avviare l'esecuzione con una sessione esplicita e agent.AllowBackgroundResponses(true), quindi eseguire il polling chiamando Run senza nuovi messaggi e il token di continuazione:

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

Per le esecuzioni di streaming interrotte, acquisire update.ContinuationToken dall'ultimo aggiornamento ricevuto e passarlo a un'esecuzione di streaming successiva con agent.WithContinuationToken(token) e agent.Stream(true).

Usare agenti A2A remoti come strumenti

Risolvere ogni agente remoto, eseguirne il wrapping con a2aprovider.NewAgente convertirlo in uno strumento con 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{}))
}

Suggerimento

Per esempi completi eseguibili, vedere l'esempio di provider A2A e gli agenti A2A come strumenti .

Passaggi successivi

Agenti host con A2A

Approfondimento:

• Agenti personalizzati
• Specifica del protocollo A2A