🐡

A2AプロトコルでCloud Run上のエージェントを連携させる

2025/08/27に公開

Google Cloud

A2A

Agent Development Kit

tech

Agent2Agent (A2A) protocol により、エージェント同士が相互に通信できます。

今回はA2Aのチュートリアルを参考に、2つのエージェントをそれぞれCloud Runにデプロイし、A2Aプロトコルで連携させます。

事前準備

まず、uvを使って環境構築します。

uv init
uv venv
source .venv/bin/activate
uv add google-adk # 1.10

今回実装するのは、サイコロを振り、出た目が素数かを判定するマルチエージェントシステムです。（チュートリアルのエージェントとほぼ同じですが一部簡略化しています）

dice_agent: 「スタート」を合図にサイコロを振り、出た目を出力します。
check_prime_agent: 数字が素数かどうかを判定します。

最終的にそれぞれCloud Runにデプロイし、A2Aで接続してみます。

check_prime_agentの実装

Agentをローカルで起動する

まず、check_prime_agentから実装します。

check_prime_agent/__init__.py

from . import agent

check_prime_agent/agent.py

from google.adk import Agent


async def check_prime(nums: list[int]) -> str:
    """与えられた数字が素数かどうか判定します。
    Args:
        nums: 素数かどうか判定する数字のリスト
    Returns:
        素数のリスト
    """
    primes = set()
    for number in nums:
        number = int(number)
        if number <= 1:
            continue
        is_prime = True
        for i in range(2, int(number**0.5) + 1):
            if number % i == 0:
                is_prime = False
                break
        if is_prime:
            primes.add(number)
    return (
        "素数はありませんでした"
        if not primes
        else f"{', '.join(str(num) for num in primes)} は素数です。"
    )


root_agent = Agent(
    model="gemini-2.0-flash",
    name="check_prime_agent",
    description="素数チェックAgent",
    instruction="与えられた数字が素数かどうか判定してください。",
    tools=[check_prime],
)

.env

GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY=AIzaxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

GOOGLE_API_KEYはこちらから入手してください。
adk webで起動し、質問してみます。

A2Aエージェントとして実装する

次に、このcheck_prime_agentをA2AエージェントとしてCloud Runにデプロイするために、AgentExecutorを実装します。（公式のhelloworldサンプルをベースに実装しています）

uv add a2a-sdk

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


class ADKAgentExecutor(AgentExecutor):
    def __init__(
        self,
        agent,
        status_message="Processing request...",
        artifact_name="response",
    ):
        """Initialize a generic ADK agent executor.

        Args:
            agent: The ADK agent instance
            status_message: Message to display while processing
            artifact_name: Name for the response artifact
        """
        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 cancel(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        """Cancel the execution of a specific task."""
        raise NotImplementedError(
            "Cancellation is not implemented for ADKAgentExecutor."
        )

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        if not context.message:
            raise ValueError("Message should be present in request context")

        query = context.get_user_input()
        task = context.current_task or new_task(context.message)
        await event_queue.enqueue_event(task)

        updater = TaskUpdater(event_queue, task.id, task.context_id)
        if context.call_context:
            user_id = context.call_context.user.user_name
        else:
            user_id = "a2a_user"

        try:
            # Update status with custom message
            await updater.update_status(
                TaskState.working,
                new_agent_text_message(self.status_message, task.context_id, task.id),
            )

            # Process with ADK agent
            session = await self.runner.session_service.create_session(
                app_name=self.agent.name,
                user_id=user_id,
                state={},
                session_id=task.context_id,
            )

            content = types.Content(
                role="user", parts=[types.Part.from_text(text=query)]
            )

            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:
                        if hasattr(part, "text") and part.text:
                            response_text += part.text + "\n"
                        elif hasattr(part, "function_call"):
                            # Log or handle function calls if needed
                            pass  # Function calls are handled internally by ADK

            # Add response as artifact with custom name
            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"Error: {e!s}", task.context_id, task.id),
                final=True,
            )

