⛳

Python A2A: Googleのエージェント間プロトコルの包括的ガイド

2025/04/15に公開

はじめに

Python A2AはGoogleのAgent-to-Agent（A2A）プロトコルの実装で、AIエージェント間の通信を標準化するために設計されています。このプロトコルは、AIエコシステムにおける大きな課題を解決します：異なるAIサービスがカスタム変換層なしでシームレスに通信できるようにすることです。

AIの分野が専門化されたサービスに分岐し、それぞれが独自のAPI形式とパラメータを持つようになるにつれて、開発者はAIロジックに集中するよりも通信インフラの構築に過剰な時間を費やしています。Python A2Aは、基盤となる実装に関係なく、AIエージェントが互いに通信する標準化された方法を提供することでこの問題に対処します。

Python A2Aを始める

インストール

# 基本インストール
pip install python-a2a

# OpenAI統合用
pip install "python-a2a[openai]"

# Anthropic Claude統合用
pip install "python-a2a[anthropic]"

# すべてのオプション依存関係用
pip install "python-a2a[all]"

主要概念

Python A2AはA2Aプロトコルからいくつかの重要な概念を実装しています：

メッセージ構造：テキスト、関数呼び出し、応答の定義された形式
会話スレッド：インタラクション間でコンテキストを維持するサポート
関数呼び出し：エージェントが関数を公開し呼び出すための標準化された方法
エラー処理：一貫したエラー形式

最初のA2Aエージェントを構築する

メッセージに応答する簡単なエコーエージェントから始めましょう：

from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server

class EchoAgent(A2AServer):
    """プレフィックス付きでメッセージをエコーバックする単純なエージェント。"""

    def handle_message(self, message):
        if message.content.type == "text":
            return Message(
                content=TextContent(text=f"Echo: {message.content.text}"),
                role=MessageRole.AGENT,
                parent_message_id=message.message_id,
                conversation_id=message.conversation_id
            )

# サーバーを実行
if __name__ == "__main__":
    agent = EchoAgent()
    run_server(agent, host="0.0.0.0", port=5000)

次に、エージェントと通信するクライアントを作成しましょう：

from python_a2a import A2AClient, Message, TextContent, MessageRole

# エージェントと通信するクライアントを作成
client = A2AClient("http://localhost:5000/a2a")

# メッセージを送信
message = Message(
    content=TextContent(text="こんにちは、これは動いていますか？"),
    role=MessageRole.USER
)
response = client.send_message(message)

# 応答を表示
print(f"エージェントの返答: {response.content.text}")

エージェント間の関数呼び出し

A2Aの強力な機能の一つは標準化された関数呼び出しです。こちらは数学関数を公開する計算機エージェントです：

import math
from python_a2a import (
    A2AServer, Message, TextContent, FunctionCallContent,
    FunctionResponseContent, FunctionParameter, MessageRole, run_server
)

class CalculatorAgent(A2AServer):
    """数学的計算機能を提供するエージェント。"""

    def handle_message(self, message):
        if message.content.type == "text":
            return Message(
                content=TextContent(
                    text="私は計算機エージェントです。以下の関数を呼び出せます：\n"
                         "- calculate: 基本的な算術演算（operation, a, b）\n"
                         "- sqrt: 平方根（value）"
                ),
                role=MessageRole.AGENT,
                parent_message_id=message.message_id,
                conversation_id=message.conversation_id
            )

        elif message.content.type == "function_call":
            function_name = message.content.name
            params = {p.name: p.value for p in message.content.parameters}

            try:
                if function_name == "calculate":
                    operation = params.get("operation", "add")
                    a = float(params.get("a", 0))
                    b = float(params.get("b", 0))

                    if operation == "add":
                        result = a + b
                    elif operation == "subtract":
                        result = a - b
                    elif operation == "multiply":
                        result = a * b
                    elif operation == "divide":
                        if b == 0:
                            raise ValueError("ゼロで割ることはできません")
                        result = a / b
                    else:
                        raise ValueError(f"不明な操作: {operation}")

                    return Message(
                        content=FunctionResponseContent(
                            name="calculate",
                            response={"result": result}
                        ),
                        role=MessageRole.AGENT,
                        parent_message_id=message.message_id,
                        conversation_id=message.conversation_id
                    )

                elif function_name == "sqrt":
                    value = float(params.get("value", 0))
                    if value < 0:
                        raise ValueError("負の数の平方根は計算できません")

                    result = math.sqrt(value)
                    return Message(
                        content=FunctionResponseContent(
                            name="sqrt",
                            response={"result": result}
                        ),
                        role=MessageRole.AGENT,
                        parent_message_id=message.message_id,
                        conversation_id=message.conversation_id
                    )

            except Exception as e:
                return Message(
                    content=FunctionResponseContent(
                        name=function_name,
                        response={"error": str(e)}
                    ),
                    role=MessageRole.AGENT,
                    parent_message_id=message.message_id,
                    conversation_id=message.conversation_id
                )

