「AIエージェントキャッチアップ #30 - A2A (Agent2Agent)」を開催しました

ジェネラティブエージェンツの大嶋です。

「AIエージェントキャッチアップ #30 - A2A (Agent2Agent)」という勉強会を開催しました。

generative-agents.connpass.com

アーカイブ動画はこちらです。

www.youtube.com

A2A (Agent2Agent)

今回は、Googleが発表したAIエージェント間の連携プロトコル「A2A (Agent2Agent)」について、公式ブログやドキュメントを読んだり、チュートリアルを試しました。

公式ブログはこちらです。

cloud.google.com

GitHubリポジトリはこちらです。

github.com

ドキュメントはこちらです。

google.github.io

今回のポイント

A2Aの概要

A2Aは、AIエージェントが他のAIエージェントと連携するためのプロトコルです。

A2AではAIエージェントの連携のための公開・呼び出しのプロトコルが定められており、プロトコルに従ったPython・JavaScriptのサーバー・クライアントの実装が提供されています。

A2Aの主要概念 - Agent Card

A2Aでは、公開するエージェントの情報をJSON形式の「Agent Card」として提供します。

以下のようにエージェントの名前・説明・対応する入出力モード・スキルなどを記述します。

{
  "name": "Google Maps Agent",
  "description": "Plan routes, remember places, and generate directions",
  "url": "https://maps-agent.google.com",
  "provider": {
    "organization": "Google",
    "url": "https://google.com"
  },
  "version": "1.0.0",
  "authentication": {
    "schemes": "OAuth2"
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain", "application/html"],
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "route-planner",
      "name": "Route planning",
      "description": "Helps plan routing between two locations",
      "tags": ["maps", "routing", "navigation"],
      "examples": [
        "plan my route from Sunnyvale to Mountain View",
        "what's the commute time from Sunnyvale to San Francisco at 9AM",
        "create turn by turn directions from Sunnyvale to Mountain View"
      ],
      // can return a video of the route
      "outputModes": ["application/html", "video/mp4"]
    },
      :
  ]
}

引用元: https://google.github.io/A2A/#/documentation?id=agent-card-1

Agent Cardはhttps://example.com/.well-known/agent.jsonのようなURLで公開することが推奨されています。

A2Aの主要概念 - Task

A2Aでは、エージェントの呼び出しで「Task」という概念がコアなオブジェクトになります。

Taskは「Artifact」や「Messages」と紐づきます。

Taskが重要なオブジェクトである点は、E2BのAgent Protocol とも似ていますね。

A2AのTaskは、以下のステータスを持ちます。

• submitted
• working
• input-required
• completed
• canceled
• failed
• unknown

あとで少し説明しますが、とくに「input-requrired」というステータスがあることが面白いですね。

A2Aサーバーの実装の概要

A2Aのドキュメントには、Pythonを使ってA2Aサーバーを実装してみるチュートリアルがあります。

google.github.io

A2Aサーバーを実装する際は、大きく3つのコードを書くことになります。

1. A2Aサーバーを起動するmainのコードの実装
2. エージェントの呼び出しのハンドラー的な存在であるTaskManagerの実装
3. エージェントの実装

このうち3つ目のエージェントの実装については、LangGraphやCrewAI、ADKなどを含め、自由に実装することができます。

こういった各種フレームワークを使った実装例は、公式リポジトリで公開されています。

github.com

A2Aサーバーを起動するmainのコードの実装

A2Aサーバーを起動する際は、google_a2aからA2AServerをimportして、AgentCardとTaskManagerを設定します。

from google_a2a.common.server import A2AServer
from google_a2a.common.types import AgentCard

agent_card = AgentCard(
    name="sample-agent",
    description="サンプルエージェント",
    url=f"http://{host}:{port}/",
    version="0.1.0",
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    capabilities=capabilities,
    skills=[skill],
)

server = A2AServer(
    agent_card=agent_card,
    task_manager=task_manager,
    host=host,
    port=port,
)
server.start()

TaskManagerの実装

TaskManagerは、google_a2aが提供するInMemoryTaskManagerなどを継承して、on_send_taskとon_send_task_subscribeを実装します。

from google_a2a.common.server.task_manager import InMemoryTaskManager

class MyAgentTaskManager(InMemoryTaskManager):
        :
    async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:
        # エージェントが非Streamingで呼び出された場合の処理を実装

    async def on_send_task_subscribe(
        self,
        request: SendTaskStreamingRequest,
    ) -> AsyncIterable[SendTaskStreamingResponse] | JSONRPCResponse:
        # エージェントがStreamingで呼び出された場合の処理を実装

• on_send_task：エージェントが非Streamingで呼び出された場合の処理を実装
• on_send_task_subscribe：エージェントがStreamingで呼び出された場合の処理を実装

on_send_taskやon_send_task_subscribeの中では、Taskをupsert・updateしたり、クライアントに応答を返したりします。

シンプルなon_send_taskの例は以下のようになります。

    async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:
        # タスクをupsert
        await self.upsert_task(request.params)

        # 実際のエージェントの呼び出しなど
        response_text = ...

        # タスクをcompletedに更新
        task = await self._update_task(
            task_id=request.params.id,
            task_state=TaskState.COMPLETED,
            response_text=response_text,
        )

        return SendTaskResponse(id=request.id, result=task)

このように、A2Aサーバーを実装する際は、コード上にはLangGraph・CrewAI・ADKなどでのエージェントの実装に加えて、A2Aに従った実装という層も加わることになります。

もしもLangGraph用のA2Aサーバーのアダプターのようなものが提供されたりすると、簡単にA2Aサーバーを公開できて便利かもしれませんね。

input-required

最後に、A2AのTaskには、input-requiredというステータスがあります。

input-requiredは、クライアントに対して追加情報を求めているといったステータスです。

Human-in-the-Loopと似ていますが、A2Aではエージェントがエージェントを呼び出すので、人間ではなく呼び出しもとのエージェントに追加情報を求めるわけです。

似た例で言えば、LangGraphの Agent Protocol でもinterrupt中のスレッドを検索できるようになっています。

エージェントのフレームワークやプロトコルを考える上で、Human-in-the-Loopのような処理をサポートすることは重要だと思っているので、A2Aがinput-requiredというかたちでサポートしているのは面白いと思いました。

次回のご案内

以上、今回は「A2A (Agent2Agent)」をキャッチアップしました。

次回は「AIエージェントキャッチアップ #31 - OpenAI Codex CLI」ということで、OpenAIのコーディングエージェント「OpenAI Codex CLI」がテーマです！

generative-agents.connpass.com

ご興味・お時間ある方はぜひご参加ください！

また、その次の回以降のテーマも募集しているので、気になるエージェントのOSSなどあれば教えてください！