はじめに
下記の記事では、「ネット記事の作成業務」のワークフローを実行するマルチエージェントシステムを Agent Engine にデプロイして、A2A で連携させる方法を説明しました。
この際、Agent Engine の前段に Cloud Run を用いて A2A サーバーをデプロイしましたが、Agent Engine が A2A 対応したため、A2A サーバーを別途デプロイする必要がなくなりました。 また、ADK が標準提供する RemoteA2aAgent クラスを利用すると、ローカルエージェントとほぼ同じ方法で Agent Engine にデプロイした A2A リモートエージェントをサブエージェントに組み込むことができます。つまり、「ネット記事の作成業務」のワークフローを実行するマルチエージェントシステムは、次の図のように実現されます。
ADK + Agent Engine による A2A マルチエージェントシステムの構成
この記事では、上図のシステムをデプロイする手順を具体的に説明します。
環境準備
Vertex AI workbench のノートブック上で実装しながら説明するために、まずは、ノートブックの実行環境を用意します。新しいプロジェクトを作成したら、Cloud Shell のコマンド端末を開いて、必要な API を有効化します。
gcloud services enable \
aiplatform.googleapis.com \
notebooks.googleapis.com \
cloudresourcemanager.googleapis.com
Agent Engine へのデプロイに必要なサービスアカウントを作成して、IAM ロールを付与しておきます。
PROJECT_ID=$(gcloud config list --format 'value(core.project)' 2>/dev/null)
PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format='value(projectNumber)' 2>/dev/null)
gcloud beta services identity create --service=aiplatform.googleapis.com --project=$PROJECT_ID
# Agent Engine 上の root agent が remote a2a を呼び出すために必要
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-aiplatform-re.iam.gserviceaccount.com" \
--role='roles/aiplatform.user'
続いて、Workbench のインスタンスを作成します。
PROJECT_ID=$(gcloud config list --format 'value(core.project)')
gcloud workbench instances create agent-development \
--project=$PROJECT_ID \
--location=us-central1-a \
--machine-type=e2-standard-2
クラウドコンソールのナビゲーションメニューから「Vertex AI」→「Workbench」を選択すると、作成したインスタンス agent-development があります。インスタンスの起動が完了するのを待って、「JUPYTERLAB を開く」をクリックしたら、「Python 3(ipykernel)」の新規ノートブックを作成します。
この後は、ノートブックのセルでコードを実行していきます。
まず、Agent Development Kit (ADK) と A2A SDK を含めた、必要なパッケージをインストールします。
%pip install --user --upgrade \
google-adk==1.27.2 \
google-genai==1.68.0 \
google-cloud-aiplatform==1.141.0 \
a2a-sdk==0.3.25
インストール時に表示される ERROR: pip's dependency resolver does not currently take into... というエラーは無視してください。
インストールしたパッケージを利用可能にするために、次のコマンドでカーネルを再起動します。
import IPython
app = IPython.Application.instance()
_ = app.kernel.do_shutdown(True)
再起動を確認するポップアップが表示されるので [Ok] をクリックします。
続いて、必要なモジュールをインポートして、実行環境の初期設定を行います。
import asyncio, datetime, httpx, json, os
from IPython.display import Markdown, display
import google.auth
from google.auth.transport.requests import AuthorizedSession
from google.auth.transport.requests import Request
import vertexai
from vertexai.agent_engines import AdkApp
# ADK
from google.genai.types import Part, Content, HttpOptions
from google.adk.agents.callback_context import CallbackContext
from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.artifacts import InMemoryArtifactService
from google.adk.auth.credential_service.in_memory_credential_service import InMemoryCredentialService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.models import LlmResponse, LlmRequest
from google.adk.runners import Runner
from google.adk.sessions import VertexAiSessionService
# A2A
from a2a.client import ClientConfig, ClientFactory
from a2a.types import AgentCard, TransportProtocol
from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
from google.adk.a2a.utils.agent_card_builder import AgentCardBuilder
# Agent Engine
from vertexai import agent_engines
from vertexai.preview.reasoning_engines import A2aAgent
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card
[PROJECT_ID] = !gcloud config list --format 'value(core.project)' 2>/dev/null
[PROJECT_NUMBER] = !gcloud projects describe {PROJECT_ID} --format='value(projectNumber)' 2>/dev/null
LOCATION = 'us-central1'
vertexai.init(project=PROJECT_ID, location=LOCATION)
os.environ['GOOGLE_CLOUD_PROJECT'] = PROJECT_ID
os.environ['GOOGLE_CLOUD_LOCATION'] = LOCATION
os.environ['GOOGLE_GENAI_USE_VERTEXAI'] = 'True'
STAGING_BUCKET = f'gs://{PROJECT_ID}'
!gsutil ls -b {STAGING_BUCKET} 2>/dev/null || \
gsutil mb -b on -l {LOCATION} {STAGING_BUCKET}
また、ノートブック上でエージェントを利用するための簡易アプリのクラスを定義します。
class ChatClient:
def __init__(self, adk_app, user_id='default_user'):
self.adk_app = adk_app
self.user_id = user_id
self.session_id = None
async def async_stream_query(self, message, show=True):
if not self.session_id:
session = await self.adk_app.async_create_session(
user_id=self.user_id,
)
self.session_id = getattr(session, 'id', None) or session['id']
result = []
async for event in self.adk_app.async_stream_query(
user_id=self.user_id,
session_id=self.session_id,
message=message,
):
author = event['author']
if ('content' in event and 'parts' in event['content']):
response = '\n'.join(
[p['text'] for p in event['content']['parts'] if 'text' in p]
)
if response:
if show:
display(Markdown(response))
result.append(response)
return '\n'.join(result)
リモートエージェントの定義と Agent Engine へのデプロイ
今回の構成では、次の 4 つのエージェントをリモートエージェントとして使用します。
-
research_agent1：調査レポートの調査項目を選定する -
research_agent2：選定した項目に基づいて調査レポートを作成する -
writer_agent：記事を作成する -
review_agent：記事をレビューする
これらのエージェントを定義した後に、Agent Engine にデプロイして、A2A リモートエージェントとして呼び出せるようにします。まずは、それぞれのエージェントを定義します。
research_agent1
instruction = '''
あなたの役割は、記事の執筆に必要な情報を収集して調査レポートにまとめる事です。
指定されたテーマの記事を執筆する際に参考となるトピックを５項目程度のリストにまとめます。
後段のエージェントがこのリストに基づいて、調査レポートを作成します。
* 出力形式
日本語で出力。
'''
research_agent1 = LlmAgent(
name='research_agent1',
model='gemini-2.5-flash',
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント（テーマ選定）',
instruction=instruction,
)
research_agent2
instruction = '''
あなたの役割は、記事の執筆に必要な情報を収集して調査レポートにまとめる事です。
前段のエージェントは、５項目程度の調査対象トピックを指定します。
* 出力形式
日本語で出力。
調査レポートは、トピックごとに客観的情報をまとめます。各トピックについて、５文以上の長さで記述すること。
'''
research_agent2 = LlmAgent(
name='research_agent2',
model='gemini-2.5-flash',
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント（レポート作成）',
instruction=instruction,
)
writer_agent
instruction = '''
あなたの役割は、特定のテーマに関する気軽な読み物記事を書くことです。
記事の「テーマ」と、その内容に関連する「調査レポート」が与えられるので、
調査レポートに記載の客観的事実に基づいて、信頼性のある読み物記事を書いてください。
**出力条件**
- トピックに関してある程度の基礎知識がある読者を前提として、数分で気軽に読める内容にしてください。
- 比較的カジュアルで語りかける口調の文章にします。
- 思考過程は出力せずに、最終結果だけを出力します。
- 記事タイトルは付けないで、次の構成で出力します。各セクションタイトルは、内容に合わせてアレンジしてください。
0. 導入：セクションタイトルを付けないで、この記事を読みたくなる導入を１〜２文にまとめます。
1. 概要：トピックの全体像をまとめて簡単に説明します。
2. 最新情報：特に注目したい新しい情報を取り上げます。
3. 実践：トピックに関して、読者自身がやってみるとよさそうな事を１つ紹介します。
4. まとめ
- 各セクションのタイトルはマークダウンヘッダ ## 付けます。必要に応じて箇条書きのマークダウンを使用します。
- それ以外のマークダウンによる装飾は行いません。
**レビュアーの指示に応じた修正**
- レビュアーが修正ポイントを提示した際は、出力条件にこだわらずに、直前の記事を指示に従って修正してください。
- 修正ポイント以外の部分は、修正する必要ありません。
'''
writer_agent = LlmAgent(
name='writer_agent',
model='gemini-2.5-flash',
description='特定のテーマに関する読み物記事を書くエージェント',
instruction=instruction,
)
review_agent
instruction = '''
あなたの役割は、読み物記事をレビューして、記事の条件にあった内容にするための改善コメントを与える事です。
* 記事の条件
- 記事は、はじめに４０文字程度のタイトルがあること。
今日から役立つ生活情報があって「すぐに読まなきゃ」と読者が感じるタイトルにすること。
タイトルはマークダウンヘッダ # をつけること。
- タイトルの直後に「なぜいまこのテーマを取り上げるのか」をまとめた導入を加えて、読者にこの記事を読む動機づけを与えます。
- 各セクションのサブタイトルには、絵文字を用いて親しみやすさを出すこと。
- 読者が今日から実践できる具体例が３つ以上紹介されていること。
* 出力形式
- 日本語で出力。
- はじめに、記事の良い点を説明します。
- 次に、修正ポイントを箇条書きで出力します。
'''
review_agent = LlmAgent(
name='review_agent',
model='gemini-2.5-flash',
description='読み物記事をレビューするエージェント',
instruction=instruction,
)
これらのエージェントを A2A リモートエージェントとして、Agent Engine にデプロイします。なお、Agent Engine にデプロイしたエージェントにはユニークなリソース ID が付与されますが、この後で説明するように、このリソース ID から A2A のエージェントカードを取得する URL が決まります。
まず、エージェント名から対応するリソース ID を取得する関数 get_agent_resource() を用意します。
client = vertexai.Client(
project=PROJECT_ID,
location=LOCATION,
http_options=HttpOptions(
api_version='v1beta1',
base_url=f'https://{LOCATION}-aiplatform.googleapis.com/'
),
)
def get_agent_resource(agent_name):
for agent in client.agent_engines.list():
if agent.api_resource.display_name == agent_name:
return agent.api_resource.name
return None
続いて、A2A サーバー機能に関連する関数を定義します。
class MyVertexAiSessionService(VertexAiSessionService):
def __init__(self, project_id: str, location: str):
super().__init__()
self.project_id = project_id
self.location = location
credentials, _ = google.auth.default(
scopes=['https://www.googleapis.com/auth/cloud-platform']
)
# AuthorizedSession handles automatic token refreshing and header injection
self.authed_session = AuthorizedSession(credentials)
def create_reasoning_engine_session(
self,
reasoning_engine_id: str,
user_id: str,
session_state: dict = {},
session_id: str = None,
):
"""
Creates a new Reasoning Engine Session via the Vertex AI REST API.
"""
api_endpoint = f'{self.location}-aiplatform.googleapis.com'
parent = f'projects/{self.project_id}/locations/{self.location}/reasoningEngines/{reasoning_engine_id}'
url = f'https://{api_endpoint}/v1beta1/{parent}/sessions'
params = {'sessionId': session_id} if session_id else {}
session_payload = {
'userId': user_id,
'sessionState': session_state,
}
print(f'Sending request to: {url}', flush=True)
response = self.authed_session.post(url, params=params, json=session_payload)
if response.status_code == 200:
print('Success! Operation initiated.', flush=True)
return response.json()
else:
print(f'Failed to create session. HTTP Status: {response.status_code}', flush=True)
print(f'Error Details: {response.text}', flush=True)
response.raise_for_status()
async def create_session(
self,
*,
app_name: str,
user_id: str,
state: dict = None,
session_id: str = None,
**kwargs,
):
try:
reasoning_engine_id = self._get_reasoning_engine_id(app_name)
# Wrap the synchronous requests call in a thread to avoid blocking the async event loop
operation_result = await asyncio.to_thread(
self.create_reasoning_engine_session,
reasoning_engine_id=reasoning_engine_id,
user_id=user_id,
session_state=state,
session_id=session_id,
)
print(f'Operation result: {operation_result}', flush=True)
created_session_id = operation_result['name'].split('/')[7]
session = None
max_retries = 3
for attempt in range(max_retries):
session = await self.get_session(
app_name=app_name,
user_id=user_id,
session_id=created_session_id,
)
if session is not None:
break
print(f'Session not ready yet. Retrying... ({attempt + 1}/{max_retries})', flush=True)
await asyncio.sleep(2**attempt)
if session is None:
print(f'Failed to fetch session for session_id={created_session_id} after {max_retries} attempts.', flush=True)
print(f'Succeeded to create a session with session_id={session.id}', flush=True)
return session
except Exception as e:
print(f"An error occurred during session creation: {e}", flush=True)
return None
def get_create_runner_class(agent):
async def create_runner():
project_id = os.environ.get('GOOGLE_CLOUD_PROJECT')
location = os.environ.get('GOOGLE_CLOUD_LOCATION')
agent_engine_id = os.environ.get('GOOGLE_CLOUD_AGENT_ENGINE_ID')
resource_name = None
if agent_engine_id:
resource_name = f'projects/{project_id}/locations/{location}/reasoningEngines/{agent_engine_id}'
session_service = MyVertexAiSessionService(
project_id=project_id,
location=location,
)
return Runner(
# Use resource_name as app_name for VertexAiSessionService.
app_name=resource_name or agent.name,
agent=agent,
artifact_service=InMemoryArtifactService(),
session_service=session_service,
memory_service=InMemoryMemoryService(),
credential_service=InMemoryCredentialService(),
)
return create_runner
def get_agent_executor_class(agent):
def agent_executor_builder():
return A2aAgentExecutor(
runner=get_create_runner_class(agent),
)
return agent_executor_builder
async def get_agent_card(agent):
builder = AgentCardBuilder(agent=agent)
adk_agent_card = await builder.build()
return create_agent_card(
agent_name=adk_agent_card.name,
description=adk_agent_card.description,
skills=adk_agent_card.skills
)
これで必要な準備ができました。次のコードで、先ほど定義した 4 つのエージェントを A2A リモートエージェントとして、Agent Engine にデプロイします。
agents = {
'research_agent1_a2a': research_agent1,
'research_agent2_a2a': research_agent2,
'writer_agent_a2a': writer_agent,
'review_agent_a2a': review_agent,
}
async def deploy_or_update_agent(agent_name, agent, client, staging_bucket):
a2a_agent = A2aAgent(
agent_card=await get_agent_card(agent),
agent_executor_builder=get_agent_executor_class(agent)
)
config = {
'display_name': agent_name,
'description': a2a_agent.agent_card.description,
'requirements': [
'google-adk==1.27.2',
'google-genai==1.68.0',
'google-cloud-aiplatform==1.141.0',
'a2a-sdk==0.3.25',
],
'staging_bucket': staging_bucket,
'gcs_dir_name': f'{agent_name}-{datetime.datetime.now().strftime("%Y%m%d-%H%M%S%f")}',
}
resource_name = get_agent_resource(agent_name)
if resource_name is None:
print(f'Deploying... {agent_name}: {agent.description}')
remote_a2a_agent = await asyncio.to_thread(
client.agent_engines.create,
agent=a2a_agent,
config=config,
)
else:
print(f'Updating... {agent_name}: {agent.description}')
remote_a2a_agent = await asyncio.to_thread(
client.agent_engines.update,
name=resource_name,
agent=a2a_agent,
config=config,
)
return remote_a2a_agent
async def deploy_all_agents():
tasks = [
deploy_or_update_agent(agent_name, agent, client, STAGING_BUCKET)
for agent_name, agent in agents.items()
]
results = await asyncio.gather(*tasks)
return results
await deploy_all_agents()
クラウドコンソールの上部にある検索バーから「Logs explorer」を検索して開くと、デプロイ中のログが確認できます。「すべてのフィールドを検索」と表示された部分に reasoning_engine と入力すると、Agent Engine に関連したログだけが表示されます。
RemoteA2aAgent を用いた root agent の定義
これで 4 つの A2A リモートエージェントが Agent Engine にデプロイできたので、これらをサブエージェントとして組み込んだ root agent（ワークフローを実行するエージェント）を定義します。
冒頭で説明したように、ADK の RemoteA2aAgent クラスを利用して、サブエージェントとして組み込みますが、この際に、A2A サーバーにアクセスする A2A クライアントのファクトリークラスが必要になります。次のコードで、ファクトリークラスを用意します。
class GoogleAuthRefresh(httpx.Auth):
def __init__(self, scopes):
self.credentials, _ = google.auth.default(scopes=scopes)
self.transport_request = Request()
self.credentials.refresh(self.transport_request)
def auth_flow(self, request):
if not self.credentials.valid:
self.credentials.refresh(self.transport_request)
request.headers['Authorization'] = f'Bearer {self.credentials.token}'
yield request
factory = ClientFactory(
ClientConfig(
supported_transports=[TransportProtocol.http_json],
use_client_preference=True,
httpx_client=httpx.AsyncClient(
timeout=60,
headers={'Content-Type': 'application/json'},
auth=GoogleAuthRefresh(scopes=['https://www.googleapis.com/auth/cloud-platform'])
),
),
)
そして、次のコードで、4 つの A2A リモートエージェントに対応した RemoteA2aAgent クラスのインスタンスを定義します。
resource_name = get_agent_resource('research_agent1_a2a')
a2a_url = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a"
research_agent1_remote = RemoteA2aAgent(
name="research_agent1",
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント（テーマ選定）',
agent_card=f"{a2a_url}/v1/card",
a2a_client_factory=factory,
)
resource_name = get_agent_resource('research_agent2_a2a')
a2a_url = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a"
research_agent2_remote = RemoteA2aAgent(
name="research_agent2",
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント（レポート作成）',
agent_card=f"{a2a_url}/v1/card",
a2a_client_factory=factory,
)
resource_name = get_agent_resource('writer_agent_a2a')
a2a_url = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a"
writer_agent_remote = RemoteA2aAgent(
name="writer_agent",
description='特定のテーマに関する読み物記事を書くエージェント',
agent_card=f"{a2a_url}/v1/card",
a2a_client_factory=factory,
)
resource_name = get_agent_resource('review_agent_a2a')
a2a_url = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a"
review_agent_remote = RemoteA2aAgent(
name="review_agent",
description='読み物記事をレビューするエージェント',
agent_card=f"{a2a_url}/v1/card",
a2a_client_factory=factory,
)
RemoteA2aAgent クラスのインスタンスは、agent_card オプションで指定した URL からエージェントカードを取得した後、エージェントカードから、エージェントがデプロイされた A2A サーバーのエンドポイントの情報を取得して、リモートエージェントにリクエストを送信します。上記のコードからわかるように、エージェントカードの URL は次で与えられます。
https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a/v1/card/
{LOCATION} は Agent Engine のリージョン（今回の例であれば us-central1）で、{resource_name} は該当エージェントのリソース ID になります。
この後は、RemoteA2aAgent クラスのインスタンスをサブエージェントに指定することで、ローカルのサブエージェントを使用する場合と同様に root agent が定義できます。
def get_print_agent(text):
def before_model_callback(
callback_context: CallbackContext, llm_request: LlmRequest
) -> LlmResponse:
return LlmResponse(
content=Content(
role='model', parts=[Part(text=text)],
)
)
return LlmAgent(
name='print_agent',
model='gemini-2.5-flash', # not used
description='',
instruction = '',
before_model_callback=before_model_callback,
)
research_agent = SequentialAgent(
name='research_agent',
sub_agents=[
get_print_agent('\n---\n## リサーチエージェントが調査レポートを作成します。\n---\n'),
get_print_agent('\n## 調査対象のトピックを選定します。\n'),
research_agent1_remote,
get_print_agent('\n## 選定したトピックに基づいて、調査レポートを作成します。\n'),
research_agent2_remote,
get_print_agent('\n#### 調査レポートが準備できました。記事の作成に取り掛かってもよいでしょうか？\n'),
],
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント',
)
write_and_review_agent = SequentialAgent(
name='write_and_review_agent',
sub_agents=[
get_print_agent('\n---\n## ライターエージェントが記事を執筆します。\n---\n'),
writer_agent_remote,
get_print_agent('\n---\n## レビューエージェントが記事をレビューします。\n---\n'),
review_agent_remote,
get_print_agent('\n#### レビューに基づいて記事の修正を依頼しますか？\n'),
],
description='記事を作成、レビューする。',
)
root_agent = LlmAgent(
name='article_generation_flow',
model='gemini-2.5-flash',
instruction = '''
何ができるか聞かれた場合は、以下の処理をすることをわかりやすくフレンドリーな文章にまとめて返答してください。
- ユーザーが指定したテーマの記事を作成する業務フローを実行する。
- はじめに、テーマに関する調査レポートを作成する。
- その後、ライターエージェントとレビューエージェントが協力して、編集方針に則した記事を作成する。
ユーザーが記事のテーマを指定した場合は、次のフローを実行します。
1. そのテーマの記事の作成に取り掛かる旨を伝えて、research_agent に転送して、調査レポートを依頼します。
2. ユーザー記事の作成を指示したら、write_and_review_agent に転送して、記事の作成とレビューを依頼します。
3. ユーザーが記事の修正を希望する場合は、write_and_review_agent に転送します。
**条件**
research_agent のニックネームは、リサーチエージェント
write_and_review_agent のネックネームは、ライターエージェントとレビューエージェント
''',
sub_agents=[
research_agent,
write_and_review_agent,
],
description='記事を作成する業務フローを実行するエージェント'
)
adk_app = AdkApp(
agent=root_agent,
app_name='article_generation_app',
)
実行例
それでは、この root agent を実際に使ってみましょう。ここでは、先に用意した簡易アプリで実行します。
chat_client = ChatClient(adk_app)
query = '''
こんにちは。何ができますか？
'''
response = await chat_client.async_stream_query(query)
[出力結果]
こんにちは！私は記事を作成する業務フローを実行するエージェントです。
具体的には、ユーザーさんが指定したテーマについて、まずリサーチエージェントが調査レポートを作成します。
その後、ライターエージェントとレビューエージェントが協力して、編集方針に沿った記事を作成します。
ご希望のテーマがあれば教えてください！
query = '''
「近場で秋を感じる工夫」をテーマに記事を作成してください。
'''
response = await chat_client.async_stream_query(query)
はい、「近場で秋を感じる工夫」というテーマの記事作成に取り掛かります！まずは、リサーチエージェントに依頼して、調査レポートを作成してもらいますね。
---
## リサーチエージェントが調査レポートを作成します。
---
...（以下省略）...
この後は、下記の記事の実行例と同じ流れで、記事の作成を進められます。
今回の実装内容をまとめたノートブックが下記にあるので、こちらも参考にしてください。
root agent を Agent Engine にデプロイ
先ほどは root agent 自体はローカルで実行しましたが、root agent も Agent Engine にデプロイして使用することができます。ただし、本記事執筆時点では、先に説明した、A2A クライアントのファクトリークラスと RemoteA2aAgent クラスに少し修正が必要になります。
ここでは、詳細な説明は割愛して、修正後のコードを紹介します。
まず、次のコードで root agent を再定義します。
class MyClientFactory(ClientFactory):
def create(self, card, consumers=None, interceptors=None):
if not self._config.httpx_client:
self._config.httpx_client=httpx.AsyncClient(
timeout=60,
headers={'Content-Type': 'application/json'},
auth=GoogleAuthRefresh(scopes=['https://www.googleapis.com/auth/cloud-platform'])
)
self._register_defaults(self._config.supported_transports)
return super().create(card, consumers, interceptors)
class MyRemoteA2aAgent(RemoteA2aAgent):
async def _ensure_httpx_client(self):
if not self._httpx_client:
self._httpx_client=httpx.AsyncClient(
timeout=60,
headers={'Content-Type': 'application/json'},
auth=GoogleAuthRefresh(scopes=['https://www.googleapis.com/auth/cloud-platform'])
)
return self._httpx_client
factory = MyClientFactory(
ClientConfig(
supported_transports=[TransportProtocol.http_json],
use_client_preference=True,
)
)
resource_name = get_agent_resource('research_agent1_a2a')
a2a_url = f'https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a'
research_agent1_remoteA2a = MyRemoteA2aAgent(
name='research_agent1',
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント（テーマ選定）',
agent_card=f'{a2a_url}/v1/card',
a2a_client_factory=factory,
)
resource_name = get_agent_resource('research_agent2_a2a')
a2a_url = f'https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a'
research_agent2_remoteA2a = MyRemoteA2aAgent(
name='research_agent2',
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント（レポート作成）',
agent_card=f'{a2a_url}/v1/card',
a2a_client_factory=factory,
)
resource_name = get_agent_resource('writer_agent_a2a')
a2a_url = f'https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a'
writer_agent_remoteA2a = MyRemoteA2aAgent(
name='writer_agent',
description='特定のテーマに関する読み物記事を書くエージェント',
agent_card=f'{a2a_url}/v1/card',
a2a_client_factory=factory,
)
resource_name = get_agent_resource('review_agent_a2a')
a2a_url = f'https://{LOCATION}-aiplatform.googleapis.com/v1beta1/{resource_name}/a2a'
review_agent_remoteA2a = MyRemoteA2aAgent(
name='review_agent',
description='読み物記事をレビューするエージェント',
agent_card=f'{a2a_url}/v1/card',
a2a_client_factory=factory,
)
def get_print_agent(text):
def before_model_callback(
callback_context: CallbackContext, llm_request: LlmRequest
) -> LlmResponse:
return LlmResponse(
content=Content(
role='model', parts=[Part(text=text)],
)
)
return LlmAgent(
name='print_agent',
model='gemini-2.5-flash', # not used
description='',
instruction = '',
before_model_callback=before_model_callback,
)
research_agent = SequentialAgent(
name='research_agent',
sub_agents=[
get_print_agent('\n---\n## リサーチエージェントが調査レポートを作成します。\n---\n'),
get_print_agent('\n## 調査対象のトピックを選定します。\n'),
research_agent1_remoteA2a,
get_print_agent('\n## 選定したトピックに基づいて、調査レポートを作成します。\n'),
research_agent2_remoteA2a,
get_print_agent('\n#### 調査レポートが準備できました。記事の作成に取り掛かってもよいでしょうか？\n'),
],
description='記事の執筆に必要な情報を収集してレポートにまとめるエージェント',
)
write_and_review_agent = SequentialAgent(
name='write_and_review_agent',
sub_agents=[
get_print_agent('\n---\n## ライターエージェントが記事を執筆します。\n---\n'),
writer_agent_remoteA2a,
get_print_agent('\n---\n## レビューエージェントが記事をレビューします。\n---\n'),
review_agent_remoteA2a,
get_print_agent('\n#### レビューに基づいて記事の修正を依頼しますか？\n'),
],
description='記事を作成、レビューする。',
)
root_agent = LlmAgent(
name='article_generation_flow',
model='gemini-2.5-flash',
instruction = '''
何ができるか聞かれた場合は、以下の処理をすることをわかりやすくフレンドリーな文章にまとめて返答してください。
- ユーザーが指定したテーマの記事を作成する業務フローを実行する。
- はじめに、テーマに関する調査レポートを作成する。
- その後、ライターエージェントとレビューエージェントが協力して、編集方針に則した記事を作成する。
ユーザーが記事のテーマを指定した場合は、次のフローを実行します。
1. そのテーマの記事の作成に取り掛かる旨を伝えて、research_agent に転送して、調査レポートを依頼します。
2. ユーザー記事の作成を指示したら、write_and_review_agent に転送して、記事の作成とレビューを依頼します。
3. ユーザーが記事の修正を希望する場合は、write_and_review_agent に転送します。
**条件**
research_agent のニックネームは、リサーチエージェント
write_and_review_agent のネックネームは、ライターエージェントとレビューエージェント
''',
sub_agents=[
research_agent,
write_and_review_agent,
],
description='記事を作成する業務フローを実行するエージェント'
)
adk_app = AdkApp(
agent=root_agent,
app_name='article_generation_app',
)
そして、次のコードで、root agent を Agent Engine にデプロイします。
agent_name = adk_app._tmpl_attrs['agent'].name
description = adk_app._tmpl_attrs['agent'].description
config = {
'agent_framework': 'google-adk',
'display_name': agent_name,
'description': description,
'requirements': [
'google-adk==1.27.2',
'google-genai==1.68.0',
'google-cloud-aiplatform==1.141.0',
'a2a-sdk==0.3.25'
],
'staging_bucket': f'gs://{PROJECT_ID}',
'gcs_dir_name': f'{agent_name}-{datetime.datetime.now().strftime("%Y%m%d-%H%M%S%f")}',
}
resource_name = get_agent_resource(agent_name)
if resource_name is None:
print(f'Deploying... {agent_name}: {description}')
remote_app = client.agent_engines.create(
agent=adk_app,
config=config,
)
else:
print(f'Updating... {agent_name}: {description}')
remote_app = client.agent_engines.update(
name=resource_name,
agent=adk_app,
config=config,
)
この後は、次のコードで root agent を実行します。
chat_client = ChatClient(remote_app)
query = '''
こんにちは。何ができますか？
'''
response = await chat_client.async_stream_query(query)
[出力結果]
こんにちは！私は、記事を作成する業務フローを実行するエージェントです。
ユーザーさんが指定したテーマの記事を作成することができます。
具体的には、まずテーマについてのリサーチエージェントが調査レポートを作成します。
その後、ライターエージェントとレビューエージェントが協力して、編集方針に沿った記事を作成します。
どんなテーマの記事をご希望ですか？
この後の実行の流れは、ローカルの場合と同様です。
まとめ
この記事では、Agent Engine に複数の A2A リモートエージェントをデプロイした上で、これらを連携したワークフローを実行する例を紹介しました。ここでは特に、A2A サーバーと A2A クライアントの双方で、ADK が標準提供するクラスを利用した点に注目してください。
- サーバー側：A2A サーバーの機能を提供する
A2aAgentExecutorクラス - クライアント側：A2A リモートエージェントをサブエージェントとして組み込む
RemoteA2aAgentクラス
下記の記事では独自実装の A2A サーバーを Cloud Run にデプロイした上で、独自実装の A2A クライアント機能をコールバック関数に組み込むという工夫をしました。
これに比べると、今回の実装例では比較的シンプルなコードになっていることがわかります。
このように、ADK の A2A 対応機能と Agent Engine の A2A 対応が連携することで、より簡単に A2A が利用できます。 詳細な説明は割愛しますが、上記の 2 つのクラスが連携する事で、root agent のセッション情報と A2A リモートエージェントのセッション情報が適切に同期されて、それぞれの A2A リモートエージェントは、ワークフロー全体の処理の流れを理解した上で、自身が担当する処理を実行していきます。
本文冒頭の注意書きにあるように、本記事執筆時点では、ADK の A2A 対応機能は「Experimental」フェーズのため、Wrapper クラスを利用するなど、多少のワークアラウンドが必要でしたが、このあたりは今後のバージョンアップで解消されていくでしょう。ADK と Agent Engine の進化には今後も要注目ですね。
Discussion