チャットベースでさらにMCPサーバを試す回【LangChain MCP Adapters】

お疲れ様です。

前回langchain-mcp-adaptersを使ってチャットボットでMCPサーバを利用した返答を出力させる実装をしました。

fallpoke-tech.hatenadiary.jp

その際にMCPサーバを使った返答生成の部分だけを単体で実行できるコードを作っていたので、今回はそれを使ってみようと思います。

ソースコード

使用したソースコードの全体はこちらのGitHubを参照ください。 github.com

メインで実行するコードは以下になります。コマンドライン引数でMCPサーバのconfigファイルを指定して実行、実行時に質問文を標準入力する形です。

実行コマンド例
python response_gemini_mcp.py --config config\mcp_config_fetch.json

# 定数
TEMPLATE = """
あなたは質問応答タスクのアシスタントです。
登録されているツールを使って質問に答えてください。

質問: {question}
答え: {agent_scratchpad}
"""

# AIの返答を作成
async def gemini_mcp_generator(user_input, mcp_config_path):        
    # モデルを準備
    llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash-preview-05-20")
    # プロンプトを設定
    prompt = PromptTemplate.from_template(TEMPLATE)
    
    with open(mcp_config_path, mode="r") as f:
        mcp_config = json.load(f)
    
    mcp_client = MultiServerMCPClient(mcp_config["mcpServers"])
    tools = await mcp_client.get_tools()
    
    # エージェントを用意
    agent = create_tool_calling_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools)
    
    # 返答を取得
    response = await executor.ainvoke({"question": user_input})
    
    print("\n返答：\n")
    print(response["output"])
    
    
if __name__ == "__main__":
    # コマンドライン引数の設定
    parser = ArgumentParser()
    parser.add_argument("--config", type=str)
    args = parser.parse_args()
    
    # 単体テスト
    mcp_config_path = Path(args.config)
    
    # コマンドラインから入力
    user_input = input("質問を入力 >>> ")
    
    asyncio.run(gemini_mcp_generator(user_input, mcp_config_path))

ちなみに前回使用した「fetch」を使用するとこんな感じ。
fetch

「Git」のMCPサーバも簡単に試せそうだったのでやってみます。 jsonファイルで下記のようにconfigを作成します。

{
    "mcpServers": {
        "git": {
            "command": "uvx",
            "args": ["mcp-server-git"],
            "transport": "stdio"
        }
    }
}