A2A-Agent

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

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:

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

Streamen

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

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)

Streamen

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

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)

Wenn Sie A2AAgent.run() mit einem AgentSession aufrufen, leitet der Agent automatisch das A2A context_id von session.service_session_id ab, wenn die ausgehende Nachricht noch keines trägt. Auf diese Weise können Sie die Unterhaltungskontinuität über mehrere A2A-Anrufe hinweg verwalten:

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)

Wenn eine Nachricht einen expliziten context_id in ihrem additional_properties enthält, hat dieser Wert Vorrang vor dem von der Sitzung abgeleiteten Fallback.

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.