A2A-Agentdienst

Die A2AAgent Anwendung ermöglicht es Ihrer Anwendung, eine Verbindung mit Remote-Agents herzustellen, die über das A2A-Protokoll (Agent-to-Agent) verfügbar gemacht werden. Er umschließt jeden A2A-kompatiblen Endpunkt als Standard AIAgent, sodass Sie vertraute Methoden wie RunAsync und RunStreamingAsync für die Interaktion mit Remote-Agents verwenden können, unabhängig davon, mit welchem Framework oder welcher Technologie sie erstellt wurden.

Informationen zum Verfügbarmachen eines Agent Framework-Agents als A2A-Server finden Sie unter Host-Agents mit A2A.

Getting Started

Fügen Sie dem Projekt das erforderliche NuGet-Paket hinzu:

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

Agent Discovery

Bevor Sie mit einem A2A-Remote-Agent kommunizieren, müssen Sie ihn ermitteln und eine AIAgent Instanz erstellen. Das A2A-Protokoll definiert drei Ermittlungsstrategien, die jeweils vom Agent Framework unterstützt werden.

Well-Known-URI

A2A-Agents können ihre Agent-Karte in einem standardisierten Pfad auffindbar machen: https://{domain}/.well-known/agent-card.json. Verwenden Sie die A2ACardResolver Karte, um die Karte abzurufen und einen Agent in einem einzigen Anruf zu erstellen:

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 akzeptiert auch einen optionalen A2AClientOptions Parameter für die Protokollauswahl.

Catalog-Based Discovery

In Unternehmensumgebungen oder öffentlichen Marketplaces werden Agent-Karten häufig von einer zentralen Registrierung verwaltet. Wenn Sie bereits über eine AgentCard solche Registrierung verfügen, konvertieren Sie sie direkt in eine 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."));

Direkte Konfiguration

Erstellen Sie für eng gekoppelte Systeme oder Entwicklungsszenarien, in denen der Agentendpunkt im Voraus bekannt ist, eine A2AClient direkte Erstellung und Konvertierung in eine 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?"));

Protokollauswahl

A2A-Agents können mehrere Protokollbindungen wie HTTP+JSON und JSON-RPC verfügbar machen. Standardmäßig wird HTTP+JSON gegenüber JSON-RPC bevorzugt. Wird A2AClientOptions.PreferredBindings verwendet, um explizit zu steuern, welche Protokollbindung verwendet wird:

Note

Der Remote-A2A-Agent muss an einem Endpunkt verfügbar sein, der die ausgewählte Protokollbindung unterstützt.

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 unterstützt Streamingantworten über Server-Sent Ereignisse. Wird verwendet RunStreamingAsync , um Updates in Echtzeit zu empfangen, während der Remote-Agent die Anforderung verarbeitet:

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

Hintergrundantworten

A2A-Agents unterstützen Hintergrundantworten für die Behandlung lang ausgeführter Vorgänge. Wenn ein A2A-Remote-Agent eine Aufgabe anstelle einer sofortigen Nachricht zurückgibt, stellt das Agent Framework ein Fortsetzungstoken bereit, mit dem Sie Ergebnisse abfragen oder eine erneute Verbindung mit unterbrochenen Datenströmen herstellen können.

Abrufen des Vorgangsabschlusses

Verwenden Sie AllowBackgroundResponses für Nicht-Streaming-Szenarien, um ein Fortsetzungstoken zu empfangen und abzufragen, bis die Aufgabe abgeschlossen ist:

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

Stream Reconnection

In Streamingszenarien kann jedes Update ein Fortsetzungstoken enthalten. Wenn der Datenstrom unterbrochen wird, verwenden Sie das Token, um die Verbindung wiederherzustellen und den Antwortdatenstrom von Anfang an abzurufen:

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

A2A-Agents unterstützen die Erneute Verbindung des Datenstroms (abrufen desselben Antwortdatenstroms von Anfang an), nicht die Datenstromaufnahme von einem bestimmten Punkt im Datenstrom.

Tools

A2AAgent ist ein Wrapper auf Transportebene um einen Remote-A2A-Agent. Alle Tools, die der Remote-Agent verwendet, sind auf der Remoteseite live und sind für Ihren Code unsichtbar. Agent Framework-Tooltypen (Funktionstools, Codedolmetscher, Dateisuche, gehostet/lokal MCP usw.) sind nicht für sich A2AAgent selbst konfiguriert – um die Funktionen des Remote-Agents zu erweitern, die Konfiguration des Remote-Agents zu ändern.

Getting Started

Installieren Sie das A2A-Paket:

pip install agent-framework-a2a --pre

Initialisierung

A2AAgent kann auf drei Arten initialisiert werden, je nachdem, wie viel Sie über den Remote-Agent im Voraus wissen.

Direkte URL

Für Entwicklung oder eng gekoppelte Systeme, bei denen der Endpunkt bekannt ist:

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)

Wenn nur eine URL bereitgestellt wird, A2AAgent wird intern eine minimale Agentkarte erstellt und eine Verbindung mit JSON-RPC hergestellt.

Agent-Karte

Wenn Sie über eine Registrierung oder einen Katalog verfügen AgentCard , übergeben Sie sie direkt:

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)

Wenn eine AgentCard Bereitgestellt wird, A2AAgent wird name standardmäßig und description von der Karte aus verwendet. Der Transport wird mit der Karte supported_interfacesausgehandelt.

Well-Known URI (A2ACardResolver)

