Skip to content
記事
GoogleのA2Aプロトコルで2つのPythonエージェントを構築する方法 - ステップバイステップチュートリアル

チュートリアル: GoogleのA2Aプロトコルを使った2つのPythonエージェントの構築ステップバイステップ

Elwynn AI
Name
Elwynn AI

Updated on

このステップバイステップチュートリアルでは、GoogleのA2Aプロトコルを使用して2つの簡単なPythonエージェントを構築する方法を説明します。セットアップ、コード例、エージェント間の通信をカバーしています。A2Aはエージェント間の通信のためのオープンスタンダードであり、異なるフレームワークやベンダー間での相互運用性を実現します。

Googleのエージェント間（A2A）プロトコルは、どのフレームワークやベンダーから来たものであるかに関係なく、異なるAIエージェントがコミュニケーションをとり共同作業を行うためのオープンスタンダードです。より簡単に言うと、A2Aはエージェントが相互にやりとりするための共通言語を提供します。このチュートリアルでは、A2Aプロトコルを使用してお互いに通信する2つの簡単なPythonエージェントを作成する方法を説明します。1つはA2Aサーバーとして機能し、もう1つはA2Aクライアントとして機能します。セットアップをカバーし、A2Aの基本的な動作を説明し、両エージェントの注釈付きコード例を提供します。A2Aの事前知識は前提とせず、複雑なツールチェーンやマルチエージェントのオーケストレーションといった高度な概念を避け、初心者向けに親しみやすくしています。

A2Aプロトコルとは？

A2Aの基本は、エージェントがメッセージ、リクエスト、およびデータを交換する方法を標準化するプロトコルです。以下はA2Aの主要な概念のいくつかです：

  • エージェントカード: エージェントの機能、スキル、エンドポイントURL、認証要件を記述した公開メタデータファイル（通常 /.well-known/agent.json にホスト）。他のエージェントやクライアントはこのカードを取得して、エージェントが何をできるかを発見します。これは、A2Aネットワーク内でのエージェントの「名刺」に相当します。
  • A2Aサーバー: HTTP APIエンドポイントを公開し、A2Aメソッドを実装するエージェント（例えば、タスクを送信するためのエンドポイント）。リクエストを受け取って、他のエージェントに代わってタスクを実行します。
  • A2Aクライアント: A2AサーバーのURLにリクエストを送信してタスクや会話を開始するアプリケーションまたはエージェント。エージェント間のシナリオでは、1つのエージェントのクライアントコンポーネントがA2Aを通じて別のエージェントのサーバーコンポーネントを呼び出します。
  • タスク: A2Aの基本的な作業単位。クライアントはメッセージを送信することによりタスクを開始します（tasks/sendリクエストを使用）。各タスクにはユニークなIDがあり、状態のライフサイクルを持ちます（例：submitted, working, input-required, completed, など）。
  • メッセージ: クライアント（役割 "user"）とエージェント（役割 "agent"）の間のコミュニケーションの単一ターン。クライアントのリクエストはメッセージ（“user”からの）、エージェントの答えはメッセージ（“agent”からの）として構成されます。
  • パート: メッセージ内のコンテンツの一部。メッセージは1つまたは複数のパートで構成されます。パートはテキスト、ファイル、または構造化データであり、例えば、TextPartはプレーンテキストを持ち、ファイルパートは画像やその他のバイナリデータを持つかもしれません。このチュートリアルでは、わかりやすくするために単純なテキストパーツを使用します。

A2A通信の仕組み（基本フロー）： 1つのエージェントがA2Aを介して他のエージェントと通信したい場合、そのインタラクションは通常、標準化されたフローに従います：

  1. 発見: クライアントエージェントは最初に、サーバーの/.well-known/agent.json URLからエージェントカードを取得して他のエージェントを発見します。このカードはクライアントにエージェントの名前、できること、リクエストを送信する場所を教えます。
  2. 開始: その後、クライアントはサーバーエージェントのA2Aエンドポイントにリクエストを送信してタスクを開始します。通常、tasks/sendエンドポイントにPOSTとして送信します。この最初のリクエストにはユニークなタスクIDと初回のユーザーメッセージ（例：質問やコマンド）が含まれます。
  3. 処理: サーバーエージェントはタスクリクエストを受け取り、処理します。エージェントがストリーミング応答をサポートしている場合、（Server-Sent Eventsを使用して）作業しながら中間更新をストリームバックするかもしれませんが、我々の簡単な例ではストリーミングは使用しません。それ以外の場合、エージェントはタスクを処理し、完了時に最終的な応答を送信します。応答にはタスクの結果が含まれ、通常はエージェントからの1つまたは複数のメッセージ（およびタスクの一部である場合はファイルのようなアーティファクト）が含まれます。基本的なテキストインタラクションの場合、最終応答にはエージェントの回答メッセージが含まれます。

