Python Quickstart Tutorial: Building an A2A Agentを試す
A2Aの基本チュートリアルを試して概念を理解していきます（基本的には下記を翻訳しつつ、気になったところを加筆してまとめています）
セットアップ
git clone
チュートリアルに従って、a2a-samplesをクローンします
git clone https://github.com/a2aproject/a2a-samples.git -b main --depth 1
cd a2a-samples
Pythonの環境設定とSDKのインストール
- Mac/Linuxの場合（Windowsの場合は上記のチュートリアルページを参照ください）
python -m venv .venv
source .venv/bin/activate
- Python 依存関係と A2A SDK およびその依存関係をインストール
pip install -r samples/python/requirements.txt
- インストールのチェック
下記を実行して「A2A SDK imported successfully」が表示されればOK！
python -c "import a2a; print('A2A SDK imported successfully')"
Agent Skills & Agent Card
実際に動かす前に、ここから少し座学になります
「Python Quickstart Tutorial: Building an A2A Agent」ではhelloworldを返すシンプルな例で確認していきます
a2a-samples/samples/python/agents/helloworld/に該当サンプルがあるので、そちらに移動して確認していきましょう
Agent Skills
エージェントスキルは、エージェントが実行できる特定の能力または機能を表します。
エージェントがどのようなタスクに適しているかをクライアントに伝えるための要素です。
AgentSkillsに関する主な属性は下記の通りで、a2a.types内で定義されます
| キー名 | 説明 |
|---|---|
| id | スキルに割り当てられた一意の識別子。 |
| name | 人間が読解可能な名前。 |
| description | スキルが何を行うかについての、より詳細な説明。 |
| tags | カテゴリ分けや発見のためのキーワード。 |
| examples | プロンプトのサンプルまたは使用例。 |
| inputModes / outputModes | 入力および出力でサポートされるメディアタイプ（例：「text/plain」、「application/json」）。 |
実際にHelloworld agentの__main__.pyの中をみてみると、下記のように定義されていることが確認できます
skill = AgentSkill(
id='hello_world',
name='Returns hello world',
description='just returns hello world',
tags=['hello world'],
examples=['hi', 'hello world'],
)
Agent Card
エージェントカードは、A2Aサーバーが通常 .well-known/agent-card.json エンドポイントで提供するJSONドキュメントです。エージェントのデジタル名刺のようなものです。
（個人的追記）
下記のエージェントを動作させたときにも登場しましたが、ホストエージェントがリモートエージェントとの連携登録する際に、エージェントカードの取得のリクエストを最初に行い、応答の有無でリモートエージェントの存在を確認することで、存在しないリモートエージェントとの誤連携を防いでいるように思います
Agent Cardに関する主な属性は下記の通りで、こちらもa2a.types内で定義されます
| キー名 | 説明 |
|---|---|
| name | 基本的な識別情報（名前）。 |
| description | 基本的な識別情報（説明）。 |
| version | 基本的な識別情報（バージョン）。 |
| url | A2Aサービスにアクセスできるエンドポイント。 |
| capabilities | ストリーミングやプッシュ通知など、サポートされるA2A機能。 |
| defaultInputModes / defaultOutputModes | エージェントのデフォルトの入出力メディアタイプ。 |
| skills | エージェントが提供するAgentSkillオブジェクトのリスト。 |
実際にHelloworld agentの__main__.pyの中をみてみると、下記のように定義されていることが確認できます
public_agent_card = AgentCard(
name='Hello World Agent',
description='Just a hello world agent',
url='http://localhost:9999/',
version='1.0.0',
default_input_modes=['text'],
default_output_modes=['text'],
capabilities=AgentCapabilities(streaming=True),
skills=[skill], # Only the basic skill for the public card
supports_authenticated_extended_card=True,
)
上記を確認することで、
エージェントの名前が "Hello World Agent", http://localhost:9999/でアクセス可能、テキストでの入出力が可能、スキルとして前述のAgent Skillで定義したhello_worldスキルを持つ、ということがわかります
（個人的追記）
supports_authenticated_extended_card=Trueがパッとわからない
認証したあとに参照可能な拡張されたAgent Cardがありそうではあるが・・・
The Agent Executor
Agent ExecutorがA2Aのエージェントがリクエストを処理し、レスポンス/イベントを生成するためのコアロジック
AgentExecutor Interface
AgentExecutorクラスが定義する２つの重要な関数
RequestContextは、ユーザーのメッセージや既存のタスクの詳細など、受信リクエストに関する情報を提供します。
EventQueue は、Executor がクライアントにイベントを返すために使用されます。
| 関数 | 説明 |
|---|---|
| async def execute(self, context: RequestContext, event_queue: EventQueue) | レスポンスまたはイベントストリームを期待する受信リクエストを処理します。ユーザー入力（コンテキスト経由で取得可能）を処理し、event_queue を使用して Message、Task、TaskStatusUpdateEvent、または TaskArtifactUpdateEvent オブジェクトを返します。 |
| async def cancel(self, context: RequestContext, event_queue: EventQueue) | 実行中のタスクをキャンセルするリクエストを処理します。RequestContext は、ユーザーのメッセージや既存のタスクの詳細など、受信リクエストに関する情報を提供します。EventQueue は、Executor がクライアントにイベントを返すために使用されます。 |
Helloworld Agent Executor
実際にHelloworld AgentのAgent Executorを見ていきます
agent_executor.py の中にHelloWorldAgentExecutorが定義されています
1. HelloWorldAgent
まず、HelloWorldAgentのクラスを見てみます。
シンプルにinvokeされたら、"Hello World"を返すようにしています
class HelloWorldAgent:
"""Hello World Agent."""
async def invoke(self) -> str:
return 'Hello World'
2. HelloWorldAgentExecutor
AgentExecutorを継承して、HelloWorldAgentExecutorを用意しています
また、HelloWorldAgentで初期化を行っています
__init__:
class HelloWorldAgentExecutor(AgentExecutor):
"""Test AgentProxy Implementation."""
def __init__(self):
self.agent = HelloWorldAgent()
また、AgentExecutorから継承したexecuteをオーバーライドしています
execute:
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))
message/send または message/stream リクエストが到着すると、（=ストリーミングであろうがなかろうが）executorで以下の処理が実行されます。
self.agent.invoke() を呼び出して「Hello World」という文字列を取得します。
new_agent_text_message ユーティリティ関数を使用して A2A メッセージオブジェクトを作成します。
このメッセージを event_queue にエンキューします。
すると、DefaultRequestHandler がこのキューを処理し、クライアントにレスポンスを送信します。
cancel: Hello Worldの例ではcancel実行時は例外処理を実行するシンプルな実装になっていますが、適宜ここも改修することでハンドリングすることができそうです
async def cancel(
self, context: RequestContext, event_queue: EventQueue
) -> None:
raise Exception('cancel not supported')
サーバ起動〜リクエストしてみる
サーバ起動
__main__.pyを実行すればサーバが起動します
python samples/python/agents/helloworld/__main__.py
クライアントからリクエスト
test_client.pyが用意されているので、それを実行します
まず、ターミナルで別ウィンドウを開きます
セットアップのセクションにてa2a-samplesディレクトリで仮想環境を作成しているため、新しく開いたターミナルにて、そのディレクトリに移動し、再度下記を実行します
Mac/Linuxの場合（Windowsの場合は上記のチュートリアルページを参照ください）
source .venv/bin/activate
続いて、クライアントを実行します
python samples/python/agents/helloworld/test_client.py
前述の想定のとおり、まずは公開されているAgentCardをチェックし、AgentCardがあれば、さらに認証済みの場合の拡張版AgentCardがないかをチェックすることで、認証の有無でのエージェントのスキルの変更を実現しています
test_client.pyの該当箇所
resolver = A2ACardResolver(
httpx_client=httpx_client,
base_url=base_url,
# agent_card_path uses default, extended_agent_card_path also uses default
)
# --8<-- [end:A2ACardResolver]
# Fetch Public Agent Card and Initialize Client
final_agent_card_to_use: AgentCard | None = None
try:
logger.info(
f'Attempting to fetch public agent card from: {base_url}{AGENT_CARD_WELL_KNOWN_PATH}'
)
_public_card = (
await resolver.get_agent_card()
) # Fetches from default public path
logger.info('Successfully fetched public agent card:')
logger.info(
_public_card.model_dump_json(indent=2, exclude_none=True)
)
final_agent_card_to_use = _public_card
logger.info(
'\nUsing PUBLIC agent card for client initialization (default).'
)
if _public_card.supports_authenticated_extended_card:
try:
logger.info(
f'\nPublic card supports authenticated extended card. Attempting to fetch from: {base_url}{EXTENDED_AGENT_CARD_PATH}'
)
auth_headers_dict = {
'Authorization': 'Bearer dummy-token-for-extended-card'
}
_extended_card = await resolver.get_agent_card(
relative_card_path=EXTENDED_AGENT_CARD_PATH,
http_kwargs={'headers': auth_headers_dict},
)
logger.info(
'Successfully fetched authenticated extended agent card:'
)
logger.info(
_extended_card.model_dump_json(
indent=2, exclude_none=True
)
)
final_agent_card_to_use = (
_extended_card # Update to use the extended card
)
logger.info(
'\nUsing AUTHENTICATED EXTENDED agent card for client initialization.'
)
except Exception as e_extended:
logger.warning(
f'Failed to fetch extended agent card: {e_extended}. Will proceed with public card.',
exc_info=True,
)
elif (
_public_card
): # supports_authenticated_extended_card is False or None
logger.info(
'\nPublic card does not indicate support for an extended card. Using public card.'
)
AgentCardの確認後、A2AClientにAgentCardを設定し、リクエストしています
test_client.pyの該当箇所
# --8<-- [start:send_message]
client = A2AClient(
httpx_client=httpx_client, agent_card=final_agent_card_to_use
)
logger.info('A2AClient initialized.')
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))
# --8<-- [end:send_message]
# --8<-- [start:send_message_streaming]
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))
# --8<-- [end:send_message_streaming]
期待出力
チュートリアルには下記が期待出力とありますが、
The streaming response (a single "Hello World" message as one chunk, after which the stream ends).と記載があるためか、
Streaming responseのfinalのキーは返却されませんでした（別途ドキュメントの確認が必要です..）
// Non-streaming response
{"jsonrpc":"2.0","id":"xxxxxxxx","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"yyyyyyyy"}}
// Streaming response (one chunk)
{"jsonrpc":"2.0","id":"zzzzzzzz","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"wwwwwwww","final":true}}
まとめ（ざっくり知りたい方はこちらを参照ください）
１つのチュートリアルでしたが、しっかり読んでいくとそこそこもボリュームでした
簡単にまとめると下記になりますので、概要知りたい方は下記を参照ください！
- AgentSkillsでAgentのスキルが何を行うかや使用例を宣言
- AgentCardでクライアントがスキル情報やアクセス情報を提供
- クライアントはAgentCardを最初にチェックし、Agentの詳細を把握（認証有無で機能が変わる場合は拡張版AgentCardも取得）し、その情報を用いて、Agentにリクエストを送る
Discussion