__main__.py

import asyncio
import functools
import logging
import os

import click
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 dotenv import load_dotenv
from starlette.applications import Starlette
from check_prime_agent.agent import root_agent
from agent_executor import ADKAgentExecutor


load_dotenv()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def make_sync(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return asyncio.run(func(*args, **kwargs))

    return wrapper


@click.command()
@click.option("--host", default="localhost")
@click.option("--port", default=8080)
@make_sync
async def main(host, port):
    agent_card = AgentCard(
        name=root_agent.name,
        description=root_agent.description,
        version="1.0.0",
        url=os.environ["APP_URL"],
        default_input_modes=["text", "text/plain"],
        default_output_modes=["text", "text/plain"],
        capabilities=AgentCapabilities(streaming=True),
        skills=[
            AgentSkill(
                id="check_prime",
                name="素数判定",
                description="数値のリストを素数かどうか判定します",
                tags=["mathematical", "computation", "prime", "numbers"],
            )
        ],
    )

    task_store = InMemoryTaskStore()

    request_handler = DefaultRequestHandler(
        agent_executor=ADKAgentExecutor(
            agent=root_agent,
        ),
        task_store=task_store,
    )

    a2a_app = A2AStarletteApplication(
        agent_card=agent_card, http_handler=request_handler
    )
    routes = a2a_app.routes()
    app = Starlette(
        routes=routes,
        middleware=[],
    )

    config = uvicorn.Config(app, host=host, port=port, log_level="info")
    server = uvicorn.Server(config)
    await server.serve()


if __name__ == "__main__":
    main()

途中に出てきたAgentCardとは、エージェントの能力を説明したものです。今回だと「素数判定」を行うエージェントであり、"mathematical", "computation", "prime", "numbers"というタグがついていることがわかります。そして、A2AにおけるAgentCardを公開するサーバーをuvicornで立ち上げています。

A2Aでは、AgentCardを提供するエンドポイントとエージェント本体のエンドポイントを同じURLで公開します。今回の実装では、両方とも同じCloud Runインスタンスでホストされます。

Dockerfile

FROM python:3.13-slim

COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

EXPOSE 8080
WORKDIR /app

COPY . ./

RUN uv sync

# 今回、portは8080にしています
ENTRYPOINT ["uv", "run", ".", "--host", "0.0.0.0", "--port", "8080"]

ここで、.envファイルに環境変数APP_URLを追加してください。
これはAgentCardに設定する公開URLです。Cloud Runの場合、デプロイ時の公開URLは以下のようにあらかじめ決められています。

https://<name>-<project_number>.<region>.run.app

<name>: サービス名です。デプロイ時に任意の文字列を指定できます。
<region>: デプロイ先のリージョンです。今回はus-central1を指定します。(注意：利用するモデルがサポートしているリージョンを選択する必要があります)
<project_number>: Google Cloudのプロジェクトナンバーです。gcloud projects listで確認できます。

例: APP_URL=https://a2a-check-prime-agent-50000000000.us-central1.run.app
（このproject_numberは仮の値です）

ローカルで検証する

このサーバーが適切に動作することをローカルで確認してみます。

$ uv run .
INFO:     Started server process [23454]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8080 (Press CTRL+C to quit)

サーバーが立ち上がったら、新しくターミナルを開いてAgentCardを取得するリクエストを送ってみます。AgentCardは/.well-known/agent-card.jsonで公開されています。(以下の出力は見やすくするために改行しています)

$ curl -X GET http://localhost:8080/.well-known/agent-card.json
{
    "capabilities":{"streaming":true},
    "defaultInputModes":["text","text/plain"],
    "defaultOutputModes":["text","text/plain"],
    "description":"素数チェックAgent",
    "name":"check_prime_agent",
    "preferredTransport":"JSONRPC",
    "protocolVersion":"0.3.0",
    "skills":[{
        "description":"数値のリストを素数かどうか判定します",
        "id":"check_prime",
        "name":"素数判定",
        "tags":["mathematical","computation","prime","numbers"]
    }],
    "url":"https://a2a-check-prime-agent-50000000000.us-central1.run.app",
    "version":"1.0.0"
}

※この時点では、AgentCardのURLはまだデプロイ前のURLを指しているため、実際にはアクセスできません。

Cloud Runへデプロイする

それでは、A2AエージェントをCloud Runにデプロイします。

GOOGLE_CLOUD_PROJECT=your-google-cloud-project
REGION=us-central1
REPOSITORY=a2a-sample
IMAGE=a2a-check-prime-agent

IMAGE_PATH=${REGION}-docker.pkg.dev/${GOOGLE_CLOUD_PROJECT}/${REPOSITORY}/${IMAGE}:latest

# DockerイメージをアップロードするArtifact Registryを作成します
gcloud artifacts repositories create a2a-sample \
      --repository-format=docker \
      --location=${REGION} \
      --project=${GOOGLE_CLOUD_PROJECT} \
      --description="A2A sample agent repository"

# ビルド
docker build --platform linux/amd64 -t ${IMAGE_PATH} .

# プッシュ
docker push ${IMAGE_PATH}

# 環境変数を指定してデプロイ
ENV_VARS=$(grep -v '^#' .env | grep -v '^$' | xargs | sed 's/ /,/g')
gcloud run deploy a2a-check-prime-agent \
    --image ${IMAGE_PATH} \
    --port=8080 \
    --allow-unauthenticated \
    --region=${REGION} \
    --project=${GOOGLE_CLOUD_PROJECT} \
    --memory=1Gi \
    --set-env-vars=${ENV_VARS}

今回は環境変数に直接APIキーを設定していますが、本番環境ではSecret Managerなどを利用してより安全な方法で管理してください。

Cloud RunにデプロイしたA2Aエージェントを検証する

デプロイが成功したら、Cloud Runの公開URLが表示されますので、リクエストを送って無事AgentCardを取得できることを確認します。

$ curl -X GET https://a2a-check-prime-agent-50000000000.us-central1.run.app/.well-known/agent-card.json
{
    "capabilities":{"streaming":true},
    "defaultInputModes":["text","text/plain"],
    "defaultOutputModes":["text","text/plain"],
    "description":"素数チェックAgent",
    "name":"check_prime_agent",
    "preferredTransport":"JSONRPC",
    "protocolVersion":"0.3.0",
    "skills":[{
        "description":"数値のリストを素数かどうか判定します",
        "id":"check_prime",
        "name":"素数判定",
        "tags":["mathematical","computation","prime","numbers"]
    }],
    "url":"https://a2a-check-prime-agent-50000000000.us-central1.run.app",
    "version":"1.0.0"
}

次に、A2AエージェントにPythonでリクエストを送って確認してみます。
この実装は先ほど同様helloworldとほぼ同じ実装ですが多少簡略化しています。

test_client.py

from typing import Any
from uuid import uuid4
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
)

