Usługa agenta A2A

Funkcja A2AAgent umożliwia aplikacji łączenie się z agentami zdalnymi, którzy są uwidaczniani za pośrednictwem protokołu Agent-to-Agent (A2A). Opakowuje on dowolny punkt końcowy zgodny ze standardem AIAgentA2A, dzięki czemu można używać znanych metod, takich jak RunAsync i RunStreamingAsync do interakcji z agentami zdalnymi niezależnie od struktury lub technologii, z którymi zostały utworzone.

Aby uwidocznić agenta programu Agent Framework jako serwer A2A, zobacz Host agents with A2A (Agenci hosta z usługą A2A).

Wprowadzenie

Dodaj wymagany pakiet NuGet do projektu:

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

Odnajdywanie agentów

Przed komunikacją ze zdalnym agentem A2A należy go odnaleźć i utworzyć AIAgent wystąpienie. Protokół A2A definiuje trzy strategie odnajdywania, z których każda jest obsługiwana przez platformę Agent Framework.

identyfikator URI Well-Known

Agenci A2A mogą wykrywać kartę agenta w ustandaryzowanej ścieżce: https://{domain}/.well-known/agent-card.json. Użyj elementu A2ACardResolver , aby pobrać kartę i utworzyć agenta w jednym wywołaniu:

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

Wskazówka

GetAIAgentAsync Akceptuje również opcjonalny A2AClientOptions parametr wyboru protokołu.

odnajdywanie Catalog-Based

W środowiskach przedsiębiorstwa lub publicznych platformach handlowych karty agentów są często zarządzane przez centralny rejestr. Jeśli masz AgentCard już uzyskany z takiego rejestru, przekonwertuj go bezpośrednio na element 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."));

Konfiguracja bezpośrednia

W przypadku ściśle powiązanych systemów lub scenariuszy programowania, w których punkt końcowy agenta jest znany przed upływem czasu, utwórz bezpośrednio i przekonwertuj A2AClient go na element 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?"));

Wybór protokołu

Agenci A2A mogą uwidaczniać wiele powiązań protokołu, takich jak HTTP+JSON i JSON-RPC. Domyślnie kod HTTP+JSON jest preferowany przez protokół JSON-RPC. Użyj A2AClientOptions.PreferredBindings polecenia , aby jawnie kontrolować, które powiązanie protokołu jest używane:

Note

Zdalny agent A2A musi być dostępny w punkcie końcowym obsługującym wybrane powiązanie protokołu.

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

Usługa A2A obsługuje odpowiedzi przesyłania strumieniowego za pośrednictwem zdarzeń Server-Sent. Użyj RunStreamingAsync polecenia , aby odbierać aktualizacje w czasie rzeczywistym, ponieważ agent zdalny przetwarza żądanie:

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

Odpowiedzi w tle

Agenci A2A obsługują odpowiedzi w tle na potrzeby obsługi długotrwałych operacji. Gdy zdalny agent A2A zwraca zadanie zamiast natychmiastowego komunikatu, struktura agenta udostępnia token kontynuacji, którego można użyć do sondowania wyników lub ponownego nawiązania połączenia ze strumieniami przerwania.

Sondowanie pod kątem ukończenia zadania

W przypadku scenariuszy nieprzesyłania strumieniowego użyj polecenia AllowBackgroundResponses , aby otrzymać token kontynuacji i sondować, dopóki zadanie nie zostanie ukończone:

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

Ponowne nawiązywanie połączenia ze strumieniem

W scenariuszach przesyłania strumieniowego każda aktualizacja może zawierać token kontynuacji. Jeśli strumień zostanie przerwany, użyj tokenu, aby ponownie nawiązać połączenie i uzyskać strumień odpowiedzi od początku:

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

Agenci A2A obsługują ponowne łączenie strumienia (uzyskiwanie tego samego strumienia odpowiedzi od początku), a nie wznowienie strumienia z określonego punktu w strumieniu.

Tools

A2AAgent to otoka na poziomie transportu wokół zdalnego agenta A2A. Niezależnie od narzędzi używanych przez agenta zdalnego po stronie zdalnej i niewidocznych dla kodu. Typy narzędzi platformy agentów (narzędzia funkcji, interpreter kodu, wyszukiwanie plików, hostowane/lokalne mcp itp.) nie są skonfigurowane na A2AAgent samym serwerze — w celu rozszerzenia możliwości agenta zdalnego, zmiany konfiguracji agenta zdalnego.

Wprowadzenie

Zainstaluj pakiet A2A:

pip install agent-framework-a2a --pre

Inicjowanie

A2AAgent Można zainicjować na trzy sposoby, w zależności od tego, ile wiesz o agencie zdalnym z wyprzedzeniem.

Bezpośredni adres URL

W przypadku systemów programistycznych lub ściśle powiązanych, w których jest znany punkt końcowy:

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)

W przypadku podania A2AAgent tylko adresu URL tworzy minimalną kartę agenta wewnętrznie i nawiązuje połączenie przy użyciu protokołu JSON-RPC.

Karta agenta

Jeśli masz rejestr AgentCard lub wykaz, przekaż go bezpośrednio:

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)

Po podaniu AgentCard wartości A2AAgent domyślnych name i description z karty. Negocjuje transport przy użyciu karty supported_interfaces.

