LoginSignup
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@youtoy(Yosuke Toyota)inKDDI株式会社

【Agent2Agent Protocol】 公式の Python を使った A2Aチュートリアルのサーバー立ち上げまでを試す（バグの対処も）

Last updated at Posted at 2025-05-10

はじめに

情報が出てから少々時間がたったかもしれない A2A に、遅ればせながら入門してみます。

●Agent2Agent プロトコル（A2A）を発表：エージェントの相互運用性の新時代 | Google Cloud 公式ブログ
https://cloud.google.com/blog/ja/products/ai-machine-learning/a2a-a-new-era-of-agent-interoperability

今回の内容

A2A に入門するのに、どんなコンテンツに手をつけようかと調べてみていたら、下記の公式チュートリアルを見かけたので、これを試すことにしました。

●Introduction - Agent2Agent Protocol (A2A)
　https://google.github.io/A2A/tutorials/python/1-introduction/

image.png

ページの内容が少し変わった？

記事を途中まで書いて、再度、上記にアクセスしたところ、チュートリアルの画面が変わっていたので、更新後のページも掲載しておきます。

image.png

image.png

チュートリアルを進めていく

それでは早速、チュートリアルを進めていきます。

必要な環境

「Set up Your Environment」というページを見ると、試すのに必要な環境の情報が掲載されています。

image.png

自分の場合、以下の組み合わせでそろっている状態でした。

  • VS Code
  • iTerm (Mac) or just the Terminal in VS Code
  • Python（>= 3.12）
  • uv

また上記のページには、環境を確認するためのコマンドとして、以下が掲載されていました。

echo 'import sys; print(sys.version)' | uv run -

下準備

uvコマンドを使って、下準備を進めます。公式ページに書かれた以下のコマンドを実行していきます。

uv init --package my-project
cd my-project
uv venv .venv
source .venv/bin/activate

上記の最後のコマンドで、仮想環境がアクティベートされた状態になりました。その中で、さらにパッケージのインストール・ファイルの作成も行います（ここで作成した 2つのファイルは、少し後の手順で実装を行うようです）。

uv add git+https://github.com/google/A2A#subdirectory=samples/python

touch src/my_project/agent.py
touch src/my_project/task_manager.py

また、以下の確認も行っておきます。

image.png

以下のとおり、上記と同じ結果を得られることを確認できました。

image.png

現時点までの内容は問題なさそうです。

Agent Skills

チュートリアルのページ左にあるメニューで示すと、次は以下の Agent Skills の部分です。

image.png

ここでは、現時点では以下の内容で用意されている src/my-project/__init__.py の内容を編集するようです。

def main() -> None:
    print("Hello from my-project!")

書きかえた後の内容は、以下のとおりです。

import google_a2a
from google_a2a.common.types import AgentSkill

def main():
  skill = AgentSkill(
    id="my-project-echo-skill",
    name="Echo Tool",
    description="Echos the input given",
    tags=["echo", "repeater"],
    examples=["I will see this echoed back to me"],
    inputModes=["text"],
    outputModes=["text"],
  )
  print(skill)

if __name__ == "__main__":
  main()

そして以下のコマンドで、上で実装したスキルの内容が表示されることを確認します。

uv run my-project

しかし想定通りの結果は得られず、以下のエラーメッセージが出ました。

(.venv) % uv run my-project

Traceback (most recent call last):
  File "/【プロジェクトを置いたルートのパス】/my-project/.venv/bin/my-project", line 4, in <module>
    from my_project import main
  File "/【プロジェクトを置いたルートのパス】/my-project/src/my_project/__init__.py", line 1, in <module>
    import google_a2a
ModuleNotFoundError: No module named 'google_a2a'

image.png

「google_a2a」が見つからない、という内容ですが、調べてみた感じだとパッケージの名称ではなさそうでした。

Agent Skills のエラーの対処

「google_a2a」はパッケージではなさそうで、当初 Web検索で情報を見つけられませんでした。それで「google_a2a」をキーワードに GitHub上を検索したところ、以下の Issue を見つけることができました。

image.png

それらを見てみると、以下に関係する内容が書かれていそうでした。

●[Bug]: Unable to find the google_a2a module when following the Python tutorial · Issue #382 · google/A2A
　https://github.com/google/A2A/issues/382

この Issue内に書かれたリンクをたどって情報を見ていくと、以下を見つけられました。

●3. Creating a project
　https://djsamseng.github.io/A2A/#/tutorials/python/3_create_a_project

image.png

ここでは 2つの方法が書いてあり、自分はコードの書きかえで対処できる 2つ目のほうを試しました。具体的には src/my-project/__init__.py のコードを以下のようにします（エラーが出る部分と、そこを修正した後の両方を掲載しています）。

# import google_a2a
# from google_a2a.common.types import AgentSkill
import common
from common.types import AgentSkill

def main():
  skill = AgentSkill(
    id="my-project-echo-skill",
    name="Echo Tool",
    description="Echos the input given",
    tags=["echo", "repeater"],
    examples=["I will see this echoed back to me"],
    inputModes=["text"],
    outputModes=["text"],
  )
  print(skill)

if __name__ == "__main__":
  main()

上記の修正を行った後に、再度 uv run my-project を実行したところ、以下の想定通りの出力を得られました。

image.png

エージェントカードの追加

次に、以下の Add Agent Card の手順へと進みます。

●Add Agent Card - Agent2Agent Protocol (A2A)
　https://google.github.io/A2A/tutorials/python/5-add-agent-card/#test-run

