Agente A2A

Introducción

Agregue el paquete NuGet necesario al proyecto:

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

Detección de agentes

Antes de comunicarse con un agente A2A remoto, debe detectarlo y crear una AIAgent instancia. El protocolo A2A define tres estrategias de detección, cada una compatible con Agent Framework.

URI de Well-Known

Los agentes de A2A pueden hacer que su tarjeta de agente se pueda detectar en una ruta de acceso estandarizada: https://{domain}/.well-known/agent-card.json. A2ACardResolver Use para capturar la tarjeta y crear un agente en una sola llamada:

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

detección de Catalog-Based

En entornos empresariales o marketplaces públicos, las tarjetas de agente suelen administrarse mediante un registro central. Si ya tiene un AgentCard obtenido de este registro, conviértelo directamente en un 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."));

Configuración directa

Para sistemas estrechamente acoplados o escenarios de desarrollo en los que el punto de conexión del agente se conoce con antelación, cree un directamente A2AClient y conviértelo en :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?"));

Selección de protocolo

Los agentes de A2A pueden exponer varios enlaces de protocolo, como HTTP+JSON y JSON-RPC. De forma predeterminada, se prefiere HTTP+JSON sobre JSON-RPC. Use A2AClientOptions.PreferredBindings para controlar explícitamente qué enlace de protocolo se usa:

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 admite respuestas de streaming a través de eventos de Server-Sent. Use RunStreamingAsync para recibir actualizaciones en tiempo real a medida que el agente remoto procesa la solicitud:

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

Respuestas en segundo plano

Los agentes de A2A admiten respuestas en segundo plano para controlar las operaciones de larga duración. Cuando un agente A2A remoto devuelve una tarea en lugar de un mensaje inmediato, Agent Framework proporciona un token de continuación que puede usar para sondear los resultados o volver a conectarse a secuencias interrumpidas.

Sondeo de finalización de tareas

Para escenarios que no son de streaming, use AllowBackgroundResponses para recibir un token de continuación y sondear hasta que se complete la tarea:

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

Reconexión de secuencias

En escenarios de streaming, cada actualización puede incluir un token de continuación. Si se interrumpe la secuencia, use el token para volver a conectarse y obtener el flujo de respuesta desde el principio:

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

Herramientas

A2AAgent es un contenedor de nivel de transporte alrededor de un agente A2A remoto. Las herramientas que usa el agente remoto residen en el lado remoto y son invisibles para el código. Los tipos de herramientas de Agent Framework (herramientas de función, intérprete de código, búsqueda de archivos, MCP hospedado o local, etc.) no están configurados en sí A2AAgent , para ampliar las funcionalidades del agente remoto, cambiar la configuración del agente remoto.

Introducción

Instale el paquete A2A:

pip install agent-framework-a2a --pre

Inicialización

A2AAgent se puede inicializar de tres maneras en función de la cantidad que conozca sobre el agente remoto con antelación.

Dirección URL directa

Para los sistemas de desarrollo o estrechamente acoplados en los que se conoce el punto de conexión:

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)

Cuando solo se proporciona una dirección URL, A2AAgent crea una tarjeta de agente mínima internamente y se conecta mediante JSON-RPC.

Tarjeta del agente

Si tiene un elemento AgentCard de un registro o catálogo, páselo directamente:

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)

Cuando se proporciona un AgentCard , A2AAgent el name valor predeterminado y description de la tarjeta. Negocia el transporte mediante la tarjeta.supported_interfaces

URI de Well-Known (A2ACardResolver)

Use A2ACardResolver desde a2a-sdk para detectar el agente remoto en la ruta de acceso conocida estándar (/.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 recibir actualizaciones en tiempo real a medida que el agente remoto procesa la solicitud:

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

Tareas de larga duración

De forma predeterminada, A2AAgent espera a que el agente remoto finalice antes de volver. En el caso de las tareas de larga duración, establezca background=True para exponer un token de continuación que puede usar para sondear o suscribirse más adelante:

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)

También puede volver a suscribirse a la secuencia de SSE en lugar de sondear:

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

Identidad de conversación (context_id)

Cuando llamas a A2AAgent.run() con un AgentSession, el agente deriva automáticamente el A2A context_id de session.service_session_id si el mensaje saliente aún no lo lleva. Esto le permite mantener la continuidad de la conversación en varias llamadas 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)

Si un mensaje tiene un explícito context_id en su additional_properties, ese valor tiene precedencia sobre el valor derivado de la sesión.

Autenticación

Use un AuthInterceptor para puntos de conexión 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!")

Configuración de tiempo de espera

A2AAgent acepta un timeout parámetro para controlar los tiempos de espera de solicitud:

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

Cuando no se especifica ningún tiempo de espera, los valores predeterminados son: 10s connect, 60s read, 10s write, 5s pool.

Herramientas

A2AAgent es un contenedor de nivel de transporte alrededor de un agente A2A remoto. Las herramientas que usa el agente remoto residen en el lado remoto y son invisibles para el código. Los tipos de herramientas de Agent Framework (herramientas de función, intérprete de código, búsqueda de archivos, MCP hospedado o local, etc.) no están configurados en sí A2AAgent , para ampliar las funcionalidades del agente remoto, cambiar la configuración del agente remoto.

Si desea que un agente de Foundry llame a un agente de A2A como herramienta, consulte la get_a2a_tool factoría en FoundryChatClient.