A2A 代理服務

這 A2AAgent 讓你的應用程式能夠連接到透過代理 對代理（Agent-to-Agent，A2A）協定暴露的遠端代理。 它將任何符合 A2A 的端點包裝成標準AIAgent，因此你可以使用熟悉的方法，RunAsync例如RunStreamingAsync與遠端代理互動，無論他們是用什麼框架或技術建置的。

若要將代理框架代理作為 A2A 伺服器暴露，請參見 帶有 A2A 的主機代理。

使用者入門

為您的專案新增所需的 NuGet 套件：

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

探員發現號

在與遠端 A2A 代理溝通前，你需要先發現它並建立一個 AIAgent 實例。 A2A 協議定義了三種 發現策略，每種策略皆由代理框架支援。

Well-Known URI

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。 預設情況下，HTTP+JSON 會優先於 JSON-RPC。 用 A2AClientOptions.PreferredBindings 來明確控制使用哪種協定綁定：

Note

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

Streaming

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

Note

A2A 代理支援串流重連（從一開始就取得相同的回應串流），而非從串流中特定點恢復串流。

工具

A2AAgent 是包裹遠端 A2A 代理的傳輸層封裝器。 遠端代理使用的任何工具都存在於遠端端，對你的程式碼是隱形的。 代理框架的工具類型（功能工具、程式碼直譯器、檔案搜尋、託管/本地 MCP 等）本身並未設定 A2AAgent ——以擴展遠端代理的功能、更改遠端代理的設定。

使用者入門

安裝 A2A 套件：

pip install agent-framework-a2a --pre

初始化

A2AAgent 根據你事先對遠端代理的了解程度，初始化方式有三種。

直接網址

對於開發或緊密耦合系統，且端點已知：

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。

Well-Known URI（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)

Streaming

在遠端代理處理請求時，使用 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以映射方式儲存持久協定狀態AgentSession.service_session_idA2AServiceSessionId：

Field	類型	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 上下文、任務 ID 與任務狀態保持一致。

對於 中的 TASK_STATE_INPUT_REQUIRED任務，下一則訊息會設定該 task_id 任務繼續執行相同任務。 對於其他任務狀態，會傳送 reference_task_ids 先前的任務 ID，讓遠端代理能從先前結果細化或繼續。

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秒寫入、5秒池。

工具

A2AAgent 是包裹遠端 A2A 代理的傳輸層封裝器。 遠端代理使用的任何工具都存在於遠端端，對你的程式碼是隱形的。 代理框架的工具類型（功能工具、程式碼直譯器、檔案搜尋、託管/本地 MCP 等）本身並未設定 A2AAgent ——以擴展遠端代理的功能、更改遠端代理的設定。

如果你想讓 Foundry 代理以工具方式呼叫 A2A 代理，請參考工廠資料get_a2a_toolFoundryChatClient。

Go 透過套件 provider/a2aprovider 支援遠端 A2A 代理。

安裝 Agent Framework 與 A2A 套件：

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

連接遠端 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()

提供者將 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 代理作為工具

解析每個遠端代理，用 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 協定規範