`langchain-mcp-adapters` を使用した LangChain と MCP サーバーの連携 #Python

# weather_server.py
from mcp.server.fastmcp import FastMCP
import uvicorn # 必要: pip install uvicorn fastapi sse-starlette

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """指定された場所の天気を取得します。"""
    print(f"サーバー側: get_weather({location}) を実行中")
    # 実際のシナリオでは、ここで天気APIを呼び出す
    return f"{location}はいつも晴天です"

if __name__ == "__main__":
    # SSEトランスポートを使用してサーバーを実行 (uvicornのようなASGIサーバーが必要)
    # mcp ライブラリは暗黙的にSSE用のFastAPIアプリを作成します。
    # デフォルトでは、ポート8000の /sse エンドポイントで実行されます。
    print("天気MCPサーバーをSSE経由でポート8000で起動中...")
    # 手動で実行する場合: uvicorn.run(mcp.app, host="0.0.0.0", port=8000)
    # mcp.run の便利な方法を使用する場合:
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

import asyncio
import os
from langchain_mcp_adapters.client import MultiServerMCPClient, StdioConnection, SSEConnection
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# --- 重要: パスを更新してください ---
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# 天気サーバーは別途実行し、URL経由で接続
# ---

async def main():
    model = ChatOpenAI(model="gpt-4o")

    # 複数のサーバーの接続情報を定義
    # キー (例: "math_service") は接続を識別するための一意な名前
    # 値は StdioConnection または SSEConnection 型の辞書
    server_connections: dict[str, StdioConnection | SSEConnection] = {
        "math_service": { # この接続の一意な名前
            "transport": "stdio",
            "command": "python",
            "args": [math_server_script_path],
            # 必要であれば他の StdioConnection パラメータを追加 (env, cwd など)
        },
        "weather_service": { # この接続の一意な名前
            "transport": "sse",
            "url": "http://localhost:8000/sse", # weather_server が実行されているURL
            # 必要であれば他の SSEConnection パラメータを追加 (headers, timeout など)
        }
    }

    print("複数のMCPサーバーに接続中...")
    # MultiServerMCPClient コンテキストマネージャを使用
    # __aenter__ で定義された接続情報に基づいて各サーバーに接続し、セッションを初期化
    async with MultiServerMCPClient(connections=server_connections) as client:
        print("接続が確立されました。")

        # *すべて* の接続済みサーバーから *すべて* のツールを取得
        # client.get_tools() は内部で保持している全サーバーのツールリストを結合して返す
        all_tools = client.get_tools()
        print(f"ロードされたツール: {[tool.name for tool in all_tools]}")

        # 結合されたツールリストでエージェントを作成
        agent = create_react_agent(model, all_tools)

        # --- エージェントとの対話 ---
        print("n算術クエリでエージェントを呼び出し中...")
        math_inputs = {"messages": [("human", "(3 + 5) * 12 は？")]}
        math_response = await agent.ainvoke(math_inputs)
        print("算術応答:", math_response['messages'][-1].content)

        print("n天気クエリでエージェントを呼び出し中...")
        weather_inputs = {"messages": [("human", "ニューヨークの天気は？")]}
        weather_response = await agent.ainvoke(weather_inputs)
        print("天気応答:", weather_response['messages'][-1].content)

        # --- プロンプト取得の例 ---
        # print("n算術サーバーのプロンプトを取得中...")
        # try:
        #     # client.get_prompt で指定したサーバーからプロンプトを取得
        #     prompt_messages = await client.get_prompt(
        #         server_name="math_service", # connections で定義した名前を使用
        #         prompt_name="configure_assistant", # MCPサーバーで定義されたプロンプト名
        #         arguments={"skills": "基本的な算術"} # プロンプトが必要とする引数
        #     )
        #     print("取得したプロンプト:", prompt_messages)
        # except Exception as e:
        #     print(f"プロンプトの取得に失敗しました: {e}")


if __name__ == "__main__":
    # 最初に別のターミナルで天気サーバーを起動します:
    # python weather_server.py
    # 次に、このクライアントスクリプトを実行します:
    asyncio.run(main())