identyfikator URI Well-Known (A2ACardResolver)

Użyj A2ACardResolver polecenia , a2a-sdk aby odnaleźć agenta zdalnego w standardowej dobrze znanej ścieżce (/.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

Użyj stream=True polecenia , aby odbierać aktualizacje w czasie rzeczywistym, ponieważ agent zdalny przetwarza żądanie:

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

Zadania długotrwałe

Domyślnie A2AAgent czeka na zakończenie pracy agenta zdalnego przed powrotem. W przypadku długotrwałych zadań ustaw background=True jako token kontynuacji, którego można użyć do sondowania lub subskrybowania później:

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)

Możesz również ponownie przypisać strumień SSE zamiast sondowania:

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

Tożsamość konwersacji (context_id)

A2AAgent przechowuje stan protokołu trwałego w AgentSession.service_session_id programie jako A2AServiceSessionId mapowanie:

Pole	Typ	Purpose
context_id	str	Identyfikuje konwersację A2A.
task_id	str \| None	Śledzi najnowsze zadanie zdalne po utworzeniu odpowiedzi.
task_state	TaskState \| None	Rejestruje najnowszy stan zadania, aby następne żądanie może kontynuować zadanie wymagane przez dane wejściowe lub odwoływać się do ukończonego zadania.

Utwórz sesję ze stanem ustrukturyzowanym, gdy aplikacja zna już kontekst 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)

Możesz również rozpocząć od AgentSession() i pozwolić A2AAgent na wypełnienie mapowania strukturalnego z pierwszej odpowiedzi. Utrwalić zwykłą session.to_dict() sesję i przywrócić ją za pomocą AgentSession.from_dict(...)polecenia ; kontekst A2A, identyfikator zadania i stan zadania pozostają razem.

W przypadku zadania w programie TASK_STATE_INPUT_REQUIREDnastępny komunikat ustawia, który task_id ma kontynuować to samo zadanie. W przypadku innych stanów zadań poprzedni identyfikator zadania jest wysyłany za pośrednictwem reference_task_ids , aby agent zdalny mógł uściślić lub kontynuować z wcześniejszego wyniku.

Authentication

Użyj elementu AuthInterceptor dla zabezpieczonych punktów końcowych A2A:

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

Konfiguracja limitu czasu

A2AAgent akceptuje parametr służący do kontrolowania timeout limitów czasu żądania:

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

Jeśli nie określono limitu czasu, wartości domyślne to: 10s connect, 60s read, 10s write, 5s pool.

Tools

A2AAgent to otoka na poziomie transportu wokół zdalnego agenta A2A. Niezależnie od narzędzi używanych przez agenta zdalnego po stronie zdalnej i niewidocznych dla kodu. Typy narzędzi platformy agentów (narzędzia funkcji, interpreter kodu, wyszukiwanie plików, hostowane/lokalne mcp itp.) nie są skonfigurowane na A2AAgent samym serwerze — w celu rozszerzenia możliwości agenta zdalnego, zmiany konfiguracji agenta zdalnego.

Jeśli chcesz, aby agent usługi Foundry wywołał agenta A2A jako narzędzie, zobacz fabrykęget_a2a_tool w witrynie FoundryChatClient.

Język Go obsługuje zdalnych agentów A2A za pośrednictwem provider/a2aprovider pakietu.

Zainstaluj pakiety Agent Framework i A2A:

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

Nawiązywanie połączenia z zdalnym agentem A2A

Rozwiąż problem z kartą agenta zdalnego, utwórz z niego klienta A2A i opakuj go jako standardowego agenta programu 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()

Dostawca przechowuje identyfikatory A2A context_id i identyfikatory zadań w sesji platformy agentów, dzięki czemu kolejne komunikaty mogą zachować ciągłość konwersacji.

Wybór protokołu

Jeśli agent zdalny anonsuje wiele powiązań transportu, skonfiguruj preferowany transport podczas tworzenia klienta A2A:

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

Użyj a2a.TransportProtocolJSONRPC polecenia , jeśli chcesz preferować kod JSON-RPC.

Długotrwałe zadania

Zadania A2A są udostępniane za pośrednictwem tokenów kontynuacji programu Agent Framework. Rozpocznij uruchomienie z jawną sesją i agent.AllowBackgroundResponses(true), a następnie sprawdzaj stan, wywołując Run bez nowych komunikatów i z tokenem kontynuacji:

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

W przypadku przerwanych sesji strumieniowania zachowaj update.ContinuationToken z ostatnio otrzymanej aktualizacji i przekaż je do kolejnej sesji strumieniowania z użyciem agent.WithContinuationToken(token) i agent.Stream(true).

Używanie zdalnych agentów A2A jako narzędzi

Rozwiąż problemy z każdym agentem zdalnym, opakuj go za pomocą a2aprovider.NewAgentpolecenia i przekonwertuj go na narzędzie za pomocą polecenia 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{}))
}

Wskazówka

Zapoznaj się z przykładowymi agentami dostawcy A2A i agentami A2A jako narzędziami , aby zapoznać się z kompletnymi przykładami z możliwością uruchamiania.

Następne kroki

Agenci hosta z usługą A2A

Głębiej:

• Agenci niestandardowi
• Specyfikacja protokołu A2A