🚀

Agent Engineの新機能リリース A2Aサポートを利用してADKとAgent EngineでA2Aエージェントを作成

に公開
Google Cloud
Vertex AI
Agent Engine
A2A
Agent Development Kit
tech

こんにちは、サントリーこと大橋です。

2025年9月10日、Vertex AI Agent Engineに多くの新機能が追加されるアップデートがありました。Agent Development Kit(以降 ADK)の進化も目まぐるしいですが、それをホストするAgent Engineも負けじと進化を続けていますね。

https://cloud.google.com/vertex-ai/generative-ai/docs/release-notes#September_10_2025

リリースノートによると、主な新機能は以下の通りです。

  • Agent Engine Code Execution: Agentが分離された安全な実行環境(サンドボックス環境)でコードを実行できるようになりました(プレビュー)。
  • Agent-to-Agent (A2A) protocol support: A2AプロトコルをサポートするAgentをAgent Engine上で開発、デプロイ、利用できるようになりました。
  • Bidirectional streaming: 双方向ストリーミング(データの送受信を同時に行い、よりリアルタイムな対話を実現する通信方式)をサポートしました。
  • Memory Bank tab: Cloud ConsoleのUIにメモリを管理するための新しいタブが追加されました。

今回は特に注目度の高い「A2Aプロトコルサポート」にフォーカスして、その背景、実装方法、利用方法までを、実際に動作するサンプルコードと共に深掘りしていきたいと思います。

対象読者

  • ADK及びAgent Engineユーザー
  • AIエージェント間の連携に興味がある開発者
  • PythonでスケーラブルなAIエージェントシステムを構築したい方

※本文中ではADKのAgentを「Agent」とアルファベット表記、ADKだけではなく一般的なAIエージェントを「AIエージェント」とカタカナ表記しています。

課題: Agent EngineにおけるAgent間連携の不在

複雑なタスクを解決するために、複数の専門性を持ったAIエージェントを協調させる「マルチエージェントシステム」は、AIエージェント開発における重要なアーキテクチャの一つです。このエージェント間の通信を標準化するのが Agent-to-Agent (A2A) プロトコルです。

ADKはRemoteA2aAgentなどを通じてA2Aをサポートしていましたが、これまでAgent EngineにはA2Aプロトコルをネイティブにサポートする仕組みがありませんでした。

そのため、Agent Engine上でホストされたAgentをA2Aのエンドポイントとして公開するには、例えばCloud Runなどでプロキシサーバー(通信を『代理』で中継してくれるサーバー、今回はA2Aプロトコルを代理で解釈し、Agent Engineへ中継する役割を持つ)を自前で構築し、リクエストをAgent Engineに転送するといった、一手間かけたアーキテクチャを組む必要がありました。

こちらの記事などが、まさにその工夫の一例と言えるでしょう。
https://zenn.dev/google_cloud_jp/articles/3dadca4ff50f1f

このような状況では、手軽にマルチエージェントシステムを試したり、スケーラブルな環境にデプロイしたりすることが困難でした。

Agent EngineのA2Aサポート

今回のアップデートで、ついにAgent EngineがA2Aプロトコルをネイティブサポート(追加の仕組みを介さず、機能として直接サポートすること)しました。これにより、開発者はAgent Engine上にA2A準拠のエージェントを直接デプロイし、公開できるようになります。

本機能の公式ドキュメントは以下です。

https://cloud.google.com/vertex-ai/generative-ai/docs/agent-engine/develop/a2a?hl=en
https://cloud.google.com/vertex-ai/generative-ai/docs/agent-engine/use/a2a?hl=en

ここからは、実際に動作するサンプルを例に、その手順を追っていきましょう。

1. 環境構築

まず、uv を使って必要なライブラリをインストールします。ADK 1.14.1以降、a2a-sdk 0.3.5以降、google-cloud-aiplatform 1.112.0以降が必要です。

uv add "google-adk[a2a] @ git+https://github.com/google/adk-python" --branch main
uv add "a2a-sdk>=0.3.5" "google-cloud-aiplatform[adk,agent-engines]>=1.112.0"

2. A2Aサーバーサイドエージェントの作成

まず、Agent Engine上で動作させるA2Aエージェント(サーバーサイド)を作成します。今回は現在時刻を返すシンプルな「時計エージェント」を例にします。

adk create ae_a2a コマンドで雛形を作成した後、agent.pyを以下のように編集します。

ae_a2a/agent.py
import datetime
from google.adk.agents.llm_agent import Agent

async def now_tool():
    """
    現在時刻を返却します。返却する時刻はUTCです。
    """
    return datetime.datetime.now(tz=datetime.timezone.utc).isoformat()

