Agent2Agent 入門 (10) - クイックスタート

npaka

2025年7月6日 08:19

「Agent2Agent」のクイックスタートをまとめました。

・Python Quickstart Tutorial: Building an A2A Agent

前回

1. Agent2Agent のクイックスタート

「A2A Python SDK」で「Hello World」を返すだけの、シンプルな A2Aエージェントを構築します。

2. セットアップ

セットアップ手順は、次のとおりです。

(1) プロジェクトフォルダの作成

mkdir a2a-samples
cd a2a-samples

(2) Pythonの仮想環境の準備。
Python 3.10以上を使用します。

・Mac

python -m venv .venv
source .venv/bin/activate

・Windows

python -m venv .venv
.venv\Scripts\activate

(2) パッケージのインストール。

pip install a2a-sdk uvicorn

3. A2Aエージェントの実装

3-1. コアエージェントの定義

「Hello World」を返すだけの、シンプルなコアエージェントを構築します。コアエージェントは通常、LLM 呼び出しやツール実行などを行います。

・agents_executor.py

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message

# コアエージェントの作成
class HelloWorldAgent:
    # 呼び出し
    async def invoke(self) -> str:
        return 'Hello World'

3-2. AgentExecutor でコアエージェントをラップ

「AgentExecutor」は、「A2Aサーバ」(DefaultRequestHandler) からのリクエストを受けてエージェント固有のロジックを実行するインターフェースです。

主な役割は、次のとおりです。

・プロトコルとロジックの橋渡し
A2Aプロトコル経由で「RequestContext」を受け取り、コアエージェントの機能を呼び出し、それに応じた「進捗」や「結果」を「EventQueue 」経由で返却します。

・タスク管理 (1回の呼び出しで処理が終わらない場合)
タスクの開始・更新・完了通知を行うために、内部で「TaskUpdater」(または同等の機能) を使い、サブミット→実行中→完了などの状態遷移を扱います。

実装すべきメソッドは、次の2つです。

async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
コアエージェントの機能を呼び出し、Message・Task・TaskStatusUpdateEvent・TaskArtifactUpdateEvent のいずれかを返す。

async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
タスクのキャンセルを実行。

・agents_executor.py (続き)

# AgentExecutorの作成
class HelloWorldAgentExecutor(AgentExecutor):

    # 初期化
    def __init__(self):
        self.agent = HelloWorldAgent()

    # 実行
    async def execute(
        self,
        context: RequestContext,  # リクエストコンテキスト
        event_queue: EventQueue,  # イベントキュー
    ) -> None:
        result = await self.agent.invoke()  # コアエージェントの呼び出し
        await event_queue.enqueue_event(
            new_agent_text_message(result)  # 結果を返す
        )

    # キャンセル
    async def cancel(
        self, 
        context: RequestContext,   # リクエストコンテキスト
        event_queue: EventQueue  # イベントキュー
    ) -> None:
        raise Exception('cancel not supported')

3-3. AgentSkill と AgentCard の準備

「AgentSkill」と「AgentCard」で他エージェントからの発見機構を提供します。「A2AStarletteApplication」を使ってHTTPサーバを立ち上げ、A2Aエンドポイントを公開します。

・__main__.py

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import (
    HelloWorldAgentExecutor,  # type: ignore[import-untyped]
)

# メイン
if __name__ == '__main__':
    # Agent Skill の準備
    skill = AgentSkill(
        id='hello_world',                         # スキルの一意な識別子
        name='Returns hello world',               # スキル名
        description='just returns hello world',   # スキルの説明
        tags=['hello world'],                     # タグ
        examples=['hi', 'hello world'],           # サンプル入力例
    )

    # Agent Card の準備
    public_agent_card = AgentCard(
        name='Hello World Agent',                 # エージェントの表示名
        description='Just a hello world agent',   # 説明文
        url='http://localhost:9999/',             # サービスURL
        version='1.0.0',                          # バージョン
        defaultInputModes=['text'],               # デフォルト入力モード
        defaultOutputModes=['text'],              # デフォルト出力モード
        capabilities=AgentCapabilities(streaming=True),  # ストリーミング対応
        skills=[skill],                           # 公開カードに含めるスキル（基本スキルのみ）
        supportsAuthenticatedExtendedCard=False,  # 認証後に拡張カードをサポート
    )

    # リクエストハンドラの作成
    request_handler = DefaultRequestHandler(
        agent_executor=HelloWorldAgentExecutor(), # エージェント実行ロジック
        task_store=InMemoryTaskStore(), # メモリ内タスクストア
    )

    # A2Aサーバの作成
    server = A2AStarletteApplication(
        agent_card=public_agent_card, # 公開カード
        http_handler=request_handler, # リクエストハンドラ
    )

    # サーバの実行
    uvicorn.run(server.build(), host='0.0.0.0', port=9999)

3-4. A2Aエージェントの実行

(1) A2Aエージェントの実行。
A2Aエンドポイントが公開されます。

python __main__.py

INFO:     Started server process [74797]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:9999 (Press CTRL+C to quit)

4. A2Aエージェントの呼び出し

(1) A2Aエージェントを呼び出すクライアントの作成。

・test_client.py

from typing import Any
from uuid import uuid4
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    MessageSendParams,
    SendMessageRequest,
)

