📘

langgraph-a2a-clientとlanggraph-a2a-serverを用いてお手軽A2A対応マルチエージェント実装

2025/10/07に公開

 概要
本記事は、前回の記事の続編として、LangGraphでA2A Protocolを使用したマルチエージェントシステムをより簡単に実装する方法を紹介します。
前回の記事では、A2A Protocolを使用するためにLangGraphAgentAdapterやLangGraphAgentExecutorなどのカスタムクラスを実装する必要がありました。

これをライブラリ化したlanggraph-a2a-clientとlanggraph-a2a-serverを使用することで、これらの複雑な実装を大幅に簡略化します。

 モチベーション前回の実装では、A2A Protocolに準拠したマルチエージェントシステムを構築できましたが、以下のような課題がありました：
カスタムアダプターとエグゼキューターの実装が必要
A2A Protocolの詳細な仕様を理解する必要がある
メンテナンスコストが高い
これらの課題を解決するため、筆者がlanggraph-a2a-clientとlanggraph-a2a-serverというライブラリを作成しました。これらのライブラリを使用することで、開発者はビジネスロジックに集中でき、より迅速にマルチエージェントシステムを構築できます。

 サンプルコード本記事で紹介しているコードはGitHubで公開しています：
https://github.com/5enxia/langgraph-multiagent-with-a2a
特にa2a-with-library/ディレクトリに今回紹介するコードがあります。

 環境設定
 必要なライブラリのインストール前提として、パッケージ管理にuvを使用しています。筆者はpython 3.13を使用して実装しています。
!langgraph-a2a-serverとlanggraph-a2a-clientは、前回の記事で実装したカスタムコードをライブラリ化したもので、筆者が作成・公開しています。
uv add langgraph-a2a-server langgraph-a2a-client langchain langchain-openai langgraph
または、pipを使用する場合：
pip install langgraph-a2a-server langgraph-a2a-client langchain langchain-openai langgraph

 環境変数の設定LLMのAPIキーを設定します。この例では、OpenAIのモデルを使用しています。
export OPENAI_API_KEY="your_openai_api_key"

 ライブラリを用いたマルチエージェント実装
 プロジェクト構成今回の実装は以下のような構成になります：
a2a-with-library/
├── sub_agents/
│   ├── currency_agent.py   # 通貨エージェント（A2Aサーバー）
│   └── weather_agent.py    # 天気エージェント（A2Aサーバー）
└── supervisor/
    └── __main__.py         # スーパーバイザーエージェント（A2Aクライアント）
前回の実装と比較して、common/ディレクトリ（adapter.pyとagent_executor.py）が不要になり、コード量が大幅に削減されています。

 サブエージェントの実装
 Currency Agent（通貨エージェント）langgraph-a2a-serverを使用することで、LangGraphエージェントを簡単にA2Aサーバーとして公開できます。
# a2a-with-library/sub_agents/currency_agent.py
import logging

from a2a.types import AgentSkill

# LangGraph A2A Server
from langgraph_a2a_server import A2AServer

# LangChain/LangGraph
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model


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

HOST = '0.0.0.0'
PORT = 10000


@tool
def get_exchange_rate():
    """Get the exchange rate between USD and JPY."""
    return "1 USD = 147円"


# Create LangGraph agent
agent = create_react_agent(
    model=init_chat_model(model='gpt-4.1-nano'),
    tools=[get_exchange_rate],
    checkpointer=MemorySaver(),
)

# Create A2A server with the agent
server = A2AServer(
    graph=agent,
    name='Currency Agent',
    description='Helps with exchange rates for currencies',
    host=HOST,
    port=PORT,
    skills=[
        AgentSkill(
            id='convert_currency',
            name='Currency Exchange Rates Tool',
            description='Helps with exchange values between various currencies',
            tags=['currency conversion', 'currency exchange'],
            examples=['What is exchange rate between USD and GBP?'],
        )
    ],
)

# Start the server
if __name__ == '__main__':
    server.serve()
前回の実装との比較：
✅ LangGraphAgentAdapterが不要
✅ LangGraphAgentExecutorが不要
✅ A2AStarletteApplicationの手動設定が不要
✅ Task管理やPush通知の設定が自動化
✅ コード量が約70%削減
A2AServerクラスが、前回実装したカスタムアダプターやエグゼキューターの役割をすべて担当してくれます。開発者は、LangGraphエージェントを作成してA2AServerにラップするだけです。

 Weather Agent（天気エージェント）同様に、天気情報エージェントも簡潔に実装できます。
# a2a-with-library/sub_agents/weather_agent.py
import logging

from a2a.types import AgentSkill

# LangGraph A2A Server
from langgraph_a2a_server import A2AServer

# LangChain/LangGraph
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain.chat_models import init_chat_model


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

HOST = '0.0.0.0'
PORT = 20000


@tool
def get_weather(city_name: str) -> str:
    """Get the current weather."""
    return f"{city_name} is Sunny"


# Create LangGraph agent
agent = create_react_agent(
    model=init_chat_model(model='gpt-4.1-nano'),
    tools=[get_weather],
    checkpointer=MemorySaver(),
)