要約すると、A2Aはエージェントのための明確なリクエスト-レスポンスサイクルを定義しています：クライアントがエージェントの機能（エージェントカード）を見つけ、タスクを送信し（初回のユーザーメッセージと共に）、エージェントの応答を受け取る（メッセージとアーティファクトとして）、すべてが一貫したJSONフォーマットで行われます。では、環境をセットアップして、これを実際に確認するための最小限の例を構築しましょう。

インストールとセットアップ

コードを書く前に、必要なツールがインストールされているか確認してください：

  • Python 3.12以降 – A2Aサンプルでは互換性のためにPython 3.12+（さらにはPython 3.13）を推奨しています。
  • UV（Pythonパッケージマネージャ） – これは依存関係の管理とサンプルの実行用にA2Aプロジェクトが推奨するオプションのツールです。UVはモダンなパッケージマネージャですが、pip/venvを使用して手動で依存関係をインストールすることもできます。このチュートリアルでは、シンプルに通常のPythonツールを使用します。
  • Google A2Aリポジトリ – GitHubリポジトリをクローンして、公式のA2Aコードを取得します。 
    git clone https://github.com/google/A2A.git
    これによりプロジェクトがマシンにダウンロードされます。我々のチュートリアルコードはこのリポジトリの一部（特にサンプルA2Aクライアントユーティリティ）を参照します。
  • 依存関係 – UVを使用する場合、依存関係を自動で扱わせることができます。例えば、リポジトリのsamples/pythonディレクトリからuv sync（依存関係のインストール）を実行したり、公式READMEに示されたようにuv run ...を実行できます。pipを使用する場合は、例のコードに必要なライブラリを手動でインストールしてください。 
    pip install flask requests
    FlaskライブラリはシンプルなHTTPサーバーをエージェントに作成するために使用し、requestsはクライアントがエージェントを呼び出すために使用します。（公式のA2AサンプルコードはFastAPI/UVICornと内部クライアントライブラリを使用していますが、初心者向けのアプローチとしては、わかりやすくするためにFlaskとrequestsを使用します。）

シンプルなA2Aサーバーエージェントの作成

最初にサーバーサイドを構築します：タスクの受信をリッスンし応答するエージェントです。我々のエージェントは非常にシンプルで、テキストプロンプトを受け取り、フレンドリーメッセージと共にエコーするだけです。そのシンプルさにもかかわらず、A2Aのプロトコル構造に従うため、どのA2Aクライアントでも通信可能です。

エージェントの機能とエージェントカード: エージェントが発見可能になるためには、エージェントカードを提供する必要があります。実際のデプロイメントでは、サーバーの/.well-known/agent.jsonにJSONファイルをホストします。Flaskアプリの場合、このデータをエンドポイントを介して提供できます。エージェントカードには、通常、エージェントの名前、説明、A2AエンドポイントのURL、サポートされる機能（能力）などのフィールドが含まれます。我々はキーとなるいくつかのフィールドのみを持った最小限のエージェントカードを構築します。

tasks/sendエンドポイント: A2Aはエージェントが特定のAPIエンドポイントを実装することを期待します。最も重要なのはtasks/sendで、クライアントが新しいタスク（ユーザーメッセージを伴う）をエージェントに送るために呼び出すものです。我々のFlaskアプリは、これらのリクエストを処理するための/tasks/sendルートを定義します。入力されるJSON（タスクIDとユーザーのテキストメッセージを含む）を解析し、処理して（我々の場合はエコー応答を生成し）、A2Aフォーマットに従ったJSON応答を返します。

Flaskを使用して、シンプルなA2Aサーバーエージェントのコードを書いてみましょう：

from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
# Define the Agent Card data (metadata about this agent)
AGENT_CARD = {
    "name": "EchoAgent",
    "description": "A simple agent that echoes back user messages.",
    "url": "http://localhost:5000",  # The base URL where this agent is hosted
    "version": "1.0",
    "capabilities": {
        "streaming": False,           # This agent doesn't support streaming responses
        "pushNotifications": False    # No push notifications in this simple example
    }
    # (In a full Agent Card, there could be more fields like authentication info, etc.)
}
 
# Serve the Agent Card at the well-known URL.
@app.get("/.well-known/agent.json")
def get_agent_card():
    """Endpoint to provide this agent's metadata (Agent Card)."""
    return jsonify(AGENT_CARD)
 
