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

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:

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

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

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)

Quando si chiama A2AAgent.run() con un AgentSession, l'agente deriva automaticamente l'A2A context_id da session.service_session_id se uno non è già presente nel messaggio in uscita. In questo modo è possibile mantenere la continuità della conversazione tra più chiamate 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 un messaggio contiene un valore esplicito context_id nel suo additional_properties, tale valore ha la precedenza sul fallback derivato dalla sessione.

Autenticazione

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.