if __name__ == "__main__":
    agent = CalculatorAgent()
    run_server(agent, host="0.0.0.0", port=5001)

計算機エージェントの関数を呼び出す方法は次のとおりです：

from python_a2a import (
    A2AClient, Message, FunctionCallContent,
    FunctionParameter, MessageRole
)

client = A2AClient("http://localhost:5001/a2a")

# 関数呼び出しメッセージを作成
function_call = Message(
    content=FunctionCallContent(
        name="calculate",
        parameters=[
            FunctionParameter(name="operation", value="add"),
            FunctionParameter(name="a", value=5),
            FunctionParameter(name="b", value=3)
        ]
    ),
    role=MessageRole.USER
)

response = client.send_message(function_call)

if response.content.type == "function_response":
    result = response.content.response.get("result")
    if result is not None:
        print(f"結果: {result}")  # 出力: 結果: 8

LLM搭載エージェント

Python A2Aには人気のあるLLMプロバイダーとの統合がすぐに使えるように含まれています。こちらはOpenAI搭載のエージェントです：

import os
from python_a2a import OpenAIA2AServer, run_server

# OpenAIを利用したエージェントを作成
agent = OpenAIA2AServer(
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4",
    system_prompt="あなたは役立つAIアシスタントです。"
)

# サーバーを実行
if __name__ == "__main__":
    run_server(agent, host="0.0.0.0", port=5002)

同様に、Anthropic Claude搭載のエージェントも作成できます：

import os
from python_a2a import ClaudeA2AServer, run_server

# Anthropic Claudeを利用したエージェントを作成
agent = ClaudeA2AServer(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    model="claude-3-opus-20240229",
    system_prompt="あなたは役立つAIアシスタントです。"
)

# サーバーを実行
if __name__ == "__main__":
    run_server(agent, host="0.0.0.0", port=5003)

マルチエージェントワークフロー

A2Aの真の力は複数のエージェントを接続する時に発揮されます。リサーチアシスタントのワークフローを構築してみましょう：

from python_a2a import (
    A2AClient, Message, TextContent, MessageRole, Conversation
)

def research_workflow(query):
    # 専門エージェントに接続
    llm_client = A2AClient("http://localhost:5002/a2a")     # LLMエージェント
    search_client = A2AClient("http://localhost:5003/a2a")  # 検索エージェント
    summarize_client = A2AClient("http://localhost:5004/a2a")  # 要約エージェント

    # ワークフロー全体を会話で追跡
    conversation = Conversation()
    conversation.create_text_message(
        text=f"研究質問: {query}",
        role=MessageRole.USER
    )

    # ステップ1: 検索クエリを生成
    print("検索クエリを生成中...")
    search_request = Message(
        content=TextContent(
            text=f"この研究質問に基づいて: '{query}', "
                 f"関連する情報を見つけるのに役立つ3つの具体的な検索クエリを生成してください。"
        ),
        role=MessageRole.USER
    )
    search_queries_response = llm_client.send_message(search_request)
    conversation.add_message(search_queries_response)

    # ステップ2: 情報を取得
    print("情報を取得中...")
    search_message = Message(
        content=TextContent(
            text=f"回答するための情報を検索: {query}\n\n"
                 f"次のクエリを使用:\n{search_queries_response.content.text}"
        ),
        role=MessageRole.USER
    )
    search_results = search_client.send_message(search_message)
    conversation.add_message(search_results)

    # ステップ3: 情報を統合
    print("情報を統合中...")
    summarize_message = Message(
        content=TextContent(
            text=f"この情報を統合して質問に答えてください: '{query}'\n\n"
                 f"情報:\n{search_results.content.text}"
        ),
        role=MessageRole.USER
    )
    summary_response = summarize_client.send_message(summarize_message)
    conversation.add_message(summary_response)

    # 最終回答を会話に追加
    conversation.create_text_message(
        text=f"研究質問への回答:\n\n{summary_response.content.text}",
        role=MessageRole.AGENT
    )

    return conversation

# 使用例
if __name__ == "__main__":
    query = input("研究質問は何ですか？ ")
    result = research_workflow(query)
    print("\nリサーチ完了!")
    print("=" * 50)
    print(result.messages[-1].content.text)

高度な例: A2Aを使用した天気と旅行計画

A2Aがエージェント間の通信をどのように簡素化するかを示す例です：

from python_a2a import A2AClient, Message, TextContent, MessageRole