ここで手順に書いてある以下を実行します。
click · PyPI のページを見てみると、コマンドラインインタフェースの作成を補助してくれるライブラリのようです。

uv add click

その後、 src/my-project/__init__.py を以下のように書きかえました。
※ 冒頭で使われてなさそうな dotenv がインポートされていて、それがエラーの原因にもなったので、コメントアウトしています

import logging

import click
# from dotenv import load_dotenv
import common
from common.types import AgentSkill, AgentCapabilities, AgentCard

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

@click.command()
@click.option("--host", default="localhost")
@click.option("--port", default=10002)
def main(host, port):
  skill = AgentSkill(
    id="my-project-echo-skill",
    name="Echo Tool",
    description="Echos the input given",
    tags=["echo", "repeater"],
    examples=["I will see this echoed back to me"],
    inputModes=["text"],
    outputModes=["text"],
  )

  capabilities = AgentCapabilities()
  agent_card = AgentCard(
    name="Echo Agent",
    description="This agent echos the input given",
    url=f"http://{host}:{port}/",
    version="0.1.0",
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    capabilities=capabilities,
    skills=[skill]
  )

  logging.info(agent_card)

if __name__ == "__main__":
  main()

手順の中で、2箇所に分かれて書かれていたものを両方対応した後のものです。ここで、どんな入出力に対応できるか（上記だとテキストのみ）や、エージェントにアクセスするための URL・ポート、エージェントが対応できる内容について記載されているようです。

そして、以下を実行します。

uv run my-project

そうすると以下のとおり、先ほど追加実装した内容が想定通りに出力されていました。

image.png

ちなみに当初、コード冒頭のコメントアウトをしていなかった時は、以下の dotenv関連のエラーが出ていました。

image.png

サーバーの起動

今回の記事で進める最後のステップです。以下に書かれた内容を進め、サーバーを起動します。

●Start server - Agent2Agent Protocol (A2A)
　https://google.github.io/A2A/tutorials/python/6-start-server/

task_manager.py の実装

上で書いていた手順で、空のファイルとして作成していた以下のうち、task_manager.py のほうの実装を行います。

  • src/my-project/task_manager.py
  • src/my-project/agent.py

実装する内容は、以下のとおりです。

src/my-project/task_manager.py
from typing import AsyncIterable

# import google_a2a
# from google_a2a.common.server.task_manager import InMemoryTaskManager
# from google_a2a.common.types import (
import common
from common.server.task_manager import InMemoryTaskManager
from common.types import (
  Artifact,
  JSONRPCResponse,
  Message,
  SendTaskRequest,
  SendTaskResponse,
  SendTaskStreamingRequest,
  SendTaskStreamingResponse,
  Task,
  TaskState,
  TaskStatus,
  TaskStatusUpdateEvent,
)

class MyAgentTaskManager(InMemoryTaskManager):
  def __init__(self):
    super().__init__()

  async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:
    # Upsert a task stored by InMemoryTaskManager
    await self.upsert_task(request.params)

    task_id = request.params.id
    # Our custom logic that simply marks the task as complete
    # and returns the echo text
    received_text = request.params.message.parts[0].text
    task = await self._update_task(
      task_id=task_id,
      task_state=TaskState.COMPLETED,
      response_text=f"on_send_task received: {received_text}"
    )

    # Send the response
    return SendTaskResponse(id=request.id, result=task)

  async def on_send_task_subscribe(
    self,
    request: SendTaskStreamingRequest
  ) -> AsyncIterable[SendTaskStreamingResponse] | JSONRPCResponse:
    pass

  async def _update_task(
    self,
    task_id: str,
    task_state: TaskState,
    response_text: str,
  ) -> Task:
    task = self.tasks[task_id]
    agent_response_parts = [
      {
        "type": "text",
        "text": response_text,
      }
    ]
    task.status = TaskStatus(
      state=task_state,
      message=Message(
        role="agent",
        parts=agent_response_parts,
      )
    )
    task.artifacts = [
      Artifact(
        parts=agent_response_parts,
      )
    ]
    return task

init.py の修正

ここで src/my-project/__init__.py に、サーバー関連の処理を少し追加します。

image.png

追加を行った後のコードは以下のとおりです。

import logging

import click
# from dotenv import load_dotenv
import common
from common.types import AgentSkill, AgentCapabilities, AgentCard
# from google_a2a.common.server import A2AServer
from common.server import A2AServer
from my_project.task_manager import MyAgentTaskManager

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

@click.command()
@click.option("--host", default="localhost")
@click.option("--port", default=10002)
def main(host, port):
  skill = AgentSkill(
    id="my-project-echo-skill",
    name="Echo Tool",
    description="Echos the input given",
    tags=["echo", "repeater"],
    examples=["I will see this echoed back to me"],
    inputModes=["text"],
    outputModes=["text"],
  )

  capabilities = AgentCapabilities()
  agent_card = AgentCard(
    name="Echo Agent",
    description="This agent echos the input given",
    url=f"http://{host}:{port}/",
    version="0.1.0",
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    capabilities=capabilities,
    skills=[skill]
  )

#   logging.info(agent_card)

  task_manager = MyAgentTaskManager()
  server = A2AServer(
    agent_card=agent_card,
    task_manager=task_manager,
    host=host,
    port=port,
  )
  server.start()

if __name__ == "__main__":
  main()

ここまでの内容ができたら、以下を実行します。

uv run my-project

そうすると、公式手順に書かれているのと同じ結果を得ることができました。

image.png

ひとまず、公式手順に従ってサーバーを立ち上げるところまでは進められました。Interact with server 以降の残りの手順も、別途進めていければと思います。

image.png

2
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?