概要

arch

本記事は、前回の記事の続編として、LangGraphでA2A Protocolを使用したマルチエージェントシステムをより簡単に実装する方法を紹介します。

前回の記事では、A2A Protocolを使用するためにLangGraphAgentAdapterやLangGraphAgentExecutorなどのカスタムクラスを実装する必要がありました。
これをライブラリ化したlanggraph-a2a-clientとlanggraph-a2a-serverを使用することで、これらの複雑な実装を大幅に簡略化します。

モチベーション

前回の実装では、A2A Protocolに準拠したマルチエージェントシステムを構築できましたが、以下のような課題がありました：

カスタムアダプターとエグゼキューターの実装が必要
A2A Protocolの詳細な仕様を理解する必要がある
メンテナンスコストが高い

これらの課題を解決するため、筆者がlanggraph-a2a-clientとlanggraph-a2a-serverというライブラリを作成しました。これらのライブラリを使用することで、開発者はビジネスロジックに集中でき、より迅速にマルチエージェントシステムを構築できます。

サンプルコード

本記事で紹介しているコードはGitHubで公開しています：

特にa2a-with-library/ディレクトリに今回紹介するコードがあります。

環境設定

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

前提として、パッケージ管理にuvを使用しています。筆者はpython 3.13を使用して実装しています。

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を用いて再実装

ライブラリ（筆者作成）

サンプルコード

GitHub - 5enxia/langgraph-multiagent-with-a2a

https://a2a-protocol.org/latest/

Discussion