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 Microsoft Foundry
dotnet add package Azure.AI.Projects --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Agents.AI.Foundry --prerelease

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

3. 設定 Microsoft Foundry 連線

此應用程式需要 Microsoft Foundry 專案連線。 用 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.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
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 AIProjectClient(
        new Uri(endpoint),
        new DefaultAzureCredential())
        .GetProjectOpenAIClient()
        .GetProjectResponsesClient()
        .AsIChatClient(deploymentName);

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

警告

DefaultAzureCredential 開發方便，但在生產過程中需謹慎考量。 在生產環境中，建議使用特定的憑證（例如 ManagedIdentityCredential），以避免延遲問題、意外的憑證探測，以及備援機制帶來的安全風險。

測試代理程式

應用程式執行後，你可以使用以下 .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 規範的代理程式，並透過 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:
    stream = agent.run("Tell me about yourself", 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 取得一個延續代幣，方便日後投票或訂閱：

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)

對話身份（context_id）

當你用 呼叫 A2AAgent.run() 時，如果外發訊息尚未攜帶 A2A，代理會自動從中推導出 A2A AgentSessioncontext_id。session.service_session_id 這讓你能在多個 A2A 通話中維持對話連續性，而不必手動設定 context_id 每則訊息：

from agent_framework import AgentSession
from agent_framework.a2a import A2AAgent

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    session = AgentSession(service_session_id="my-conversation-1")

    # context_id is automatically set to "my-conversation-1"
    response = await agent.run("Hello!", session=session)

    # Subsequent calls with the same session continue the conversation
    response = await agent.run("Follow-up question", session=session)

如果訊息中context_id明確包含 additional_properties ，該值優先於會話衍生的備援。

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

透過 A2A 暴露代理框架中的代理程式

它 A2AExecutor 會將任何代理框架 Agent 調整為 A2A 伺服器端協定。 你可以用官方 a2a-sdk Starlette/ASGI 伺服器來架設，讓其他 A2A 客戶能發現並聯絡你的經紀人。

import uvicorn
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
from agent_framework import Agent
from agent_framework.a2a import A2AExecutor
from agent_framework.openai import OpenAIChatClient
from starlette.applications import Starlette

flight_skill = AgentSkill(
    id="Flight_Booking",
    name="Flight Booking",
    description="Search and book flights across Europe.",
    tags=["flights", "travel", "europe"],
    examples=[],
)

public_agent_card = AgentCard(
    name="Europe Travel Agent",
    description="Helps users search and book flights and hotels across Europe.",
    version="1.0.0",
    default_input_modes=["text"],
    default_output_modes=["text"],
    capabilities=AgentCapabilities(streaming=True),
    supported_interfaces=[
        AgentInterface(url="http://localhost:9999/", protocol_binding="JSONRPC"),
    ],
    skills=[flight_skill],
)

agent = Agent(
    client=OpenAIChatClient(),
    name="Europe Travel Agent",
    instructions="You are a helpful Europe Travel Agent.",
)

request_handler = DefaultRequestHandler(
    agent_executor=A2AExecutor(agent),
    task_store=InMemoryTaskStore(),
    agent_card=public_agent_card,
)

server = Starlette(
    routes=[
        *create_agent_card_routes(public_agent_card),
        *create_jsonrpc_routes(request_handler, "/"),
    ]
)

uvicorn.run(server, host="0.0.0.0", port=9999)

A2AExecutor 會在底層代理支援串流時，將代理程式更新作為 A2A 成品串流傳送，並將 A2A context_id 傳遞為代理程式工作階段的 session_id。 你可以子類別 A2AExecutor 並覆寫該 handle_events 方法，將代理的輸出格式自訂轉換成 A2A 協定事件。

小提示

請參閱 agent_framework_to_a2a.py 範例 以獲得完整的可執行範例。

A2A 協議

Go Agent Framework 支援代理對代理（Agent-to-Agent，A2A）協定，可用於託管 Agent Framework 代理以及使用遠端 A2A 代理。 Go 整合使用 provider/a2aprovider 套件來支援伺服器端託管配接器和用戶端存取。

在你的 Go 模組中安裝 Agent Framework 和 A2A 套件：

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

透過 A2A 託管代理程式

建立或重複使用 Agent Framework 代理程式，以 A2A 代理卡加以描述，並透過其中一種 A2A 傳輸繫結對外公開。 在此範例中， hostAgent 是任意代理框架 *agent.Agent;伺服器在 JSON-RPC / 端點作為主機，並在眾所周知的 A2A 路徑上服務代理卡。

import (
    "fmt"
    "net/http"

    "github.com/a2aproject/a2a-go/v2/a2a"
    "github.com/a2aproject/a2a-go/v2/a2asrv"
    "github.com/microsoft/agent-framework-go/provider/a2aprovider"
)

url := "http://localhost:5000"

card := &a2a.AgentCard{
    Name:               "InvoiceAgent",
    Description:        "Handles requests relating to invoices.",
    Version:            "1.0.0",
    DefaultInputModes:  []string{"text"},
    DefaultOutputModes: []string{"text"},
    Capabilities: a2a.AgentCapabilities{
        Streaming: false,
    },
    SupportedInterfaces: []*a2a.AgentInterface{
        a2a.NewAgentInterface(url, a2a.TransportProtocolJSONRPC),
    },
}

mux := http.NewServeMux()
requestHandler := a2asrv.NewHandler(
    a2aprovider.NewExecutor(hostAgent, a2aprovider.ExecutorConfig{}),
    a2asrv.WithExtendedAgentCard(card),
)
mux.Handle("/", a2asrv.NewJSONRPCHandler(requestHandler))
mux.Handle(a2asrv.WellKnownAgentCardPath, a2asrv.NewStaticAgentCardHandler(card))

if err := http.ListenAndServe(":5000", mux); err != nil {
    panic(fmt.Errorf("A2A server failed: %w", err))
}

當您想要公開 HTTP+JSON 傳輸繫結時，請使用 a2asrv.NewRESTHandler 包裝同一個請求處理常式。 若應允許託管代理回傳用於長時間執行工作的 A2A 任務，請將 ExecutorConfig.AllowBackgroundResponses 設為 true。

使用 A2A 代理程式

解析遠端代理卡，從它建立一個 A2A 用戶端，並將用戶端包裝成標準的代理框架代理：

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

a2aprovider提供者將 A2A context_id 與任務 ID 儲存在代理框架會話中，以便後續訊息能維持對話的連續性。

協定選擇

如果遠端代理宣告多個傳輸綁定，建立 A2A 用戶端時請設定優先傳輸方式：

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

當你想偏好 JSON-RPC 時再用 a2a.TransportProtocolJSONRPC 。

長時間執行的工作

A2A 任務透過代理框架的延續標記浮現。 以明確指定的工作階段和 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 代理作為工具

你也可以將遠端 A2A 代理暴露給主機代理，作為功能工具。 解析每個遠端代理，使用 a2aprovider.NewAgent 將其包裝，然後使用 agenttool.New 將其轉換為工具。 在此範例中，主機代理使用Microsoft Foundry專案支援的模型部署。

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

host := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Instructions: "Use your tools to delegate requests to specialized remote agents.",
    Config: agent.Config{
        Tools: tools,
    },
})

小提示

完整可執行範例請參閱 A2A 客戶端-伺服器範例、 A2A 提供者範例及 A2A 代理工具範例 。

另請參閱

後續步驟

AG-UI 協議

  • Last updated on