Служба агента A2A

Это A2AAgent позволяет приложению подключаться к удаленным агентам, предоставляемым через протокол "агент — агент" (A2A). Она упаковывает любую конечную точку, совместимую с A2A, как стандартную AIAgent, поэтому вы можете использовать знакомые методы, такие как RunAsync и RunStreamingAsync взаимодействовать с удаленными агентами независимо от того, с какой платформой или технологией они были созданы.

Чтобы предоставить агент Agent Framework в качестве сервера A2A, см. статью "Агенты узла" с помощью A2A.

Getting Started

Добавьте необходимый пакет NuGet в проект:

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

Обнаружение агентов

Прежде чем взаимодействовать с удаленным агентом A2A, необходимо обнаружить его и создать AIAgent экземпляр. Протокол A2A определяет три стратегии обнаружения, каждая из которых поддерживается платформой агента.

URI Well-Known

Агенты A2A могут сделать карточку агента обнаруживаемой по стандартизованному пути: https://{domain}/.well-known/agent-card.json A2ACardResolver Используйте для получения карточки и создания агента в одном вызове:

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 также принимает необязательный A2AClientOptions параметр для выбора протокола.

Обнаружение Catalog-Based

В корпоративных средах или общедоступных платформах карточки агента часто управляются центральным реестром. Если у вас уже есть полученный AgentCard из такого реестра, преобразуйте его непосредственно в 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."));

Прямая конфигурация

Для тесно связанных систем или сценариев разработки, когда конечная точка агента известна заранее, создайте напрямую A2AClient и преобразуйте ее в 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?"));

Выбор протокола

Агенты A2A могут предоставлять несколько привязок протокола, таких как HTTP+JSON и JSON-RPC. По умолчанию для JSON-RPC предпочтительнее http+JSON. Используется A2AClientOptions.PreferredBindings для явного управления используемой привязкой протокола:

Замечание

Удаленный агент A2A должен быть доступен в конечной точке, поддерживающей выбранную привязку протокола.

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

Стриминг

A2A поддерживает потоковую передачу ответов с помощью событий Server-Sent. Используется RunStreamingAsync для получения обновлений в режиме реального времени, так как удаленный агент обрабатывает запрос:

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

Фоновые ответы

Агенты A2A поддерживают фоновые ответы для обработки длительных операций. Когда удаленный агент A2A возвращает задачу вместо немедленного сообщения, платформа агента предоставляет маркер продолжения, который можно использовать для опроса результатов или повторного подключения к прерванным потокам.

Опрос для завершения задачи

Для сценариев без потоковой передачи используйте AllowBackgroundResponses для получения маркера продолжения и опроса до завершения задачи:

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

Повторное подключение потока

В сценариях потоковой передачи каждое обновление может включать маркер продолжения. Если поток прерван, используйте маркер для повторного подключения и получения потока ответа с самого начала:

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

Замечание

Агенты A2A поддерживают повторное подключение потока (получение одного потока отклика с самого начала), а не возобновление потоков из определенной точки в потоке.

Tools

A2AAgent — это оболочка уровня транспорта вокруг удаленного агента A2A. Все средства удаленного агента используются в режиме реального времени на удаленной стороне и невидимы для кода. Типы инструментов Agent Framework (средства функций, интерпретатор кода, поиск файлов, размещенный или локальный MCP и т. д.) не настроены A2AAgent самостоятельно, чтобы расширить возможности удаленного агента, изменить конфигурацию удаленного агента.

Getting Started

Установите пакет A2A:

pip install agent-framework-a2a --pre

Инициализация

A2AAgent можно инициализировать тремя способами в зависимости от того, сколько вы знаете о удаленном агенте заранее.

Прямой URL-адрес

Для разработки или тесно связанных систем, где известна конечная точка:

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)

Если указан только URL-адрес, A2AAgent создается минимальная карточка агента и подключается с помощью JSON-RPC.

Карточка агента

Если у вас есть AgentCard реестр или каталог, передайте его напрямую:

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 При предоставлении по A2AAgent умолчанию name и description из карточки. Он согласовывает транспорт с помощью карточки supported_interfaces.

URI Well-Known (A2ACardResolver)

