共用方式為


A2A 整合

代理對代理(Agent-to-Agent,A2A)協定允許代理間標準化通訊,使使用不同框架與技術打造的代理能無縫溝通。

什麼是A2A?

A2A 是一個標準化協定,支援:

  • 透過代理卡發現代理
  • 代理間的訊息通訊
  • 透過任務進行的長期代理程序
  • 不同代理框架間的跨平台互通性

欲了解更多資訊,請參閱 A2A 協定規範

函式庫 Microsoft.Agents.AI.Hosting.A2A.AspNetCore 提供 ASP.NET Core 整合,以 A2A 協定公開你的代理。

NuGet 套件:

Example

這個最小的例子展示了如何透過 A2A 暴露代理人。 範例包含與 OpenAPI 和 Swagger 相關的相依性,以簡化測試。

1. 建立 ASP.NET 核心網頁API專案

建立一個新的 ASP.NET 核心網頁 API 專案,或使用現有的專案。

2. 安裝所需的相依性套件

安裝下列套件:

在專案目錄中執行以下指令來安裝所需的 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. 配置Azure OpenAI connection

該應用程式需要 Azure OpenAI 連線。 用 dotnet user-secrets 或 環境變數設定端點和部署名稱。 你也可以直接編輯 appsettings.json,但對於部署在生產環境的應用程式來說,這不建議,因為有些資料可能被視為機密。

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. 將程式碼加入Program.cs

將 的內容 Program.cs 替換為以下程式碼並執行應用程式:

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

測試代理程式

應用程式執行後,你可以使用以下 .http 檔案或 Swagger UI 測試 A2A 代理。

輸入格式符合 A2A 規範。 你可以提供以下數值:

  • messageId - 此訊息的唯一識別碼。 你可以建立自己的 ID(例如 GUID),或設定為 讓 null 代理自動產生。
  • contextId - 對話識別碼。 提供您自己的 ID 來開啟新對話,或透過重複使用先前的 contextId 來繼續現有的對話。 客服人員會為此 contextId維持對話紀錄。 如果沒有提供,代理程式也會為您生成一份。
# 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"
  }
}

注意:請用你的伺服器端點替換 {{baseAddress}}

此請求回傳以下 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"
}

回應內容包括 contextId (對話識別碼)、 messageId (訊息識別碼)以及盜版代理的實際內容。

代理卡配置

提供 AgentCard 關於代理程式的元資料,供發現與整合:

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

您可以透過發送以下請求取得代理人卡:

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

注意:請用你的伺服器端點替換 {{baseAddress}}

AgentCard 屬性

  • 名稱:代理人的顯示名稱
  • 描述:代理人簡要描述
  • 版本:代理的版本字串
  • 網址:端點網址(未指定時自動指派)
  • 功能:可選的串流、推播通知及其他功能的元資料

揭露多位特工

你可以在單一應用程式中暴露多個代理,只要它們的端點不碰撞。 以下為範例:

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

agent-framework-a2a 套件讓您能連接並與符合 A2A 標準的外部客服人員溝通。

pip install agent-framework-a2a --pre

連接 A2A 代理

使用A2AAgent 來包裝任何遠端的 A2A 端點。 代理透過其代理卡解析遠端代理的能力,並處理所有協定細節。

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

串流回應

A2A 自然支援透過 Server-Sent 事件進行串流——隨著遠端代理工作,更新會即時到來:

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

長時間執行的任務

預設情況下, A2AAgent 會等待遠端代理完成後才返回。 對於長時間執行的任務,設定 background=True 以取得延續標記,便於日後輪詢或重新訂閱:

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

使用 AuthInterceptor 以確保 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!")

另請參閱

後續步驟