# Create A2A server with the agent
server = A2AServer(
    graph=agent,
    name='Weather Expert',
    description='Fetch the current weather for a location',
    host=HOST,
    port=PORT,
    skills=[
        AgentSkill(
            id='get_weather',
            name='Get Weather',
            description='Fetch the current weather for a location',
            tags=['weather', 'location'],
            examples=['What is the weather like in New York?', 'Tell me the weather in San Francisco'],
        )
    ],
)

# Start the server
if __name__ == '__main__':
    server.serve()

 スーパーバイザーエージェントの実装スーパーバイザーエージェントでは、langgraph-a2a-clientのA2AClientToolProviderを使用します。
# a2a-with-library/supervisor/__main__.py
import asyncio
import logging

# LangGraph A2A Client
from langgraph_a2a_client import A2AClientToolProvider

# LangChain/LangGraph
from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model


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


# Create A2A client tool provider with known agent URLs
provider = A2AClientToolProvider(
    known_agent_urls=[
        "http://0.0.0.0:10000",  # Currency Agent
        "http://0.0.0.0:20000",  # Weather Expert
    ]
)

# Create supervisor agent with A2A client tools
supervisor = create_react_agent(
    model=init_chat_model(model='gpt-4.1-mini'),
    tools=provider.tools,
    name="supervisor",
    prompt="You are a team supervisor managing a currency agent and a weather information agent."
)


async def main():
    response = await supervisor.ainvoke({
        "messages": [
            {
                "role": "user",
                "content": "今日の東京の天気は?"
                # "content": "1ドルは為替で何円ですか?"
            }
        ]
    })
    messages = response['messages']
    for message in messages:
        print(message.content)


if __name__ == '__main__':
    asyncio.run(main())

 実行方法
 1. サブエージェントの起動まず、別々のターミナルでサブエージェントを起動します。
# ターミナル1: 通貨エージェント
uv run a2a-with-library/sub_agents/currency_agent

# ターミナル2: 天気エージェント
uv run a2a-with-library/sub_agents/weather_agent
各エージェントがHTTPサーバーとして起動し、それぞれ以下のエンドポイントでアクセス可能になります：
Currency Agent: http://0.0.0.0:10000
Weather Agent: http://0.0.0.0:20000

 2. スーパーバイザーエージェントの実行サブエージェントが起動したら、スーパーバイザーを実行します。
# ターミナル3: スーパーバイザー
uv run a2a-with-library/supervisor
スーパーバイザーは、各サブエージェントの/.well-known/agent-card.jsonにアクセスして情報を取得し、ユーザーからの質問に基づいて適切なサブエージェントにタスクを割り当てます。

 実行例天気に関する質問：
content: "今日の東京の天気は?"
スーパーバイザーがWeather Agentにタスクを委譲し、「東京 is Sunny」という回答が返されます。
通貨に関する質問：
content: "1ドルは為替で何円ですか?"
スーパーバイザーがCurrency Agentにタスクを委譲し、「1 USD = 147円」という回答が返されます。

 ライブラリを使用する利点
 1. コード量の大幅削減前回の実装と比較して、以下のファイルが不要になりました：

common/adapter.py (約120行)

common/agent_executor.py (約90行)
カスタムA2AClientToolProvider (約300行以上)
合計で約500行以上のコードを削減できました。

 2. 学習コストの低減A2A Protocolの詳細な実装を理解する必要がない
LangGraphの知識があれば、すぐに使い始められる

 まとめ本記事では、筆者が作成したlanggraph-a2a-clientとlanggraph-a2a-serverライブラリを使用して、LangGraphベースのA2A対応マルチエージェントシステムを簡単に構築する方法を紹介しました。これらのライブラリは、前回の記事で実装したカスタムコードをベースに、より使いやすく汎用的な形にパッケージ化したものです。
主な改善点：

コード量の大幅削減: 約500行以上のコードが不要に

実装の簡素化: Executorなどの実装が不要に

学習コストの低減: ほぼLangGraphの知識だけで実装可能
前回の記事でA2A Protocolの仕組みを理解した上で、このライブラリを使用することで、より実用的で保守性の高いマルチエージェントシステムを構築できます。
今後、LangGraphとA2A Protocolを組み合わせた疎結合なマルチエージェントシステムの開発をされる方の参考になれば幸いです。

 参考
 前回の記事langgraph-supervisorで作成したSupervisor型マルチエージェントをA2A Protocolを用いて再実装

 ライブラリ（筆者作成）langgraph-a2a-server (PyPI)
langgraph-a2a-client (PyPI)

 サンプルコードGitHub - 5enxia/langgraph-multiagent-with-a2a
https://a2a-protocol.org/latest/

概要

モチベーション

サンプルコード

環境設定

必要なライブラリのインストール

環境変数の設定

ライブラリを用いたマルチエージェント実装

プロジェクト構成

サブエージェントの実装

Currency Agent（通貨エージェント）

Weather Agent（天気エージェント）

スーパーバイザーエージェントの実装

実行方法

1. サブエージェントの起動

2. スーパーバイザーエージェントの実行

実行例

ライブラリを使用する利点

1. コード量の大幅削減

2. 学習コストの低減

まとめ

参考

前回の記事

ライブラリ（筆者作成）

サンプルコード

Discussion