root_agent = Agent(
    model='gemini-2.5-flash',
    name='root_agent',
    description='時計エージェント',
    instruction="""
    あなたは現在時刻を教える時計エージェントです。 `now_tool` を 利用して、現在時刻を取得してユーザーの問いに答えてください。

    [出力フォーマット]
    JSON オブジェクト

    [出力スキーマ]
    time: string iso 8601 形式 
    例: {"time": "2022-09-04T16:07:48.53+0000"}
    """,
    tools=[now_tool]
)

3. A2A Agent ExecutorとAgent Cardの作成

次に、作成したADK AgentをA2Aプロトコルで動作させるためのAgentExecutorと、エージェントの名刺であるAgentCardを作成します。ADKにはこれらを簡単に生成するユーティリティが用意されています。

ae_a2a/agent_executor.py
from a2a.types import AgentCard
from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.memory import InMemoryMemoryService
from google.adk.artifacts import InMemoryArtifactService
from google.adk.auth.credential_service.in_memory_credential_service import InMemoryCredentialService
from google.adk.a2a.utils.agent_card_builder import AgentCardBuilder

# ADKで作成したAgentをインポート
from .agent import root_agent

async def create_runner() -> Runner:
    """ADK Agentを実行するためのRunnerを生成します。"""
    return Runner(
        app_name=root_agent.name or "adk_agent",
        agent=root_agent,
        # 実際のアプリケーションでは永続化されたサービスを利用します
        artifact_service=InMemoryArtifactService(),
        session_service=InMemorySessionService(),
        memory_service=InMemoryMemoryService(),
        credential_service=InMemoryCredentialService(),
    )


# A2A Agent Executorを生成するビルダー関数
def agent_executor_builder():
    return A2aAgentExecutor(
        runner=create_runner,
    )

# Agent Cardを生成する非同期関数
async def create_agent_card() -> AgentCard:
    builder = AgentCardBuilder(agent=root_agent)
    return await builder.build()

4. Agent Engineへのデプロイ

作成したA2AエージェントをAgent Engineにデプロイします。まず、Google CloudのプロジェクトIDなどを.envファイルに設定します。

GOOGLE_CLOUD_PROJECT=ご自身のGoogle CloudプロジェクトIDに置き換えてください
GOOGLE_CLOUD_LOCATION=us-central1

次に、デプロイ用のスクリプトを作成します。vertexai SDKのclient.agent_engines.create()メソッドを使用します。
※ STAGING_BUCKETに記載してあるGoogle Cloud Storageのバケットを事前に作成しておいてください。

deploy.py
import asyncio
import os
import vertexai
from dotenv import load_dotenv
from google.genai import types
from vertexai.preview.reasoning_engines import A2aAgent
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card
from ae_a2a.agent_executor import agent_executor_builder, create_agent_card as create_adk_agent_card

load_dotenv()

PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
STAGING_BUCKET = f"gs://{PROJECT_ID}-agent-engine-bucket"

vertexai.init(project=PROJECT_ID, location=LOCATION, staging_bucket=STAGING_BUCKET)

async def main():
    # ADKのユーティリティでAgentCardの情報を生成
    adk_agent_card = await create_adk_agent_card()

    # Agent Engineが要求する形式にAgentCardを変換
    agent_card = create_agent_card(agent_name=adk_agent_card.name, description=adk_agent_card.description, skills=adk_agent_card.skills)
    
    # デプロイ用のA2aAgentオブジェクトを作成
    a2a_agent = A2aAgent(agent_card=agent_card, agent_executor_builder=agent_executor_builder)
    a2a_agent.set_up()

    client = vertexai.Client(
        project=PROJECT_ID,
        location=LOCATION,
        http_options=types.HttpOptions(
            api_version="v1beta1", base_url=f"https://{LOCATION}-aiplatform.googleapis.com/"
        ),
    )
        
    remote_a2a_agent = client.agent_engines.create(
        agent=a2a_agent,
        config={
            "display_name": a2a_agent.agent_card.name,
            "description": a2a_agent.agent_card.description,
            "requirements": [
                "google-cloud-aiplatform[agent_engines,adk]>=1.112.0",
                "a2a-sdk>=0.3.5",
                "cloudpickle==3.0"
            ],
            "extra_packages": ["ae_a2a"],
            "http_options": {
                "base_url": f"https://{LOCATION}-aiplatform.googleapis.com",
                "api_version": "v1beta1",
            },
            "staging_bucket": STAGING_BUCKET,
        },
    )

    print(remote_a2a_agent)

if __name__ == "__main__":
    asyncio.run(main())

このスクリプトを実行してデプロイします。

