A2A Agent

L’application A2AAgent permet à votre application de se connecter à des agents distants exposés via le protocole Agent-à-Agent (A2A). Il encapsule tout point de terminaison conforme A2A en tant que norme AIAgent. Vous pouvez donc utiliser des méthodes familières comme RunAsync et RunStreamingAsync interagir avec des agents distants, quelle que soit l’infrastructure ou la technologie avec laquelle ils ont été créés.

Getting Started

Ajoutez le package NuGet requis à votre projet :

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

Découverte de l’agent

Avant de communiquer avec un agent A2A distant, vous devez le découvrir et créer une AIAgent instance. Le protocole A2A définit trois stratégies de découverte, chacune prise en charge par Agent Framework.

URI Well-Known

Les agents A2A peuvent rendre leur carte d’agent détectable à un chemin standardisé : https://{domain}/.well-known/agent-card.json. Utilisez la A2ACardResolver carte pour récupérer la carte et créer un agent dans un seul appel :

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

Conseil / Astuce

GetAIAgentAsync accepte également un paramètre facultatif A2AClientOptions pour la sélection du protocole.

découverte Catalog-Based

Dans les environnements d’entreprise ou les places de marché publiques, les cartes d’agent sont souvent gérées par un registre central. Si vous avez déjà obtenu à AgentCard partir d’un tel registre, convertissez-le directement 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."));

Configuration directe

Pour les systèmes ou les scénarios de développement étroitement couplés où le point de terminaison de l’agent est connu à l’avance, créez un A2AClient système directement et convertissez-le 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?"));

Sélection du protocole

Les agents A2A peuvent exposer plusieurs liaisons de protocole telles que HTTP+JSON et JSON-RPC. Par défaut, HTTP+JSON est préféré à JSON-RPC. Permet A2AClientOptions.PreferredBindings de contrôler explicitement la liaison de protocole utilisée :

Note

L’agent A2A distant doit être disponible sur un point de terminaison qui prend en charge la liaison de protocole sélectionnée.

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

Diffusion en continu

A2A prend en charge la diffusion en continu des réponses via des événements Server-Sent. Permet RunStreamingAsync de recevoir des mises à jour en temps réel, car l’agent distant traite la demande :

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

Réponses en arrière-plan

Les agents A2A prennent en charge les réponses en arrière-plan pour la gestion des opérations de longue durée. Lorsqu’un agent A2A distant retourne une tâche au lieu d’un message immédiat, Agent Framework fournit un jeton de continuation que vous pouvez utiliser pour interroger les résultats ou se reconnecter aux flux interrompus.

Interrogation de l’achèvement de la tâche

Pour les scénarios de non diffusion en continu, utilisez AllowBackgroundResponses cette option pour recevoir un jeton de continuation et interroger jusqu’à ce que la tâche se termine :

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

Reconnexion de flux

Dans les scénarios de streaming, chaque mise à jour peut inclure un jeton de continuation. Si le flux est interrompu, utilisez le jeton pour vous reconnecter et obtenir le flux de réponse à partir du début :

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

Les agents A2A prennent en charge la reconnexion de flux (obtention du même flux de réponse à partir du début), et non la reprise du flux à partir d’un point spécifique dans le flux.

Tools

A2AAgent est un wrapper au niveau du transport autour d’un agent A2A distant. Quels que soient les outils utilisés par l’agent distant sur le côté distant et qui sont invisibles pour votre code. Les types d’outils Agent Framework (outils de fonction, interpréteur de code, recherche de fichiers, MCP hébergé/local, etc.) ne sont pas configurés sur lui-même A2AAgent pour étendre les fonctionnalités de l’agent distant, modifier la configuration de l’agent distant.

Getting Started

Installez le package A2A :

pip install agent-framework-a2a --pre

Initialisation

A2AAgent peut être initialisé de trois façons en fonction de la quantité que vous connaissez à l’avance de l’agent distant.

URL directe

Pour les systèmes de développement ou étroitement couplés où le point de terminaison est connu :

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)

Quand seule une URL est fournie, A2AAgent crée une carte d’agent minimale en interne et se connecte à l’aide de JSON-RPC.

Carte de l’agent

Si vous disposez d’un AgentCard registre ou d’un catalogue, transmettez-le directement :

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)

Lorsqu’un AgentCard est fourni, A2AAgent les name valeurs par défaut et description à partir de la carte. Il négocie le transport à l’aide de supported_interfacesla carte.

URI Well-Known (A2ACardResolver)

Utilisez-le A2ACardResolvera2a-sdk pour découvrir l’agent distant sur le chemin d’accès connu 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)

Diffusion en continu

Permet stream=True de recevoir des mises à jour en temps réel, car l’agent distant traite la demande :

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

Tâches de longue durée

Par défaut, A2AAgent attend que l’agent distant termine avant de retourner. Pour les tâches de longue durée, définissez background=True la surface d’un jeton de continuation que vous pouvez utiliser pour interroger ou s’abonner ultérieurement :

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)

Vous pouvez également vous réinscrire au flux SSE au lieu d’interroger :

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

Identité de conversation (context_id)

Lorsque vous appelez A2AAgent.run() avec un AgentSession, l’agent dérive automatiquement l’A2A context_id à partir de session.service_session_id si le message sortant n’en contient pas déjà un. Cela vous permet de maintenir la continuité des conversations entre plusieurs appels 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 message a un context_id explicite dans son additional_properties, cette valeur est prioritaire sur la valeur par défaut dérivée de la session.

Authentification

Utilisez un AuthInterceptor point de terminaison A2A sécurisé :

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

Configuration du délai d’expiration

A2AAgent accepte un timeout paramètre pour contrôler les délais d’expiration des demandes :

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

Quand aucun délai d’expiration n’est spécifié, les valeurs par défaut sont : 10s connect, 60s read, 10s write, 5s pool.

Tools

A2AAgent est un wrapper au niveau du transport autour d’un agent A2A distant. Quels que soient les outils utilisés par l’agent distant sur le côté distant et qui sont invisibles pour votre code. Les types d’outils Agent Framework (outils de fonction, interpréteur de code, recherche de fichiers, MCP hébergé/local, etc.) ne sont pas configurés sur lui-même A2AAgent pour étendre les fonctionnalités de l’agent distant, modifier la configuration de l’agent distant.

Si vous souhaitez qu’un agent Foundry appelle un agent A2A en tant qu’outil, consultez la get_a2a_tool fabrique sur FoundryChatClient.

Étapes suivantes

Fournisseur personnalisé