Udostępnij przez

Integracja usługi A2A

Protokół Agent-to-Agent (A2A) umożliwia ustandaryzowaną komunikację między agentami, umożliwiając agentom skompilowaną przy użyciu różnych struktur i technologii bezproblemowe komunikowanie się.

Co to jest A2A?

A2A to standardowy protokół obsługujący:

  • Odnajdywanie agentów za pośrednictwem kart agentów
  • Komunikacja oparta na komunikatach między agentami
  • Długotrwałe procesy agentyczne poprzez zadania
  • Współdziałanie międzyplatformowe między różnymi frameworkami agentów

Aby uzyskać więcej informacji, zobacz specyfikację protokołu A2A.

Biblioteka Microsoft.Agents.AI.Hosting.A2A.AspNetCore zapewnia integrację ASP.NET Core na potrzeby uwidaczniania agentów za pośrednictwem protokołu A2A.

Pakiety NuGet:

Example

W tym minimalnym przykładzie pokazano, jak uwidocznić agenta za pośrednictwem usługi A2A. Przykład obejmuje zależności OpenAPI i Swagger w celu uproszczenia testowania.

1. Tworzenie projektu internetowego interfejsu API platformy ASP.NET Core

Utwórz nowy projekt internetowego interfejsu API platformy ASP.NET Core lub użyj istniejącego.

2. Zainstaluj wymagane zależności

Zainstaluj następujące pakiety:

Uruchom następujące polecenia w katalogu projektu, aby zainstalować wymagane pakiety NuGet:

# Hosting.A2A.AspNetCore for A2A protocol integration
dotnet add package Microsoft.Agents.AI.Hosting.A2A.AspNetCore --prerelease

# Libraries to connect to Azure OpenAI
dotnet add package Azure.AI.OpenAI --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Extensions.AI
dotnet add package Microsoft.Extensions.AI.OpenAI --prerelease

# Swagger to test app
dotnet add package Microsoft.AspNetCore.OpenApi
dotnet add package Swashbuckle.AspNetCore

3. Konfigurowanie połączenia azure OpenAI

Aplikacja wymaga połączenia azure OpenAI. Skonfiguruj punkt końcowy i nazwę wdrożenia przy użyciu dotnet user-secrets lub zmiennych środowiskowych. Można również po prostu edytować element appsettings.json, ale nie jest to zalecane dla aplikacji wdrożonych w środowisku produkcyjnym, ponieważ niektóre dane mogą być uważane za tajne.

dotnet user-secrets set "AZURE_OPENAI_ENDPOINT" "https://<your-openai-resource>.openai.azure.com/"
dotnet user-secrets set "AZURE_OPENAI_DEPLOYMENT_NAME" "gpt-4o-mini"

4. Dodaj kod do Program.cs

Zastąp zawartość Program.cs pliku następującym kodem i uruchom aplikację:

using A2A.AspNetCore;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI.Hosting;
using Microsoft.Extensions.AI;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddOpenApi();
builder.Services.AddSwaggerGen();

string endpoint = builder.Configuration["AZURE_OPENAI_ENDPOINT"]
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = builder.Configuration["AZURE_OPENAI_DEPLOYMENT_NAME"]
    ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT_NAME is not set.");