import os
import dotenv

dotenv.load_dotenv()


base_url = os.environ["APP_URL"]


async def main() -> None:
    async with httpx.AsyncClient() as httpx_client:
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=base_url,
        )
        final_agent_card_to_use: AgentCard = await resolver.get_agent_card()

        client = A2AClient(
            httpx_client=httpx_client, agent_card=final_agent_card_to_use
        )

        send_message_payload: dict[str, Any] = {
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": "7は素数ですか？"}],
                "messageId": uuid4().hex,
            },
        }
        request = SendMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )

        response = await client.send_message(request)
        print("--- send_message ---")
        print(response.model_dump(mode="json", exclude_none=True))

        streaming_request = SendStreamingMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )
        stream_response = client.send_message_streaming(streaming_request)
        print("--- send_message_streaming ---")
        async for chunk in stream_response:
            print(chunk.model_dump(mode="json", exclude_none=True))


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

それでは実行してみます。

$ python test_client.py
--- send_message ---
{'id': '93a85af2-6127-4090-9121-5a2fe45c393d', 'jsonrpc': '2.0', 'result': {'artifacts': [{'artifactId': '8b4bd453-a598-4701-a0ad-26d55ddaccbd', 'name': 'response', 'parts': [{'kind': 'text', 'text': 'はい、7は素数です。\n'}]}], 'contextId': 'a4a0ac98-8418-4d2b-8948-4ad09ad82e46', 'history': [{'contextId': 'a4a0ac98-8418-4d2b-8948-4ad09ad82e46', 'kind': 'message', 'messageId': '68d7d617de854739ab1d70359be8d698', 'parts': [{'kind': 'text', 'text': '7は素数ですか？'}], 'role': 'user', 'taskId': '794ac6c8-df9e-42b8-a444-04da08da9c78'}, {'contextId': 'a4a0ac98-8418-4d2b-8948-4ad09ad82e46', 'kind': 'message', 'messageId': '1b0c5fb7-0ad2-49d8-81e3-f774140af757', 'parts': [{'kind': 'text', 'text': 'Processing request...'}], 'role': 'agent', 'taskId': '794ac6c8-df9e-42b8-a444-04da08da9c78'}], 'id': '794ac6c8-df9e-42b8-a444-04da08da9c78', 'kind': 'task', 'status': {'state': 'completed', 'timestamp': '2025-08-26T09:45:20.616038+00:00'}}}
--- send_message_streaming ---
{'id': 'e2c713a9-b0a3-4773-bafb-2f1cbf1135eb', 'jsonrpc': '2.0', 'result': {'contextId': '088e8ee2-64c6-494d-8b46-f10bd1a61cb0', 'history': [{'contextId': '088e8ee2-64c6-494d-8b46-f10bd1a61cb0', 'kind': 'message', 'messageId': '68d7d617de854739ab1d70359be8d698', 'parts': [{'kind': 'text', 'text': '7は素数ですか？'}], 'role': 'user', 'taskId': 'd0ee7a51-42ee-4f60-9edc-e0befd01476f'}], 'id': 'd0ee7a51-42ee-4f60-9edc-e0befd01476f', 'kind': 'task', 'status': {'state': 'submitted'}}}
{'id': 'e2c713a9-b0a3-4773-bafb-2f1cbf1135eb', 'jsonrpc': '2.0', 'result': {'contextId': '088e8ee2-64c6-494d-8b46-f10bd1a61cb0', 'final': False, 'kind': 'status-update', 'status': {'message': {'contextId': '088e8ee2-64c6-494d-8b46-f10bd1a61cb0', 'kind': 'message', 'messageId': 'e1fa2bfe-f82a-4cc7-8687-220457009355', 'parts': [{'kind': 'text', 'text': 'Processing request...'}], 'role': 'agent', 'taskId': 'd0ee7a51-42ee-4f60-9edc-e0befd01476f'}, 'state': 'working', 'timestamp': '2025-08-26T09:45:20.784097+00:00'}, 'taskId': 'd0ee7a51-42ee-4f60-9edc-e0befd01476f'}}
{'id': 'e2c713a9-b0a3-4773-bafb-2f1cbf1135eb', 'jsonrpc': '2.0', 'result': {'artifact': {'artifactId': 'b36069e2-a29a-4608-9762-4686f268f418', 'name': 'response', 'parts': [{'kind': 'text', 'text': 'はい、7は素数です。\n'}]}, 'contextId': '088e8ee2-64c6-494d-8b46-f10bd1a61cb0', 'kind': 'artifact-update', 'taskId': 'd0ee7a51-42ee-4f60-9edc-e0befd01476f'}}
{'id': 'e2c713a9-b0a3-4773-bafb-2f1cbf1135eb', 'jsonrpc': '2.0', 'result': {'contextId': '088e8ee2-64c6-494d-8b46-f10bd1a61cb0', 'final': True, 'kind': 'status-update', 'status': {'state': 'completed', 'timestamp': '2025-08-26T09:45:21.760058+00:00'}, 'taskId': 'd0ee7a51-42ee-4f60-9edc-e0befd01476f'}}