# A2Aを使用した場合: どのエージェント → 他のどのエージェントも
def plan_trip(location):
    # 専門エージェントに接続 - すべて同じプロトコルを使用
    weather_client = A2AClient("http://localhost:5001/a2a")
    openai_client = A2AClient("http://localhost:5002/a2a")

    # 天気エージェントに質問
    weather_message = Message(
        content=TextContent(text=f"{location}の天気予報はどうですか？"),
        role=MessageRole.USER
    )
    weather_response = weather_client.send_message(weather_message)

    # 天気情報を含めてOpenAIエージェントに質問
    planning_message = Message(
        content=TextContent(
            text=f"{location}への旅行を計画しています。天気予報: {weather_response.content.text}"
                 f"おすすめのアクティビティを提案してください。"
        ),
        role=MessageRole.USER
    )
    planning_response = openai_client.send_message(planning_message)

    return planning_response.content.text

# これはA2A互換の任意のエージェントで動作します - カスタムアダプターは不要！

会話の管理

Python A2Aは複数ターンのインタラクションを管理するためのConversationクラスを提供しています：

from python_a2a import Conversation, MessageRole, A2AClient

# 新しい会話を作成
conversation = Conversation()

# ユーザーメッセージを追加
conversation.create_text_message(
    text="今日の天気はどうですか？",
    role=MessageRole.USER
)

# エージェントに接続
weather_agent = A2AClient("http://localhost:5001/a2a")

# 会話の最後のメッセージを送信
last_message = conversation.messages[-1]
response = weather_agent.send_message(last_message)

# 応答を会話に追加
conversation.add_message(response)

# 会話を継続
conversation.create_text_message(
    text="ありがとう！傘は必要ですか？",
    role=MessageRole.USER
)

# 会話履歴を含む新しいメッセージを送信
response = weather_agent.send_message(
    conversation.messages[-1],
    conversation_id=conversation.conversation_id
)

# この応答も追加
conversation.add_message(response)

# 会話全体を表示
for msg in conversation.messages:
    role = "ユーザー" if msg.role == MessageRole.USER else "エージェント"
    print(f"{role}: {msg.content.text}")

エラー処理

Python A2Aは標準化されたエラー処理を提供します：

from python_a2a import A2AClient, Message, TextContent, MessageRole

client = A2AClient("http://localhost:5001/a2a")

try:
    message = Message(
        content=TextContent(text="こんにちは"),
        role=MessageRole.USER
    )
    response = client.send_message(message)

except ConnectionError as e:
    print(f"接続エラー: {e}")

except TimeoutError as e:
    print(f"タイムアウトエラー: {e}")

except Exception as e:
    print(f"予期しないエラー: {e}")

カスタムA2Aエージェントの作成

独自のA2A互換エージェントを作成するためのテンプレートです：

from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server

class MyCustomAgent(A2AServer):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # ここでエージェント固有のコンポーネントを初期化

    def handle_message(self, message):
        """受信メッセージを処理し、応答を返します。

        これはエージェントでメッセージを処理するためのメインエントリーポイントです。
        """
        if message.content.type == "text":
            # テキストメッセージを処理
            user_text = message.content.text

            # 応答を生成（あなたのロジックに置き換え）
            response_text = f"あなたが言ったこと: {user_text}"

            return Message(
                content=TextContent(text=response_text),
                role=MessageRole.AGENT,
                parent_message_id=message.message_id,
                conversation_id=message.conversation_id
            )

        elif message.content.type == "function_call":
            # 関数呼び出しを処理（エージェントがサポートしている場合）
            # ...
            pass

        # 他の条件に一致しない場合はデフォルト応答を返す
        return Message(
            content=TextContent(text="このメッセージタイプの処理方法がわかりません。"),
            role=MessageRole.AGENT,
            parent_message_id=message.message_id,
            conversation_id=message.conversation_id
        )

# エージェントを実行
if __name__ == "__main__":
    my_agent = MyCustomAgent()
    run_server(my_agent, host="0.0.0.0", port=5005)

結論

Python A2AはAIエージェントが通信するための標準化された方法を提供し、マルチエージェントシステムの構築を容易にします。このプロトコルを採用することで、以下のことが可能になります：

通信インフラ開発に費やす時間の削減
エージェントの実装を簡単に交換・アップグレード
モジュール式で拡張可能なAIシステムの構築
異なるタスクに特化したエージェントの活用

A2AプロトコルとそのPython実装は、AIシステムがレゴブロックのように組み合わせられる未来を構築し、専門化されたエージェントがシームレスに連携できるよう支援しています。

詳細については、Python A2A GitHubリポジトリ、PyPIページ、およびドキュメントを確認してください。