# メイン
async def main() -> None:
    # 定数
    BASE_URL = 'http://localhost:9999'
    PUBLIC_AGENT_CARD_PATH = '/.well-known/agent.json'

    async with httpx.AsyncClient() as httpx_client:
        # A2ACardResolverの準備
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=BASE_URL,
            # デフォルトパスのためpath指定は省略
        )

        # Agent Card の取得
        try:
            print(f'Agent Card の取得中: {BASE_URL}{PUBLIC_AGENT_CARD_PATH}')
            _public_card = (
                await resolver.get_agent_card()
            )
            print('Agent Card の取得完了:')
            print(_public_card.model_dump_json(indent=2, exclude_none=True))
        except Exception as e:
            print(f'公開エージェントカードの取得に失敗: {e}', exc_info=True)
            raise RuntimeError('公開エージェントカードの取得に失敗') from e

        # A2AClientの準備
        client = A2AClient(httpx_client=httpx_client, agent_card=_public_card)

        # メッセージ送信
        send_message_payload: dict[str, Any] = {
            'message': {
                'role': 'user',
                'parts': [
                    {'kind': 'text', 'text': 'how much is 10 USD in INR?'}
                ],
                'messageId': uuid4().hex,
            },
        }
        request = SendMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )
        response = await client.send_message(request)
        print(response.model_dump(mode='json', exclude_none=True))

# メイン
if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

(2) A2Aエージェントを呼び出すクライアントの実行。

python test_client.py

Agent Card の取得中: http://localhost:9999/.well-known/agent.json
Agent Card の取得完了:
{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text"
  ],
  "defaultOutputModes": [
    "text"
  ],
  "description": "Just a hello world agent",
  "name": "Hello World Agent",
  "protocolVersion": "0.2.5",
  "skills": [
    {
      "description": "just returns hello world",
      "examples": [
        "hi",
        "hello world"
      ],
      "id": "hello_world",
      "name": "Returns hello world",
      "tags": [
        "hello world"
      ]
    }
  ],
  "supportsAuthenticatedExtendedCard": false,
  "url": "http://localhost:9999/",
  "version": "1.0.0"
}
{'id': '5c81efb8-f7b6-4aac-a470-b1bcdc0e3ea6', 'jsonrpc': '2.0', 'result': {'kind': 'message', 'messageId': '553cfc71-4ee9-4848-9965-1fe878107055', 'parts': [{'kind': 'text', 'text': 'Hello World'}], 'role': 'agent'}}

5. ストリーミング

同期メッセージでなく、ストリーミングメッセージを送信する場合は、「SendStreamingMessageRequest」を使います。

(1) SendStreamingMessageRequestのテストコードの追加。

・test_client.py に追加

from a2a.types import (
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
)

        # ストリーミングメッセージ送信
        streaming_request = SendStreamingMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )
        stream_response = client.send_message_streaming(streaming_request)
        async for chunk in stream_response:
            print(chunk.model_dump(mode='json', exclude_none=True))

(2) 実行。

python test_client.py

{'id': '9f3f1e81-44ed-429f-8cf7-4c9991431195', 'jsonrpc': '2.0', 'result': {'kind': 'message', 'messageId': 'd7fb6f27-e305-4361-9dc9-fc66d0409632', 'parts': [{'kind': 'text', 'text': 'Hello World'}], 'role': 'agent'}}

6. サンプルエージェント

公式リポジトリでサンプルエージェントが提供されています。異なるフレームワーク上に構築され、それぞれ異なる機能を備えたサンプルです。各エージェントはスタンドアロンのA2Aサーバとして動作します。

サーバとやり取りするには、ホストアプリ (CLIなど) でA2AClientを使用します。詳しくはホストアプリを参照してください。

・Google ADK Facts
Grounding と Google 検索、ADK を使用して楽しい事実を伝えるサンプルエージェント。

・Google ADK Expense Reimbursement
経費報告書の記入を模擬的に行うサンプルエージェント。複数ターンのやり取りや、A2Aを介したWebフォームへの返信/返信の仕組みを紹介。

・AG2 MCP Agent with A2A Protocol
A2A プロトコルを通じて公開される AG2 で構築された MCP 対応エージェント。

・Azure AI Foundry Agent Service
Azure AI Foundry Agent Serviceを使用したサンプルエージェント。

・LangGraph
ツールを使用して通貨を変換できるサンプルエージェント。複数ターンのインタラクション、ツールの使用、ストリーミング更新を紹介。

・CrewAI
画像を生成できるサンプルエージェント。CrewAIの使用とA2A経由の画像送信を紹介。

・LlamaIndex
ファイルを解析し、解析されたコンテンツをコンテキストとしてユーザーとチャットできるサンプルエージェントです。マルチターンインタラクション、ファイルのアップロードと解析、ストリーミング更新などの機能を紹介。

・Marvin Contact Extractor Agent
Agent2Agent (A2A) プロトコルと統合され、Marvinフレームワークを使用してテキストから構造化された連絡先情報を抽出するエージェント。

・Enterprise Data Agent
あらゆるデータベース、データウェアハウス、アプリからの質問に答えることができるサンプルエージェント。 Gemini 2.5 flash + MindsDB を搭載。

・Semantic Kernel Agent
Semantic Kernel上に構築され、A2A プロトコルを通じて公開される旅行代理店を実装する方法を示す。

・travel planner Agent
Google の公式 a2a-python SDK に基づいて実装され、A2A プロトコルを通じて実装された旅行アシスタントデモ。