// Register the chat client
IChatClient chatClient = new AzureOpenAIClient(
        new Uri(endpoint),
        new DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();
builder.Services.AddSingleton(chatClient);

// Register an agent
var pirateAgent = builder.AddAIAgent("pirate", instructions: "You are a pirate. Speak like a pirate.");

var app = builder.Build();

app.MapOpenApi();
app.UseSwagger();
app.UseSwaggerUI();

// Expose the agent via A2A protocol. You can also customize the agentCard
app.MapA2A(pirateAgent, path: "/a2a/pirate", agentCard: new()
{
    Name = "Pirate Agent",
    Description = "An agent that speaks like a pirate.",
    Version = "1.0"
});

app.Run();

Testowanie agenta

Po uruchomieniu aplikacji możesz przetestować agenta A2A przy użyciu następującego .http pliku lub za pośrednictwem interfejsu użytkownika programu Swagger.

Format danych wejściowych jest zgodny ze specyfikacją A2A. Możesz podać wartości dla:

  • messageId - Unikatowy identyfikator dla tej konkretnej wiadomości. Możesz utworzyć własny identyfikator (np. identyfikator GUID) lub ustawić go tak, aby null umożliwić agentowi automatyczne wygenerowanie identyfikatora.
  • contextId - Identyfikator konwersacji. Podaj własny identyfikator, aby rozpocząć nową konwersację lub kontynuować istniejącą, ponownie używając poprzedniego contextIdelementu . Agent zachowa historię konwersacji dla tego samego contextIdelementu . Agent wygeneruje jeden dla ciebie również, jeśli żaden nie zostanie podany.
# Send A2A request to the pirate agent
POST {{baseAddress}}/a2a/pirate/v1/message:stream
Content-Type: application/json
{
  "message": {
    "kind": "message",
    "role": "user",
    "parts": [
      {
        "kind": "text",
        "text": "Hey pirate! Tell me where have you been",
        "metadata": {}
      }
    ],
	"messageId": null,
    "contextId": "foo"
  }
}

Uwaga: zastąp {{baseAddress}} punktem końcowym serwera.

To żądanie zwraca następującą odpowiedź JSON:

{
	"kind": "message",
	"role": "agent",
	"parts": [
		{
			"kind": "text",
			"text": "Arrr, ye scallywag! Ye’ll have to tell me what yer after, or be I walkin’ the plank? 🏴‍☠️"
		}
	],
	"messageId": "chatcmpl-CXtJbisgIJCg36Z44U16etngjAKRk",
	"contextId": "foo"
}

Odpowiedź zawiera contextId (identyfikator konwersacji), messageId (identyfikator wiadomości) i faktyczną zawartość od agenta pirackiego.

Konfiguracja AgentCard

Zawiera AgentCard metadane dotyczące agenta na potrzeby odnajdywania i integracji:

app.MapA2A(agent, "/a2a/my-agent", agentCard: new()
{
    Name = "My Agent",
    Description = "A helpful agent that assists with tasks.",
    Version = "1.0",
});

Dostęp do karty agenta można uzyskać, wysyłając następujące żądanie:

# Send A2A request to the pirate agent
GET {{baseAddress}}/a2a/pirate/v1/card

Uwaga: zastąp {{baseAddress}} punktem końcowym serwera.

Właściwości karty agenta

  • Nazwa: Nazwa wyświetlana agenta
  • Opis: krótki opis agenta
  • Wersja: ciąg wersji agenta
  • Adres URL: adres URL punktu końcowego (automatycznie przypisany, jeśli nie zostanie określony)
  • Możliwości: metadane opcjonalne dotyczące przesyłania strumieniowego, powiadomień typu push i innych funkcji

Uwidacznianie wielu agentów

Możesz uwidocznić wielu agentów w jednej aplikacji, o ile ich punkty końcowe nie zderzają się. Oto przykład:

var mathAgent = builder.AddAIAgent("math", instructions: "You are a math expert.");
var scienceAgent = builder.AddAIAgent("science", instructions: "You are a science expert.");

app.MapA2A(mathAgent, "/a2a/math");
app.MapA2A(scienceAgent, "/a2a/science");

Pakiet agent-framework-a2a umożliwia nawiązywanie połączenia z zewnętrznymi agentami zgodnymi z usługą A2A i komunikowanie się z nimi.

pip install agent-framework-a2a --pre

Nawiązywanie połączenia z agentem A2A

Użyj A2AAgent, aby opakować dowolny zdalny punkt końcowy A2A. Agent ustala zdolności agenta zdalnego za pomocą AgentCard i zarządza wszystkimi szczegółami protokołu.

import asyncio
import httpx
from a2a.client import A2ACardResolver
from agent_framework.a2a import A2AAgent

async def main():
    a2a_host = "https://your-a2a-agent.example.com"

    # 1. Discover the remote agent's capabilities
    async with httpx.AsyncClient(timeout=60.0) as http_client:
        resolver = A2ACardResolver(httpx_client=http_client, base_url=a2a_host)
        agent_card = await resolver.get_agent_card()
        print(f"Found agent: {agent_card.name}")

    # 2. Create an A2AAgent and send a message
    async with A2AAgent(
        name=agent_card.name,
        agent_card=agent_card,
        url=a2a_host,
    ) as agent:
        response = await agent.run("What are your capabilities?")
        for message in response.messages:
            print(message.text)

asyncio.run(main())

Odpowiedzi w strumieniowaniu

A2A w sposób naturalny obsługuje przesyłanie strumieniowe za pośrednictwem zdarzeń wysyłanych przez serwer — aktualizacje są dostarczane w czasie rzeczywistym, w miarę jak działa agent zdalny:

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    async with agent.run("Tell me about yourself", 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))")

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 opcję uzyskania tokenu kontynuacji, którego można użyć do sondowania lub ponownego przypisania później:

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)

Authentication

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

from a2a.client.auth.interceptor import AuthInterceptor

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

Zobacz też

Dalsze kroki

protokółAG-UI

  • Last updated on