A2A 統合

エージェント間 (A2A) プロトコルを使用すると、エージェント間の標準化された通信が可能になり、さまざまなフレームワークとテクノロジで構築されたエージェントがシームレスに通信できるようになります。

A2A とは

A2A は、次をサポートする標準化されたプロトコルです。

エージェントカードを使用したエージェントの検出
エージェント間のメッセージベースの通信
タスクを介した実行時間の長いエージェントプロセス
異なるエージェントフレームワーク間のクロスプラットフォーム相互運用性

詳細については、 A2A プロトコルの仕様を参照してください。

Microsoft.Agents.AI.Hosting.A2A.AspNetCore ライブラリは、A2A プロトコルを介してエージェントを公開するための ASP.NET Core 統合を提供します。

NuGet パッケージ:

例

この最小限の例は、A2A を介してエージェントを公開する方法を示しています。このサンプルには、テストを簡略化するための OpenAPI と Swagger の依存関係が含まれています。

1. ASP.NET Core Web API プロジェクトを作成する

新しい ASP.NET Core Web API プロジェクトを作成するか、既存のプロジェクトを使用します。

2. 必要な依存関係をインストールする

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

.NET CLI

プロジェクトディレクトリで次のコマンドを実行して、必要な 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"

$env:AZURE_OPENAI_ENDPOINT = "https://<your-openai-resource>.openai.azure.com/"
$env:AZURE_OPENAI_DEPLOYMENT_NAME = "gpt-4o-mini"

export AZURE_OPENAI_ENDPOINT="https://<your-openai-resource>.openai.azure.com/"
export AZURE_OPENAI_DEPLOYMENT_NAME="gpt-4o-mini"

  "AZURE_OPENAI_ENDPOINT": "https://<your-openai-resource>.openai.azure.com/",
  "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();

Warnung

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 の構成

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 のプロパティ

名前: エージェントの表示名
説明: エージェントの簡単な説明
バージョン: エージェントのバージョン文字列
URL: エンドポイント URL (指定されていない場合は自動的に割り当てられます)
機能: ストリーミング、プッシュ通知、およびその他の機能に関する省略可能なメタデータ

複数のエージェントを公開する

エンドポイントが競合しない限り、1 つのアプリケーションで複数のエージェントを公開できます。次に例を示します。

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 プロトコル経由で Agent Framework エージェントを公開できます。

pip install agent-framework-a2a --pre

A2A エージェントへの接続

A2AAgentを使用して、リモート A2A エンドポイントをラップします。エージェントは、AgentCard を使用してリモートエージェントの機能を解決し、すべてのプロトコルの詳細を処理します。

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()をAgentSessionと共に呼び出すと、送信メッセージにA2A context_idがまだ含まれていない場合、エージェントは自動的にsession.service_session_idからA2A context_idを導き出します。これにより、すべてのメッセージに対して context_id を手動で設定することなく、複数の A2A 呼び出し間で会話の継続性を維持できます。

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がある場合、その値はセッションから派生したフォールバックよりも優先されます。

認証

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

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 クライアントがエージェントを検出して呼び出すことができるように、公式の a2a-sdk Starlette/ASGI サーバーでホストできます。

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 context_id をエージェントセッションの session_idとして伝達する場合に、エージェントの更新を A2A 成果物としてストリーミングします。 A2AExecutorをサブクラス化し、handle_events メソッドをオーバーライドして、エージェントの出力形式から A2A プロトコルイベントへのカスタム変換を実装できます。

Tip

実行可能な完全な例については、 agent_framework_to_a2a.py サンプルを参照してください。

A2A プロトコル

Go Agent Framework では、Agent Framework エージェントのホストとリモート A2A エージェントの使用の両方に対して、エージェント間 (A2A) プロトコルがサポートされています。 Go 統合では、サーバー側ホスティングアダプターとクライアント側アクセスの両方に provider/a2aprovider パッケージが使用されます。

Go モジュールにエージェントフレームワークと A2A パッケージをインストールします。

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

A2A 経由でエージェントをホストする

エージェントフレームワークエージェントを作成または再利用し、A2A エージェントカードで説明し、A2A トランスポートバインディングのいずれかを介して公開します。この例では、 hostAgent は任意の Agent Framework *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 でラップします。 ExecutorConfig.AllowBackgroundResponsesをtrueに設定します。長時間の作業のために、ホストされるエージェントが A2A タスクを返すように許可する必要がある場合は、この設定を行います。

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

a2aprovider プロバイダーは、A2A context_idとタスク ID を Agent Framework セッションに格納するため、フォローアップメッセージは会話の継続性を維持できます。

プロトコルの選択

リモートエージェントが複数のトランスポートバインディングをアドバタイズする場合は、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 エージェントをツールとして使用する

また、リモート A2A エージェントを関数ツールとしてホストエージェントに公開することもできます。各リモートエージェントを解決し、 a2aprovider.NewAgentでラップしてから、 agenttool.Newを使用してツールに変換します。この例では、ホストエージェントは、Foundry プロジェクトに基づくMicrosoftモデルデプロイを使用します。

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

Tip

完全な実行可能な例については、 A2A クライアント/サーバーのサンプル、 A2A プロバイダーのサンプル、およびツールとしての A2A エージェントのサンプルを参照してください。

こちらもご覧ください

次のステップ

AG-UI プロトコル

フィードバック

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

Last updated on 2026-07-10