Služba agenta A2A

Umožňuje A2AAgent vaší aplikaci připojit se ke vzdáleným agentům, kteří jsou vystaveni prostřednictvím protokolu Agent-to-Agent (A2A). Zabalí každý koncový bod kompatibilní s A2A jako standard AIAgent, takže můžete používat známé metody jako RunAsync a RunStreamingAsync pracovat se vzdálenými agenty bez ohledu na to, s jakou architekturou nebo technologií byly vytvořeny.

Pokud chcete vystavit agenta Agent Framework jako server A2A, přečtěte si téma Agenti hostitelů s A2A.

Začínáme

Přidejte do projektu požadovaný balíček NuGet:

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

Zjišťování agentů

Před komunikací se vzdáleným agentem A2A ho musíte zjistit a vytvořit AIAgent instanci. Protokol A2A definuje tři strategie zjišťování, které podporuje rozhraní Agent Framework.

identifikátor URI Well-Known

Agenti A2A mohou zjistit svou kartu agenta na standardizované cestě: https://{domain}/.well-known/agent-card.json. A2ACardResolver Použijte k načtení karty a vytvoření agenta v jednom volání:

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 také přijímá volitelný A2AClientOptions parametr pro výběr protokolu.

zjišťování Catalog-Based

V podnikových prostředích nebo veřejných marketplacech se karty agentů často spravují centrálním registrem. Pokud již máte AgentCard získaný z takového registru, převeďte ho přímo na 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."));

Přímá konfigurace

V případě úzce propojených systémů nebo vývojových scénářů, ve kterých je koncový bod agenta známý předem, vytvořte A2AClient přímo a převeďte ho na 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?"));

Výběr protokolu

Agenti A2A můžou vystavit více vazeb protokolu, jako jsou HTTP+JSON a JSON-RPC. Ve výchozím nastavení se upřednostňují protokoly HTTP+JSON před protokolem JSON-RPC. Slouží A2AClientOptions.PreferredBindings k explicitnímu řízení použité vazby protokolu:

Note

Vzdálený agent A2A musí být dostupný v koncovém bodu, který podporuje vybranou vazbu protokolu.

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 podporuje streamování odpovědí prostřednictvím Server-Sent Událostí. Umožňuje RunStreamingAsync přijímat aktualizace v reálném čase, protože vzdálený agent zpracovává požadavek:

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

Odezvy na pozadí

Agenti A2A podporují odpovědi na pozadí pro zpracování dlouhotrvajících operací. Když vzdálený agent A2A místo okamžité zprávy vrátí úlohu, poskytuje rozhraní agenta token pokračování, který můžete použít k dotazování výsledků nebo opětovné připojení k přerušeným datovým proudům.

Dotazování na dokončení úkolu

V případě scénářů bez streamování použijte AllowBackgroundResponses k získání tokenu pokračování a dotazování, dokud se úloha nedokončí:

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

Opětovné připojení streamu

Ve scénářích streamování může každá aktualizace obsahovat token pro pokračování. Pokud dojde k přerušení datového proudu, pomocí tokenu se znovu připojte a získejte stream odpovědí od začá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

Agenti A2A podporují opětovné připojení streamu (získání stejného streamu odpovědí od začátku), ne obnovení streamu z konkrétního bodu v datovém proudu.

Tools

A2AAgent je obálka na úrovni přenosu kolem vzdáleného agenta A2A. Jakékoli nástroje, které vzdálený agent používá, se nacházejí na vzdálené straně a nejsou pro váš kód neviditelné. Typy nástrojů agenta Framework (nástroje funkcí, interpret kódu, vyhledávání souborů, hostované nebo místní MCP atd.) nejsou nakonfigurované na A2AAgent sobě – aby se rozšířily možnosti vzdáleného agenta, změnily konfiguraci vzdáleného agenta.

Začínáme

Nainstalujte balíček A2A:

pip install agent-framework-a2a --pre

Inicializace

A2AAgent lze inicializovat třemi způsoby v závislosti na tom, kolik víte o vzdáleném agentu předem.

Přímá adresa URL

U vývojových nebo úzce propojených systémů, kde je koncový bod známý:

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)

Pokud je k dispozici pouze adresa URL, A2AAgent vytvoří interně minimální kartu agenta a připojí se pomocí JSON-RPC.

Karta agenta

Pokud máte registr AgentCard nebo katalog, předejte ho přímo:

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)

AgentCard Pokud je zadaný, A2AAgent použije se výchozí hodnota name a description karta. Vyjedná dopravu pomocí karty supported_interfaces.

identifikátor URI Well-Known (A2ACardResolver)

