はじめまして、ふっきーです

LangGraphのマルチエージェントグラフで、ClaudeCodeをツールとして持ったエージェントを使いたいなと思ったときに、サンプルの方法ではうまくいかなかったので力技で解決した方法を紹介します。

ClaudeCodeをエージェントとして組み込んだマルチエージェントグラフを”どうしても”構築したい方は参考にしてください。

これはベストプラクティスではないです
他に実現する方法を知っている方いらっしゃれば、教えてください！！

なぜうまくいかないのか(予想)

公式のサンプル実装を参考にすると、ざっくり以下のような実装になると思います。

ClaudeCodeMCP = {
    "command": "claude",
    "args": ["mcp", "serve"],
    "env": {},
    "transport": "stdio"
}

async def invoke_claude_code_agent(query: str) -> str:
    async with MultiServerMCPClient({"claude_code": ClaudeCodeMCP}) as client:
        # MCPをToolとして渡しつつ、エージェントを作成する
        agent = create_react_agent(ChatOpenAI(model="o3-mini"), client.get_tools())
        response = await agent.ainvoke({"messages": query})
        return response["messages"][-1].content

await invoke_claude_code_agent("hello")  # 一生待ち続ける

おそらくClaudeCodeのMCPサーバーの仕様上、ツールとしてClaudeCodeのMCPサーバーを起動した場合に、ツール実行後も起動したままになってしまいます。

たぶんチャットアプリであれば問題ないと思われますが、マルチエージェントグラフにおいてはエージェントによるツールを実行後、MCPサーバーが終了するのを待機し続けてしまいます。

結果、永久に来ないMCPサーバーの終了をずっと待機する状態となり、処理が止まってしまうといった挙動になりました。

いろいろ試してみた結果、returnやbreakでもうまく待機状態から抜け出すことはできませんでした。。。

どうしたか

ClaudeCodeMCPをツールとして持ったエージェントを別プロセスで稼働させて、応答受け取ったらエージェントをプロセスごとkillするような実装にしました。

エージェントとのメッセージのやりとりは、Queueを使っています。

実装

エージェントクラス

import asyncio
import multiprocessing
from time import sleep
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent


class ClaudeCodeAgent():
    def __init__(self):
        self.send_queue = multiprocessing.Queue()
        self.recv_queue = multiprocessing.Queue()

    @property
    def claude_code_mcp(self):
        return {
            "command": "claude",
            "args": ["mcp", "serve"],
            "env": {},
            "transport": "stdio"
        }

    async def _aserve(self):
        async with MultiServerMCPClient({"claude_code": ClaudeCodeMCP}) as client:

            # MCPをToolとして渡しつつ、エージェントを作成する
            agent = create_react_agent(ChatOpenAI(model="o3-mini"), client.get_tools())

            while True:
                if not self.send_queue.empty():  # MCPサーバーがメッセージを受信するまで待機
                    query = self.send_queue.get()
                    response = await agent.ainvoke({"messages": query})
                    self.recv_queue.put(response["messages"][-1].content)

    def _serve(self):
        asyncio.run(self._aserve())

    def invoke(self, query: str) -> str:
        # サブプロセスを起動
        p = multiprocessing.Process(target=self._serve)
        p.start()

        # メッセージ送信
        self.send_queue.put(query)

        # メッセージを受信するまで待機
        while self.recv_queue.empty():
            sleep(1)

        # メッセージ受信
        recv = self.recv_queue.get()
        p.terminate()  # 応答を受け取ったら、mcpサーバーを無理やり閉じる
        return recv

呼び出し元

claude_code_agent = ClaudeCodeAgent()
res = claude_code_agent.invoke("hello_world.pyを実装して")
print(f"res => {res}")

応答

res => hello_world.py の実装が完了しました。内容は以下の通りです:

#!/usr/bin/env python3

"""A simple hello world script."""

def main():
    print("Hello, world!")

if __name__ == "__main__":
    main()

応答を受け取った後、プロセス全体が終了したので、処理が止まってしまう現象を回避できました！

また、上記の応答通りの実装になっていたので、ファイル作成等もできていて、ClaudeCode?としての動きも問題なさそうです。

これで、マルチエージェントグラフにClaudeCodeを疑似的に組み込めるようになりました！

やったぜ！！

最後に

”他に方法はなかったのか”と今も思いますが、とにかく動いてよかったです。

他にもいろいろマルチエージェントグラフを使った、アプリやツールを作っています。
つぶやき少なめですが、興味ある方ぜひフォローしてください！！