Используйте A2ACardResolver его для a2a-sdk обнаружения удаленного агента по стандартному известному пути (/.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)

Стриминг

Используется stream=True для получения обновлений в режиме реального времени, так как удаленный агент обрабатывает запрос:

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

длительные задачи

По умолчанию A2AAgent ожидает завершения удаленного агента перед возвратом. Для длительных задач задайте background=True для получения маркера продолжения, который можно использовать для опроса или подписки позже:

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)

Вы также можете повторно подписаться на поток SSE вместо опроса:

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

Идентификатор беседы (context_id)

A2AAgentсохраняет состояние устойчивого A2AServiceSessionId протокола в AgentSession.service_session_id виде сопоставления:

Поле	Тип	Purpose
context_id	str	Определяет беседу A2A.
task_id	str \| None	Отслеживает последнюю удаленную задачу при создании ответа.
task_state	TaskState \| None	Записывает последнее состояние задачи, чтобы следующий запрос продолжал входную задачу или ссылался на завершенную задачу.

Создайте сеанс со структурированным состоянием, когда приложение уже знает контекст 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)

Вы также можете начать с AgentSession() и позволить A2AAgent заполнить структурированное сопоставление из первого ответа. Сохраните регулярный сеанс и session.to_dict() восстановите его с AgentSession.from_dict(...)помощью; контекст A2A, идентификатор задачи и состояние задачи остаются вместе.

Для задачи в TASK_STATE_INPUT_REQUIREDследующем наборе сообщений, которые task_id будут продолжать ту же задачу. Для других состояний задачи идентификатор предыдущей задачи отправляется с reference_task_ids помощью удаленного агента, чтобы уточнить или продолжить от предыдущего результата.

Authentication

Используйте AuthInterceptor для защищенных конечных точек 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!")

Настройка времени ожидания

A2AAgent timeout принимает параметр для управления временем ожидания запроса:

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

Если время ожидания не указано, по умолчанию используются следующие значения: 10-е подключение, 60-е время чтения, 10-е время записи, пул 5s.

Tools

A2AAgent — это оболочка уровня транспорта вокруг удаленного агента A2A. Все средства удаленного агента используются в режиме реального времени на удаленной стороне и невидимы для кода. Типы инструментов Agent Framework (средства функций, интерпретатор кода, поиск файлов, размещенный или локальный MCP и т. д.) не настроены A2AAgent самостоятельно, чтобы расширить возможности удаленного агента, изменить конфигурацию удаленного агента.

Если вы хотите, чтобы агент Foundry вызывал агент A2A в качестве инструмента, см. фабрикуget_a2a_toolFoundryChatClient.

Go поддерживает удаленные агенты A2A через provider/a2aprovider пакет.

Установите пакеты Agent Framework и A2A:

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

Подключение к удаленному агенту A2A

Определите карточку удалённого агента, создайте из неё клиент A2A и представьте этот клиент в виде стандартного агента 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()

Поставщик сохраняет идентификаторы A2A context_id и задач в сеансе Agent Framework, чтобы последующие сообщения могли сохранять непрерывность бесед.

Выбор протокола

Если удаленный агент объявляет несколько привязок транспорта, настройте предпочтительный транспорт при создании клиента A2A:

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

Используйте a2a.TransportProtocolJSONRPC , если вы хотите предпочесть JSON-RPC.

Длительные задачи

Задачи A2A отображаются с помощью маркеров продолжения Agent Framework. Запустите выполнение с явно указанным сеансом и agent.AllowBackgroundResponses(true), затем выполняйте опрос, вызывая Run без новых сообщений и с токеном продолжения:

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

Для прерванных запусков потоковой передачи извлеките update.ContinuationToken из последнего полученного обновления и передайте его в последующий запуск потоковой передачи с agent.WithContinuationToken(token) и agent.Stream(true).

Использование удаленных агентов A2A в качестве инструментов

Разрешите каждый удаленный агент, заключите его в a2aprovider.NewAgentоболочку и преобразуйте его в инструмент.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

Примеры поставщика A2A и агенты A2A см. в качестве примера инструментов для полных примеров запуска.

Дальнейшие действия

Агенты узлов с A2A

Вернитесь глубже:

• Пользовательские агенты
• спецификация протокола A2A