Pomocí A2ACardResolver příkazu from the a2a-sdk discover the remote agent at the standard well-known path (/.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

Umožňuje stream=True přijímat aktualizace v reálném čase, protože vzdálený agent zpracovává požadavek:

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

Dlouhotrvající úkoly

Ve výchozím nastavení A2AAgent čeká, až vzdálený agent dokončí svou činnost, než se vrátí. V případě dlouhotrvajících úloh nastavte background=True možnost zobrazit token pro pokračování, který můžete použít k dotazování nebo přihlášení k odběru později:

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)

Místo dotazování můžete datový proud SSE znovu odeslat:

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

Identita kontextu konverzace (context_id)

A2AAgent ukládá trvalý stav AgentSession.service_session_id protokolu jako A2AServiceSessionId mapování:

Pole	Typ	Purpose
context_id	str	Identifikuje konverzaci A2A.
task_id	str \| None	Sleduje poslední vzdálenou úlohu při vytvoření odpovědi.
task_state	TaskState \| None	Zaznamenává nejnovější stav úkolu, aby další požadavek mohl pokračovat ve vstupním požadovaném úkolu nebo odkazovat na dokončený úkol.

Vytvořte relaci se strukturovaným stavem, když už vaše aplikace zná kontext 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)

Můžete také začít a AgentSession() nechat A2AAgent naplnit strukturované mapování z první odpovědi. Zachovají běžnou relaci a session.to_dict() obnovují ji AgentSession.from_dict(...)pomocí kontextu A2A, ID úkolu a stavu úkolu.

U úkolu v TASK_STATE_INPUT_REQUIREDnásledující zprávě se nastaví, aby task_id pokračoval ve stejném úkolu. V případě jiných stavů úloh se předchozí ID úlohy odešle, reference_task_ids aby vzdálený agent mohl upřesnit nebo pokračovat z dřívějšího výsledku.

Authentication

Použijte AuthInterceptor pro zabezpečené A2A koncové body:

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

Konfigurace časového limitu

A2AAgent timeout přijímá parametr pro řízení časových limitů požadavků:

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

Pokud není zadaný žádný časový limit, výchozí hodnoty jsou: 10s connect, 60s read, 10s write, 5s pool.

Tools

A2AAgent je obálka na úrovni přenosu kolem vzdáleného agenta A2A. Jakékoli nástroje, které vzdálený agent používá, se nacházejí na vzdálené straně a nejsou pro váš kód neviditelné. Typy nástrojů agenta Framework (nástroje funkcí, interpret kódu, vyhledávání souborů, hostované nebo místní MCP atd.) nejsou nakonfigurované na A2AAgent sobě – aby se rozšířily možnosti vzdáleného agenta, změnily konfiguraci vzdáleného agenta.

Pokud chcete, aby agent Foundry volal agentA A2A jako nástroj, podívejte se na továrnuget_a2a_tool na .FoundryChatClient

Go podporuje vzdálené agenty A2A prostřednictvím provider/a2aprovider balíčku.

Nainstalujte balíčky Agent Framework a A2A:

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

Připojení ke vzdálenému agentovi A2A

Vyřešte kartu vzdáleného agenta, vytvořte z ní klienta A2A a zabalte klienta jako standardního agenta 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()

Poskytovatel ukládá ID úloh a A2A context_id v relaci rozhraní agenta, aby následné zprávy mohly zachovat kontinuitu konverzací.

Výběr protokolu

Pokud vzdálený agent inzeruje více přenosových vazeb, nakonfigurujte upřednostňovaný přenos při vytváření klienta A2A:

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

Použijte a2a.TransportProtocolJSONRPC , když chcete preferovat JSON-RPC.

Dlouhotrvající úlohy

Úlohy A2A se zobrazují prostřednictvím tokenů pokračování v rámci Agent Frameworku. Spusťte běh s explicitní relací a agent.AllowBackgroundResponses(true), poté se dotazujte voláním Run bez nových zpráv a s tokenem pokračová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)
    }
}

U přerušených běhů streamování zachyťte update.ContinuationToken z poslední přijaté aktualizace a předejte jej do pozdějšího běhu streamování pomocí agent.WithContinuationToken(token) a agent.Stream(true).

Použití vzdálených agentů A2A jako nástrojů

Vyřešte jednotlivé vzdálené agenty, zabalte ho a2aprovider.NewAgenta převeďte ho na nástroj s 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

Kompletní příklady spustitelných příkladů najdete v ukázce zprostředkovatele A2A a agentech A2A.

Další kroky

Hostování agentů s A2A

Jděte hlouběji:

• Speciální agenti
• Specifikace protokolu A2A