「7は素数ですか？」というこちらの入力に対し、「はい、7は素数です。」と返していることがわかります。A2Aプロトコルでは、相手のエージェントが内部でどのような動作をしているのかを公開する必要がありません。上の例を見ても、Cloud Runにデプロイしたcheck_prime_agentの内部処理の詳細は隠蔽されています。

ここまででデプロイしたエージェントに対して、A2Aプロトコルでローカルから接続できることがわかりました。

dice_agentの実装

A2Aエージェントを呼び出すエージェントの実装

次は、もう一つのエージェントdice_agentからA2Aプロトコルを使って、今デプロイしたcheck_prime_agentに接続してみます。

dice_agent/__init__.py

from . import agent

dice_agent/agent.py

import random
from google.adk import Agent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent


APP_URL = "https://a2a-check-prime-agent-50000000000.us-central1.run.app"


def roll_die() -> int:
    """6面サイコロを振ります"""
    return random.randint(1, 6)


remote_check_prime_agent = RemoteA2aAgent(
    name="check_prime_agent",
    description="素数判定エージェント",
    agent_card=(f"{APP_URL}/.well-known/agent-card.json"),
)

root_agent = Agent(
    model="gemini-2.0-flash",
    name="dice_agent",
    description="サイコロAgent",
    instruction="ユーザーが「スタート」と言ったら、6面サイコロを振ります。出た目が素数かどうかをエージェントに判定してもらい、結果をユーザーに通知します。",
    tools=[roll_die],
    sub_agents=[remote_check_prime_agent],
)

