Agent2Agent 入門 (12) - ADK Agent

npaka

2025年7月6日 15:59

「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__.py

INFO:     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では朝のニュースをダイジェストで放送しています。 また、ウェザーニュースではその日の暦の情報などを提供しています。