# Handle incoming task requests at the A2A endpoint.
@app.post("/tasks/send")
def handle_task():
    """Endpoint for A2A clients to send a new task (with an initial user message)."""
    task_request = request.get_json()  # parse incoming JSON request
    # Extract the task ID and the user's message text from the request.
    task_id = task_request.get("id")
    user_message = ""
    try:
        # According to A2A spec, the user message is in task_request["message"]["parts"][0]["text"]
        user_message = task_request["message"]["parts"][0]["text"]
    except Exception as e:
        return jsonify({"error": "Invalid request format"}), 400
 
    # For this simple agent, the "processing" is just echoing the message back.
    agent_reply_text = f"Hello! You said: '{user_message}'"
 
    # Formulate the response in A2A Task format.
    # We'll return a Task object with the final state = 'completed' and the agent's message.
    response_task = {
        "id": task_id,
        "status": {"state": "completed"},
        "messages": [
            task_request.get("message", {}),             # include the original user message in history
            {
                "role": "agent",                        # the agent's reply
                "parts": [{"text": agent_reply_text}]   # agent's message content as a TextPart
            }
        ]
        # We could also include an "artifacts" field if the agent returned files or other data.
    }
    return jsonify(response_task)
 
# Run the Flask app (A2A server) if this script is executed directly.
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

サーバーコードの理解: 上記のコードで何が起こっているのか見てみましょう：

  • Flaskアプリを作成し、グローバルなAGENT_CARDディクショナリを定義しました。これには我々のエージェントの基本情報が含まれています。特に、名前、説明、自身を指すURL（ローカルホストとポート5000を指す）、および能力のセクションがあります。我々のエージェントはシンプルな同期応答だけを行うためstreaming: Falseを設定しました（エージェントがストリーミングをサポートする場合、tasks/sendSubscribeを処理し、部分結果をストリームしますが、ここではカバーしません）。
  • /.well-known/agent.jsonルートを設定し、エージェントカードJSONを返します。これは、発見のためにエージェントのカードが見つかると期待される標準的な場所を模しています。
  • /tasks/sendのPOSTルートを定義します。A2Aクライアントがタスクを送信すると、Flaskはhandle_task()を呼び出します。この関数内で：
    • リクエストのJSONボディをtask_requestに解析します。A2Aスキーマによれば、これは"id"（タスクID）とユーザーの最初のメッセージを表す"message"オブジェクトを含みます。
    • ユーザーのメッセージのテキストを抽出します。A2Aではメッセージには複数のパーツ（テキスト、ファイル）が含まれることがあります。ここでは最初のパーツがテキストであると仮定し、task_request["message"]["parts"][0]["text"]を取得します。より堅牢な実装では、パーツタイプを検証しエラーを処理します（我々のコードではシンプルなtry/exceptを行い、フォーマットが期待通りでない場合は400エラーを返します）。
    • 次にエージェントの応答を生成します。我々のロジックは微妙で、ユーザーテキストを用いて"Hello! You said: ..."を前置します。実際のエージェントでは、ここが主要なAIロジックが行われる場所です（LLMを呼び出したり、計算を実行したりします）。
    • 次に、応答Taskオブジェクトを構築します。クライアントがこれはどのタスクかを知るためにidを返し（つまりおよび終了したタスクの状態を"completed"として設定し、メッセージのリストを提供します。我々はオリジナルのユーザーメッセージを含め（文脈／履歴のため）、それに続いてエージェントの返事としてもう一つのメッセージを追加します。各メッセージにはrole（"user"または"agent"）とpartsのリストがあります。我々のエージェントのメッセージにはテキスト応答を含む単一のTextPartを使用します。
    • 最後に、これをJSONとして返します。FlaskはPythonの辞書をJSONにシリアライズしてくれます。

このサーバーが稼働している状態で、我々のエージェントは事実上http://localhost:5000で“ライブ”になります。エージェントカードを提供し、タスクリクエストを処理することができます。

シンプルなA2Aクライアントエージェントの作成

次に、サーバーエージェントと通信するクライアントが必要です。このクライアントは別のエージェントやユーザー向けアプリケーションでも良いです。A2A用語でそれはA2Aクライアントです。なぜなら、サーバーエージェントのA2A APIを使用するためです。チュートリアルとして、私たちは簡単なPythonスクリプトを書き、EchoAgentに質問を送信し、答えを受け取る役を果たします。

クライアントのステップ:

  1. エージェントの発見: クライアントはエージェントカードを取得してサーバーエージェントのエンドポイントや機能を知る必要があります。エージェントカードはhttp://localhost:5000/.well-known/agent.jsonにあると分かっているので、Pythonのrequestsライブラリを使用してこのURLをGETできます。
  2. タスクを送信: 次に、クライアントはエージェントのtasks/sendエンドポイントにタスクを表現するJSONペイロードをPOSTリクエストで送信します。このペイロードには以下が含まれます：
    • ユニークな"id"（タスクID、任意のユニークな文字列または番号で。例："task1"）
    • "message"オブジェクト（role: "user"とユーザーのクエリを含むTextPartから成るpartsリスト）
  3. 応答を受け取る: エージェントはタスクを表現するJSONで応答し、エージェントの返信メッセージを含めます。クライアントはこれを読み取り、エージェントの返信を抽出する必要があります。
調査ノート: OpenAIのGPT-4.5リリースの探求