Agent2Agent 入門 (12) - ADK Agent
「Agent2Agent」の公式サンプル「ADK Agent」を試したのでまとめました。
前回
1. ADK Agent
Google Agent Development Kit (ADK) を使用して、A2A を使用して通信するシンプルなファクトジェネレータを作成します。
2. セットアップ
セットアップ手順は、次のとおりです。
(1) プロジェクトフォルダの作成
mkdir adk_facts
cd adk_facts(2) Pythonの仮想環境の準備。
Python 3.11以上を使用します。
・Mac
python -m venv .venv
source .venv/bin/activate・Windows
python -m venv .venv
.venv\Scripts\activate(3) パッケージのインストール。
pip install a2a-sdk uvicorn
pip install google-adk(4) Gemini APIキーを環境変数に設定。
export GOOGLE_API_KEY=<Gemini APIキー>3. A2Aエージェントの実装
3-1. コアエージェントの定義
Google検索ツールを持つADKエージェントを準備します。
・agent.py
from google.adk.agents import Agent
from google.adk.tools import google_search
# コアエージェントの準備
root_agent = Agent(
name="facts_agent",
model="gemini-2.5-flash-lite-preview-06-17",
description=("興味深い事実を伝えるエージェント。"),
instruction=("あなたは興味深い事実を提供できる役に立つエージェントです。"),
tools=[google_search],
)Agentのパラメータは、次のとおりです。
・name : 名前
・model : モデルID
・description : エージェントのメタデータ的な短い説明文
・instruction : LLMのシステムメセージ
・tool : ツール
3-2. AgentExecutor でコアエージェントをラップ
・agent_executor.py
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import (
Part,
TaskState,
TextPart,
)
from a2a.utils import new_agent_text_message, new_task
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
# AgentExecutorの作成
class ADKAgentExecutor(AgentExecutor):
# 初期化
def __init__(
self,
agent,
status_message="リクエストを処理中…\n",
artifact_name="response",
):
self.agent = agent
self.status_message = status_message
self.artifact_name = artifact_name
self.runner = Runner(
app_name=agent.name,
agent=agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
# 実行
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
# ユーザー入力を取得
query = context.get_user_input()
# 既存タスクがなければ新規作成
task = context.current_task or new_task(context.message)
await event_queue.enqueue_event(task)
# TaskUpdaterの準備
updater = TaskUpdater(event_queue, task.id, task.contextId)
if context.call_context:
user_id = context.call_context.user.user_name
else:
user_id = "a2a_user"
try:
# カスタムメッセージでステータスを更新
await updater.update_status(
TaskState.working,
new_agent_text_message(self.status_message, task.contextId, task.id),
)
# セッションを作成し、ユーザーからの問い合わせをContent型にラップ
session = await self.runner.session_service.create_session(
app_name=self.agent.name,
user_id=user_id,
state={},
session_id=task.contextId,
)
content = types.Content(
role="user", parts=[types.Part.from_text(text=query)]
)
# runnerを非同期で実行し、生成されたイベントを逐次処理
response_text = ""
async for event in self.runner.run_async(
user_id=user_id, session_id=session.id, new_message=content
):
# 最終レスポンスかつコンテンツがある場合
if event.is_final_response() and event.content and event.content.parts:
for part in event.content.parts:
# text
if hasattr(part, "text") and part.text:
response_text += part.text + "\n"
# function_call
elif hasattr(part, "function_call"):
pass
# 完了後、テキストレスポンスをアーティファクトとして追加
await updater.add_artifact(
[Part(root=TextPart(text=response_text))],
name=self.artifact_name,
)
# タスクを完了状態に更新
await updater.complete()
except Exception as e:
await updater.update_status(
TaskState.failed,
new_agent_text_message(f"エラー: {e!s}", task.contextId, task.id),
final=True,
)
# キャンセル
async def cancel(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
raise NotImplementedError("キャンセルは実装されていません。")
3-3. AgentSkill と AgentCard の準備
・__main__.py
import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
AgentCapabilities,
AgentCard,
AgentSkill,
)
from agent import root_agent as facts_agent
from agent_executor import ADKAgentExecutor
# メイン
def main():
# Agent Skill の準備
skill = AgentSkill(
id="give_facts",
name="Provide Interesting Facts",
description="Googleで興味深い事実を検索",
tags=["search", "google", "facts"],
examples=[
"ニューヨーク市についての興味深い事実を教えてください。",
],
)
# Agent Card の準備
agent_card = AgentCard(
name=facts_agent.name,
description=facts_agent.description,
url='http://localhost:10001/',
version="1.0.0",
defaultInputModes=["text", "text/plain"],
defaultOutputModes=["text", "text/plain"],
capabilities=AgentCapabilities(streaming=True),
skills=[skill],
)
# リクエストハンドラの準備
request_handler = DefaultRequestHandler(
agent_executor=ADKAgentExecutor(
agent=facts_agent,
),
task_store=InMemoryTaskStore(),
)
# サーバの準備
server = A2AStarletteApplication(
agent_card=agent_card,
http_handler=request_handler
)
uvicorn.run(server.build(), host="localhost", port=10001)
# メインの実行
if __name__ == "__main__":
main()
3-4. A2Aエージェントの実行
(1) A2Aエージェントの実行。
A2Aエンドポイントが公開されます。
python __main__.pyINFO: Started server process [77075]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:10002 (Press CTRL+C to quit)4. A2Aエージェントの呼び出し
(1) A2Aエージェントを呼び出すクライアントの作成。
・loop_client.py
import asyncio
from typing import Any
from uuid import uuid4
import httpx
from a2a.client import A2AClient
from a2a.types import (
MessageSendParams,
SendStreamingMessageRequest,
)
# ウェルカムメッセージの出力
def print_welcome_message() -> None:
print('汎用 A2A クライアントへようこそ!')
print("クエリを入力してください (終了するには「exit」と入力)")
# ユーザークエリの取得
def get_user_query() -> str:
return input('\n> ')
# サーバとのやりとり
async def interact_with_server(client: A2AClient) -> None:
while True:
user_input = get_user_query()
if user_input.lower() == 'exit':
break
send_message_payload: dict[str, Any] = {
'message': {
'role': 'user',
'parts': [{'type': 'text', 'text': user_input}],
'messageId': uuid4().hex,
},
}
try:
streaming_request = SendStreamingMessageRequest(
id=uuid4().hex,
params=MessageSendParams(**send_message_payload)
)
stream_response = client.send_message_streaming(streaming_request)
async for chunk in stream_response:
print(get_response_text(chunk), end='', flush=True)
await asyncio.sleep(0.1)
except Exception as e:
print(f'エラーが発生しました: {e}')
# レスポンステキストの取得
def get_response_text(chunk):
data = chunk.model_dump(mode='json', exclude_none=True)
result = data.get('result', {})
# アーティファクト更新イベント
if 'artifact' in result:
parts = result['artifact']['parts']
# ステータス更新イベント
elif 'status' in result and result['status'].get('message'):
parts = result['status']['message']['parts']
# その他のイベント
else:
return ''
# 最初のパートのテキストを返す
return parts[0].get('text', '')
# メイン
async def main() -> None:
print_welcome_message()
async with httpx.AsyncClient() as httpx_client:
client = await A2AClient.get_client_from_agent_card_url(
httpx_client, 'http://localhost:10001'
)
await interact_with_server(client)
# メインの実行
if __name__ == '__main__':
asyncio.run(main())
(2) A2Aエージェントを呼び出すクライアントの実行。
python loop_client.py 汎用 A2A クライアントへようこそ!
クエリを入力してください (終了するには「exit」と入力)
> (3) 質問。
Google検索で最新の事実情報を教えてくれることを確認します。
> 2025年7月6日の日本のニュースを1つ教えて？リクエストを処理中…
2025年7月6日（日）の日本のニュースとしては、以下の情報があります。
* **バスケットボール男子日本代表、オランダ戦に臨む**
バスケットボール男子日本代表チームは、この日有明アリーナにて「日本生命カップ2025（東京大会）」のオランダ戦に臨みました。試合登録メンバー（ロスター）も発表されています。
* **参議院選挙、与党の過半数獲得は微妙な情勢**
朝日新聞社の情勢調査によると、参議院選挙において、与党は目標とする過半数獲得が微妙な状況です。自民党は改選前の議席を下回る見通しで、公明党も目標議席に届かない情勢です。一方、立憲民主党や国民民主党は議席を伸ばしそうです。
* **鹿児島県十島村で最大震度5強の地震発生**
6日午後2時1分ごろ、鹿児島県の十島村で最大震度5強を観測する強い地震がありました。震源地はトカラ列島近海で、津波の心配はありませんでした。
この他にも、TBS NEWS DIGでは朝のニュースをダイジェストで放送しています。 また、ウェザーニュースではその日の暦の情報などを提供しています。