uv run deploy.py

5. A2Aクライアントエージェントの作成

最後に、デプロイしたAgent Engine上のA2Aエージェントを呼び出すクライアントを作成します。ADKのRemoteA2aAgentを使い、オーケストレーター(オーケストラの『指揮者』のように、複数のエージェントを協調させてタスクを達成させる役割)となる親Agentのサブエージェントとして組み込みます。

まず、.envファイルにデプロイしたAgent EngineのIDを設定します。

ae_client_agent/.env
GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=ご自身のGoogle CloudプロジェクトIDに置き換えてください
GOOGLE_CLOUD_LOCATION=us-central1
REASONING_ENGINE_ID=Agent EngineのID(数値)を設定してください

次に、クライアントとなるAgentを定義します。

ae_client_agent/agent.py
import os
import httpx
from a2a.client import ClientConfig, ClientFactory
from a2a.types import TransportProtocol
from google.adk import Agent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.auth import default
from google.auth.transport.requests import Request

# .envから設定を読み込み
PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
REASONING_ENGINE_ID = os.getenv("REASONING_ENGINE_ID")
AGENT_ENGINE_RESOURCE = f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{REASONING_ENGINE_ID}"

# GCPの認証情報を取得
credentials, _ = default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
credentials.refresh(Request())

# A2Aクライアントファクトリを認証情報付きで設定
factory = ClientFactory(
    ClientConfig(
        supported_transports=[TransportProtocol.http_json],
        use_client_preference=True,
        httpx_client=httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {credentials.token}",
                "Content-Type": "application/json",
            }
        ),
    )
)

a2a_url = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{AGENT_ENGINE_RESOURCE}/a2a"

# A2Aクライアントエージェント(リモートの時計エージェントを指す)
time_agent = RemoteA2aAgent(
    name="time_agent",
    description="時計エージェント",
    agent_card=f"{a2a_url}/v1/card",
    a2a_client_factory=factory,
)

# オーケストレーションエージェント
root_agent = Agent(
    name="root_agent",
    model="gemini-2.5-flash",
    description="You are helpful assistant for user questions.",
    instruction="""
    あなたはたいへん役に立つ、ユーザーの質問に答えるAIエージェントです。
    ユーザーが時間について問い合わせた場合は、必ず毎回 time_agent に問い合わせ、現在時刻を取得した後に、ユーザーにわかりやすい形式に修正したうえで丁寧に答えてください。
    """,
    sub_agents=[time_agent],
)

ClientFactoryに認証情報を設定することで、IAM(Google Cloudのアクセス権限を管理するサービス)で保護されたAgent Engineのエンドポイント(サービスと通信するための接続窓口となるURLなど)に安全にアクセスできます。

6. 実行

作成したクライアントエージェントをadk webで実行します。

adk web .

ブラウザでhttp://localhost:8000にアクセスし、「今何時？」と質問すると、ローカルのroot_agentがAgent Engine上のtime_agentを呼び出し、現在時刻を取得して応答する様子が確認できるはずです。


adk webでA2A経由で現在時刻を取得し返却 プロンプトが雑なのでJSONそのままですね...

まとめ

今回は、Agent Engineの大型アップデートの中から、特に待望されていたA2Aプロトコルのネイティブサポートについて、実際に動作するサンプルを交えて解説しました。

これまで自前でプロキシを立てる必要があったAgent間連携が、Agent Engine上で直接サポートされたことで、スケーラブルで堅牢なマルチエージェントシステムの構築が格段に容易になりました。ADKのA2aAgentExecutorやRemoteA2aAgentと組み合わせることで、簡単にローカルでの開発からAgent Engineへのデプロイ、そして利用までをシームレスに行うことができます。

Code Executionや双方向ストリーミングなど、他の新機能も非常に強力であり、Agent EngineがエンタープライズレベルのAIエージェントプラットフォームとして大きく飛躍したことを感じさせるアップデートでした。これらの機能についても、また別の記事で深掘りしていきたいと思います。

最後にお知らせ/宣伝

先日、ADKユーザー向けのイベント「ADK UG #0」が開催され、ADK開発者が集う日本語のDiscordコミュニティが誕生しました。ADKに関する情報交換や議論に興味がある方は、ぜひご参加ください！

https://discord.gg/BKpGRzjtqZ

また、ADKの最新のコミットログやリリースノートを分かりやすく解説するPodcastを、月・水・金に配信しています。ADKの動向を追いかけたい方は、ぜひ聴いてみてください。

https://www.youtube.com/playlist?list=PL0Zc2RFDZsM_MkHOzWNJpaT4EH5fQxA8n

Discussion