先ほどCloud Runにデプロイしたエージェントcheck_prime_agentをRemoteA2aAgentとして定義し、root_agentからsub_agentsで指定しています。

ローカルで検証する

adk webで立ち上げてdice_agentを選択してみます。

無事適切に動くことが確認できました。

Cloud Runにデプロイして接続する

ADKでは、Cloud Runへ1コマンドでデプロイする機能があります。
(参考: Agent Development Kit | Deploy | Cloud Run)

uv pip freeze > dice_agent/requirements.txt

adk deploy cloud_run \
    --project=$GOOGLE_CLOUD_PROJECT \
    --region=$REGION \
    --service_name=a2a-dice-agent \
    --with_ui \
    ./dice_agent

デプロイされたら公開URLにアクセスして検証してみます。

無事Cloud Runにデプロイした2体のエージェントがA2Aプロトコルで接続できることを確認できました。

最後に

ADKやA2Aのサンプル実装としてa2a-samples/samples/python/agentsを参考にしました。A2Aの公式ドキュメントも丁寧にまとめられているので、繰り返し読み返したいと思います。

今回A2Aを実装してみて、将来的なAgent Marketplaceを見据えた設計になっていることを実感しました。例えば、AgentCardだけをMarketplace上に公開しておけば、Agentが実際に稼働するサーバーはどこでも良いことになります。ロードマップにはAgent Registoryなども予定されているようですし、Google Cloudでは最近Agentspaceが全ユーザーに公開されたので、機能が十分に整ったら触ってみたいと思います。

ここまで読んでいただきありがとうございました！

事前準備

check_prime_agentの実装

Agentをローカルで起動する

A2Aエージェントとして実装する

ローカルで検証する

Cloud Runへデプロイする

Cloud RunにデプロイしたA2Aエージェントを検証する

dice_agentの実装

A2Aエージェントを呼び出すエージェントの実装

ローカルで検証する

Cloud Runにデプロイして接続する

最後に

Discussion