A2A エージェントサービス

A2AAgentを使用すると、アプリケーションはエージェント間 (A2A) プロトコルを介して公開されているリモートエージェントに接続できます。 A2A 準拠エンドポイントが標準の AIAgentとしてラップされるため、 RunAsync や RunStreamingAsync などの使い慣れた方法を使用して、構築されたフレームワークやテクノロジに関係なくリモートエージェントと対話できます。

エージェントフレームワークエージェントを A2A サーバーとして公開するには、 A2A を使用したホストエージェントに関するページを参照してください。

はじめに

必要な NuGet パッケージをプロジェクトに追加します。

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

エージェントの検出

リモート A2A エージェントと通信する前に、それを検出し、 AIAgent インスタンスを作成する必要があります。 A2A プロトコルは、エージェントフレームワークでサポートされる 3 つの検出戦略を定義します。

Well-Known URI

A2A エージェントは、標準化されたパス () でhttps://{domain}/.well-known/agent-card.jsonを検出できるようにします。 A2ACardResolverを使用してカードをフェッチし、1 回の呼び出しでエージェントを作成します。

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

ストリーミング

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 エージェントが即時メッセージではなくタスクを返すと、Agent Framework は、結果のポーリングや中断されたストリームへの再接続に使用できる継続トークンを提供します。

タスク完了のポーリング

ストリーミング以外のシナリオでは、 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 エージェントは、ストリーム内の特定のポイントからのストリーム再開ではなく、ストリーム再接続 (最初から同じ応答ストリームを取得) をサポートします。

Tools

A2AAgent は、リモート A2A エージェントを囲むトランスポートレベルのラッパーです。リモートエージェントがリモート側でライブで使用し、コードからは見えないツール。エージェントフレームワークツールの種類 (関数ツール、コードインタープリター、ファイル検索、ホスト/ローカル MCP など) は、リモートエージェントの機能を拡張し、リモートエージェントの構成を変更するために、 A2AAgent 自体では構成されません。

はじめに

A2A パッケージをインストールします。

pip install agent-framework-a2a --pre

初期化

A2AAgent は、リモートエージェントについて事前に知っている程度に応じて、3 つの方法で初期化できます。

直接 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を使用してトランスポートをネゴシエートします。

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)

ストリーミング

リモートエージェントが要求を処理する際に、 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 コンテキスト、タスク ID、タスクの状態は一緒に保持されます。

TASK_STATE_INPUT_REQUIREDのタスクの場合、次のメッセージは、同じタスクを続行task_id設定します。その他のタスク状態の場合、リモートエージェントが前の結果を絞り込んだり続行したりできるように、前のタスク ID が reference_task_ids 経由で送信されます。

Authentication

セキュリティで保護された A2A エンドポイントに AuthInterceptor を使用します。

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

Foundry エージェントで A2A エージェントをツールとして呼び出す場合は、get_a2a_toolのFoundryChatClient ファクトリを参照してください。

Go は、 provider/a2aprovider パッケージを介してリモート A2A エージェントをサポートします。

エージェントフレームワークと 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 とタスク ID をエージェントフレームワークセッションに格納するため、フォローアップメッセージは会話の継続性を維持できます。

プロトコルの選択

リモートエージェントが複数のトランスポートバインディングをアドバタイズする場合は、A2A クライアントの作成時に優先トランスポートを構成します。

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

JSON-RPC を使用する場合は、 a2a.TransportProtocolJSONRPC を使用します。

実行時間の長いタスク

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 を使用するホストエージェント

より深く進む:

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-13

A2A エージェント サービス

はじめに

エージェントの検出

Well-Known URI

Catalog-Based 検出

ダイレクト構成

プロトコルの選択

ストリーミング

バックグラウンド応答

タスク完了のポーリング

ストリームの再接続

Tools

はじめに

初期化

直接 URL

エージェント カード

Well-Known URI (A2ACardResolver)

ストリーミング

長時間実行タスク

会話の識別 (context_id)

Authentication

タイムアウト構成

Tools

リモート A2A エージェントに接続する

プロトコルの選択

実行時間の長いタスク

リモート A2A エージェントをツールとして使用する

次のステップ

フィードバック

その他のリソース

A2A エージェントサービス

エージェントカード