Verwenden Sie A2ACardResolver den Remote-Agent anhand des a2a-sdk standardmäßigen bekannten Pfads (/.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

Wird verwendet stream=True , um Updates in Echtzeit zu empfangen, während der Remote-Agent die Anforderung verarbeitet:

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

Lang laufende Aufgaben

Standardmäßig wartet A2AAgent darauf, dass der Remote-Agent beendet ist, bevor er zurückkehrt. Legen Sie für lange ausgeführte Aufgaben fest, dass ein Fortsetzungstoken angezeigt wird, background=True das Sie zum Abrufen oder Abonnieren später verwenden können:

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)

Sie können den SSE-Datenstrom auch erneut abonnieren, anstatt abfragen zu müssen:

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

Konversations-ID (context_id)

A2AAgent speichert den dauerhaften Protokollstatus AgentSession.service_session_id als A2AServiceSessionId Zuordnung:

Feld	Typ	Purpose
context_id	str	Identifiziert die A2A-Unterhaltung.
task_id	str \| None	Verfolgt die letzte Remoteaufgabe nach, wenn die Antwort erstellt wurde.
task_state	TaskState \| None	Zeichnet den aktuellen Vorgangsstatus auf, damit die nächste Anforderung einen eingaberelevanten Vorgang fortsetzen oder auf einen abgeschlossenen Vorgang verweisen kann.

Erstellen Sie eine Sitzung mit strukturiertem Zustand, wenn Ihre Anwendung den A2A-Kontext bereits kennt:

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)

Sie können auch mit AgentSession() der strukturierten Zuordnung beginnen und die strukturierte Zuordnung aus der ersten Antwort auffüllen lassen A2AAgent . Speichern Sie die normale Sitzung mit session.to_dict() und stellen Sie sie wieder her; AgentSession.from_dict(...)der A2A-Kontext, die Aufgaben-ID und der Aufgabenstatus bleiben zusammen.

Bei einem Vorgang in TASK_STATE_INPUT_REQUIREDder nächsten Meldung wird festgelegt, dass task_id die gleiche Aufgabe fortgesetzt wird. Bei anderen Vorgangszuständen wird die vorherige Vorgangs-ID gesendet reference_task_ids , sodass der Remote-Agent das frühere Ergebnis verfeinern oder fortsetzen kann.

Authentifizierung

Verwenden Sie eine AuthInterceptor für gesicherte A2A-Endpunkte:

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

Timeoutkonfiguration

A2AAgent akzeptiert einen timeout Parameter zum Steuern von Anforderungstimeouts:

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

Wenn kein Timeout angegeben wird, sind die Standardwerte: 10s connect, 60s read, 10s write, 5s pool.

Tools

A2AAgent ist ein Wrapper auf Transportebene um einen Remote-A2A-Agent. Alle Tools, die der Remote-Agent verwendet, sind auf der Remoteseite live und sind für Ihren Code unsichtbar. Agent Framework-Tooltypen (Funktionstools, Codedolmetscher, Dateisuche, gehostet/lokal MCP usw.) sind nicht für sich A2AAgent selbst konfiguriert – um die Funktionen des Remote-Agents zu erweitern, die Konfiguration des Remote-Agents zu ändern.

Wenn Sie möchten, dass ein Foundry-Agent einen A2A-Agent als Tool aufruft, lesen Sie die get_a2a_tool Factory auf FoundryChatClient.

Go unterstützt Remote-A2A-Agents über das provider/a2aprovider Paket.

Installieren Sie die Agent Framework- und A2A-Pakete:

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

Herstellen einer Verbindung mit einem A2A-Remote-Agent

Lösen Sie die Remote-Agent-Karte auf, erstellen Sie einen A2A-Client daraus, und schließen Sie den Client als Standard-Agent-Framework-Agent ein:

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

Der Anbieter speichert die A2A context_id - und Aufgaben-IDs in der Agent Framework-Sitzung, sodass Nachverfolgungsnachrichten die Unterhaltungskontinuität beibehalten können.

Protokollauswahl

Wenn ein Remote-Agent mehrere Transportbindungen ankündigen, konfigurieren Sie den bevorzugten Transport beim Erstellen des A2A-Clients:

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

Verwenden Sie diese Einstellung a2a.TransportProtocolJSONRPC , wenn Sie JSON-RPC bevorzugen möchten.

Lang andauernde Aufgaben

A2A-Aufgaben werden über Agent Framework-Fortsetzungstoken angezeigt. Starten Sie die Ausführung mit einer expliziten Sitzung und agent.AllowBackgroundResponses(true), und fragen Sie dann ab, indem Sie Run ohne neue Nachrichten und mit dem Fortsetzungstoken aufrufen:

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

Bei unterbrochenen Streaming-Durchläufen erfassen Sie update.ContinuationToken aus dem zuletzt empfangenen Update und übergeben update.ContinuationToken in einem späteren Streaming-Durchlauf mit agent.Stream(true) und .

Verwenden Sie Remote-A2A-Agenten als Werkzeuge

Lösen Sie jeden Remote-Agent, umschließen Sie ihn mit a2aprovider.NewAgent, und konvertieren Sie ihn in ein Tool mit 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

Sehen Sie sich das Beispiel für A2A-Anbieter undA2A-Agents als Toolsbeispiel für vollständige Runnable-Beispiele an.

Nächste Schritte

Host-Agents mit A2A

Gehen Sie tiefer:

• Benutzerdefinierte Agents
• A2A-Protokollspezifikation