執筆日
2025/6/6
やること
GitHubに公開されているA2Aのサンプルコードを見ながらA2Aの理解を深めていこうかなと。
手順
git clone https://github.com/google/A2A
cd A2A
git checkout v0.1.0
A2Aの主要なコンポーネントをコードから読み解く
以前、A2Aの主要コンポーネントをまとめました。
これを元に各コンポーネントについて調べていこうかなと思います。
Agent Card（エージェントカード）
AgentCard は、エージェントの自己情報を JSON 形式で公開する構造体です。通常 .well-known/agent.json によって他エージェントへ公開されます。
主な属性
| 属性名 | 説明 |
|---|---|
name |
エージェントの名前 |
description |
エージェントの説明 |
version |
バージョン番号 |
url |
エージェント API のルート URL |
capabilities |
対応する機能群（例: streaming, pushNotifications） |
defaultInputModes |
デフォルトで受け取る MIME タイプのリスト |
defaultOutputModes |
デフォルトで返す MIME タイプのリスト |
skills |
提供するスキル（AgentSkill）のリスト |
コード
A2A/samples/python/agents/ag2/__main__.py
agent_card = AgentCard(
name='YouTube Captions Agent',
description='AI agent that can extract closed captions and transcripts from YouTube videos.',
url=f'http://{host}:{port}/',
version='1.0.0',
defaultInputModes=YoutubeMCPAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=YoutubeMCPAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
Agent Skill（エージェントスキル）
AgentSkill は、エージェントが提供する機能（スキル）を定義する構造体です。
主な属性
| 属性名 | 説明 |
|---|---|
id |
スキルを識別するユニークな ID |
name |
スキル名（ユーザー向け） |
description |
スキルの説明 |
tags |
検索用タグ（カテゴリ） |
examples |
スキル使用例やプロンプト例 |
inputModes |
対応する入力の MIME タイプ |
outputModes |
対応する出力の MIME タイプ |
コード
A2A/samples/python/agents/ag2/__main__.py
skills = [
AgentSkill(
id='download_closed_captions',
name='Download YouTube Closed Captions',
description='Retrieve closed captions/transcripts from YouTube videos',
tags=['youtube', 'captions', 'transcription', 'video'],
examples=[
'Extract the transcript from this YouTube video: https://...',
'Download the captions for this YouTube tutorial',
],
)
]
Agent Client（A2A クライアント）
Agent Client（A2A Client） は、他のエージェントと通信するためのクライアント側のインターフェースです。
具体的には、対象のエージェントが公開している .well-known/agent.json を取得し、エージェントの能力やスキルを把握した上で、タスク送信などの API を呼び出します。
A2A Client は、Agent 間通信における発信側の役割を果たします。
主な属性
| 属性名 | 説明 |
|---|---|
base_url |
通信先エージェントのベース URL（例：http://localhost:10003） |
agent_card_path |
AgentCard の取得パス（通常は /.well-known/agent.json） |
get_agent_card() |
対象エージェントの AgentCard を取得する関数 |
send_task() |
タスクを送信する関数（別クラスで呼び出される） |
コード
A2A/samples/python/agents/ag2/agent.py
import json
import httpx
from common.types import A2AClientJSONError, AgentCard
class A2ACardResolver:
def __init__(self, base_url, agent_card_path='/.well-known/agent.json'):
self.base_url = base_url.rstrip('/')
self.agent_card_path = agent_card_path.lstrip('/')
def get_agent_card(self) -> AgentCard:
with httpx.Client() as client:
response = client.get(f"{self.base_url}/{self.agent_card_path}")
response.raise_for_status()
try:
return AgentCard(**response.json())
except json.JSONDecodeError as e:
raise A2AClientJSONError(str(e)) from e
A2A Server（サーバー）
A2AServer は、エージェントの API を外部に公開し、リクエストを処理する HTTP サーバーです。
主な属性
| 属性名 | 説明 |
|---|---|
agent_card |
公開するエージェントの自己情報（AgentCard） |
task_manager |
タスク処理を担当するマネージャ（AgentTaskManager 等） |
host |
バインドするホスト名（例：localhost） |
port |
バインドするポート番号（例：10003） |
app |
ASGI ベースのアプリ（Starlette） |
コード
A2A/samples/python/common/client.py
server = A2AServer(
agent_card=agent_card,
task_manager=AgentTaskManager(agent=YoutubeMCPAgent()),
host=host,
port=port,
)
Task（タスク）
Task は、クライアントがエージェントに依頼する処理単位を表します。
主な属性
| 属性名 | 説明 |
|---|---|
taskId |
タスクの一意な識別子 |
status |
タスクの状態（例: submitted, completed） |
messages |
タスク内の会話履歴（Message の配列） |
artifacts |
タスク結果として生成される成果物（Artifact の配列） |
contextId |
タスクをグループ化するためのコンテキスト ID |
Message（メッセージ）
Message は、タスク内での会話の1往復（1メッセージ）を構成します。
主な属性
| 属性名 | 説明 |
|---|---|
role |
発言者の種別（user or agent） |
parts |
メッセージの内容（Part の配列） |
コード
A2A/samples/python/agents/ag2/task_manager.py
parts = [TextPart(type='text', text=agent_response['content'])]
artifact = None
if agent_response['require_user_input']:
task_status = TaskStatus(
state=TaskState.INPUT_REQUIRED,
message=Message(role='agent', parts=parts),
)
else:
task_status = TaskStatus(state=TaskState.COMPLETED)
artifact = Artifact(parts=parts)
# Update task store and get result for response
updated_task = await self.update_store(
task_id, task_status, None if artifact is None else [artifact]
)
Part（パート）
Part は、Message や Artifact の中身となる最小構成要素です。
主な属性
| 属性名 | 説明 |
|---|---|
type |
コンテンツの種別（例：text, file, data） |
text |
type=text の場合に含まれるテキスト内容 |
uri |
type=file の場合に参照されるファイル URI |
data |
type=data の場合に含まれる構造化データ（JSON等） |
Artifact（アーティファクト）
Artifact は、タスクが完了したときに生成される成果物です。
主な属性
| 属性名 | 説明 |
|---|---|
name |
アーティファクトの名前（任意） |
description |
アーティファクトの説明（任意） |
parts |
出力データ（Part の配列） |
コード
A2A/samples/python/agents/ag2/task_manager.py
if agent_response['require_user_input']:
task_status = TaskStatus(
state=TaskState.INPUT_REQUIRED,
message=Message(role='agent', parts=parts),
)
else:
task_status = TaskStatus(state=TaskState.COMPLETED)
artifact = Artifact(parts=parts)
Streaming（ストリーミング）
Streaming は、Server-Sent Events（SSE）によって、タスクの状態や出力をリアルタイムで送信します。
主な属性
| 属性名 | 説明 |
|---|---|
TaskStatusUpdateEvent |
タスクのステータス変更通知 |
TaskArtifactUpdateEvent |
アーティファクトの新規生成通知 |
コード
A2A/samples/python/agents/ag2/task_manager.py
task_update_event = TaskStatusUpdateEvent(
id=task_send_params.id, status=task_status, final=end_stream
)
await self.enqueue_events_for_sse(
task_send_params.id, task_update_event
)
Push Notifications（プッシュ通知）
Push Notifications は、クライアントが非接続時にも Webhook を通じて状態を受け取れる仕組みです。
主な属性
| 属性名 | 説明 |
|---|---|
notificationUrl |
通知先の Webhook URL（クライアントが事前に登録） |
jwt |
通知時に送信される署名付き JWT |
/.well-known/jwks.json |
公開鍵を取得するためのエンドポイント |
感想
A2Aがちょっとわかってきたぞ！
Discussion