Servicio del agente A2A

A2AAgent permite a la aplicación conectarse a agentes remotos que se exponen a través del protocolo Agent-to-Agent (A2A). Encapsula cualquier punto de conexión compatible con A2A como estándar AIAgent, por lo que puede usar métodos conocidos como RunAsync e RunStreamingAsync interactuar con agentes remotos independientemente del marco o la tecnología con los que se crearon.

Para exponer un agente de Agent Framework como un servidor A2A, consulte Host agents with A2A (Agentes de host con A2A).

Getting Started

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

Tip

GetAIAgentAsync también acepta un parámetro opcional A2AClientOptions para la selección de protocolo.

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:

Note

El agente A2A remoto debe estar disponible en un punto de conexión que admita el enlace de protocolo seleccionado.

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

Note

Los agentes de A2A admiten la reconexión de flujos (obteniendo la misma secuencia de respuesta desde el principio), no la reanudación de la secuencia desde un punto específico de la secuencia.

Tools

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.

Getting Started

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

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)

A2AAgent almacena el estado del protocolo duradero en AgentSession.service_session_id como una A2AServiceSessionId asignación:

Campo	Tipo	Purpose
context_id	str	Identifica la conversación A2A.
task_id	str \| None	Realiza un seguimiento de la tarea remota más reciente, cuando la respuesta creó una.
task_state	TaskState \| None	Registra el estado de tarea más reciente para que la siguiente solicitud pueda continuar una tarea necesaria para la entrada o hacer referencia a una tarea completada.

Cree una sesión con estado estructurado cuando la aplicación ya conozca el contexto de 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)

También puede empezar con AgentSession() y dejar rellenar A2AAgent la asignación estructurada desde la primera respuesta. Conserve la sesión regular con session.to_dict() y restáurela con AgentSession.from_dict(...); el contexto de A2A, el identificador de tarea y el estado de la tarea permanecen juntos.

Para una tarea en TASK_STATE_INPUT_REQUIRED, el siguiente mensaje establece que task_id continúe con la misma tarea. Para otros estados de tarea, el identificador de tarea anterior se envía a través reference_task_ids para que el agente remoto pueda refinar o continuar desde el resultado anterior.

Authentication

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.

Tools

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.

Go admite agentes A2A remotos a través del provider/a2aprovider paquete.

Instale el marco de trabajo del agente y los paquetes A2A:

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

Conexión a un agente A2A remoto

Resuelva la tarjeta del agente remoto, cree un cliente A2A a partir de ella y encapsula el cliente como agente estándar de 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()

El proveedor almacena los identificadores de tarea y A2A context_id en la sesión de Agent Framework para que los mensajes de seguimiento puedan conservar la continuidad de la conversación.

Selección de protocolo

Si un agente remoto anuncia varios enlaces de transporte, configure el transporte preferido al crear el cliente A2A:

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

Use a2a.TransportProtocolJSONRPC cuando desee preferir JSON-RPC.

Tareas de larga duración

Las tareas de A2A se muestran mediante tokens de continuación de Agent Framework. Inicie la ejecución con una sesión explícita y agent.AllowBackgroundResponses(true), a continuación, sondee mediante una llamada a Run sin nuevos mensajes y el token de continuación:

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 las ejecuciones de streaming interrumpidas, capture update.ContinuationToken de la última actualización recibida y pásela a una ejecución de streaming posterior con agent.WithContinuationToken(token) y agent.Stream(true).

Uso de agentes A2A remotos como herramientas

Resuelva cada agente remoto, encapsulado con a2aprovider.NewAgenty conviértelo en una herramienta 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{}))
}

Tip

Consulte el ejemplo de proveedor A2A y los agentes de A2A como ejemplo de herramientas para ver ejemplos completos de ejecución.

Pasos siguientes

Hosts de agentes con A2A

Vaya más profundamente:

• Agentes personalizados
• especificación del protocolo A2A