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

## はじめに

情報が出てから少々時間がたったかもしれない [A2A](https://github.com/google/A2A) に、遅ればせながら入門してみます。


https://x.com/googlecloud_jp/status/1910196039711142294

●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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/26ccd048-691d-4afc-9aa6-318b3b84705a.png)


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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/bc845930-0269-4559-8625-170d75a2586f.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/ca818b3b-6429-46fa-801f-692e06f4273c.png)




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

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

### 必要な環境

「[Set up Your Environment](https://google.github.io/A2A/tutorials/python/2-setup/)」というページを見ると、試すのに必要な環境の情報が掲載されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/11d298a2-cda9-4bb9-b5ac-a9258cd728db.png)

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

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

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

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


### 下準備

uvコマンドを使って、下準備を進めます。[公式ページ](https://google.github.io/A2A/tutorials/python/3-create-project/)に書かれた以下のコマンドを実行していきます。


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

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

```zsh
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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/af21f720-d277-4b52-9ac0-27cd2914970d.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/d6bbe24c-e361-4b74-a2e2-f8e7de1cd983.png)

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


### Agent Skills

チュートリアルのページ左にあるメニューで示すと、次は以下の [Agent Skills](https://google.github.io/A2A/tutorials/python/4-agent-skills/) の部分です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/ca27b6c9-e6a2-4b6a-bef9-80c075d3cebe.png)


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

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


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

```python
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()
```


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

```zsh
uv run my-project
```


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

```zsh
(.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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/d37a2efc-7384-4696-a55d-107202cd4a1c.png)


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


#### Agent Skills のエラーの対処

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/e0793e8d-8b23-4c00-b8a4-1ad22f7b9ea8.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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/dec18c86-fa3a-4e5b-bfdd-211c2b14e22b.png)

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

```python
# 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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/024df562-aa63-49a4-a332-ab65e9073f63.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](https://pypi.org/project/click/) のページを見てみると、コマンドラインインタフェースの作成を補助してくれるライブラリのようです。

```zsh
uv add click
```


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


```python
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・ポート、エージェントが対応できる内容について記載されているようです。

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

```zsh
uv run my-project
```

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/a66bac89-9a6c-45b8-8b7e-cd9d2075735c.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/e1798c25-95ff-472a-aef5-f8562d6e95b4.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 


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

```python: 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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/e93a4939-2180-447c-9c4a-266e283e39b7.png)


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


```python:
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()
```


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

```zsh
uv run my-project
```


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/7e8d80ba-5c63-4344-b6b7-1543540f2573.png)


ひとまず、公式手順に従ってサーバーを立ち上げるところまでは進められました。[Interact with server](https://google.github.io/A2A/tutorials/python/7-interact-with-server/) 以降の残りの手順も、別途進めていければと思います。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/32a584ac-838d-4a2a-b961-375c5f62a442.png)





## 【追記】 いったん公式デモを試した話

上記チュートリアルの手順に関し、続きの内容がかたまってなさそうな感じもあったため、いったん以下の記事に書いたとおり A2A の公式デモアプリを試しました。

●【Agent2Agent】 公式の A2A のデモアプリを試してみる（3つのエージェントを組み合わせる） - Qiita
　https://qiita.com/youtoy/items/da42061852bf9ad3b79b