LoginSignup
5
5

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@nanami_bitwise(木村 七海)

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

Posted at

はじめに

Model Context Protocol (MCP) は、AIモデルが外部ツールやサービスとどのように対話するかを標準化することを目的としています。これは共通のインターフェースを定義し、異なるモデルやツールプロバイダーが効果的に通信できるようにします。MCPは、AIがアクセスできる機能（ツール）を記述するための言語非依存の方法を提供し、ツール呼び出しの要求と応答のフォーマットを標準化します。これにより、AIモデルは特定のツール実装の詳細を知らなくても、多様なツールセットを利用できるようになります。

しかし、これらのMCP準拠ツールをLangChainのような既存のAIフレームワークに直接統合するには、アダプテーションが必要です。LangChainは独自のツール抽象化（BaseTool）とエコシステムを持っているため、MCPツールをそのまま利用することはできません。

ここで langchain-mcp-adapters ライブラリが重要な役割を果たします。これは重要な架け橋として機能し、MCPツールをLangChainおよびその強力なエージェントフレームワークであるLangGraphが理解し利用できる形式にシームレスに変換します。このライブラリは軽量なラッパーを提供し、開発者はLangChainアプリケーション内で成長し続けるMCPツールのエコシステムを活用できるようになります。

主な機能:

  • MCPツール変換: MCPツールをLangChain互換の BaseTool オブジェクトに自動的に変換します。これにより、LangChainエージェント（ReActエージェントなど）がこれらのツールを他のLangChainネイティブツールと同様に扱うことができます。
  • マルチサーバークライアント: 複数のMCPサーバーに同時に接続し、さまざまなソースからツールを集約できる堅牢なクライアント (MultiServerMCPClient) を提供します。これにより、異なる機能を持つ複数のマイクロサービスのようなMCPサーバーを組み合わせて利用できます。
  • トランスポートの柔軟性: 標準入出力 (stdio) やサーバーセントイベント (sse) など、一般的なMCP通信トランスポートをサポートします。これにより、ローカルでの開発からネットワーク経由のリモートサーバー接続まで、様々なデプロイメントシナリオに対応できます。

このチュートリアルでは、MCPサーバーの設定方法、アダプターライブラリを使用したサーバーへの接続方法、そしてロードされたツールをLangGraphエージェントに統合する方法を順を追って説明します。

インストール

まず、必要なパッケージをインストールします。

pip install langchain-mcp-adapters langgraph langchain-openai # またはお好みのLangChain LLMインテグレーション

また、選択した言語モデルプロバイダーのAPIキーを設定する必要があります。通常、環境変数を設定することで行います。

export OPENAI_API_KEY=<your_openai_api_key>
# または export ANTHROPIC_API_KEY=<...> など

コアコンセプト

サンプルに進む前に、いくつかのコアコンセプトを理解することが不可欠です。

  1. MCPサーバー:

    • MCPサーバーは、AIモデルが呼び出すことができるツール（関数）を公開します。これは、特定の機能を提供するマイクロサービスと考えることができます。
    • mcp ライブラリ（langchain-mcp-adapters の依存関係）は、Pythonでこれらのサーバーを簡単に作成するための FastMCP のようなツールを提供します。
    • ツールは @mcp.tool() デコレーターを使用して定義されます。このデコレーターは、型ヒントとdocstringから入力スキーマ（ツールが必要とする引数とその型）を自動的に推論します。これにより、ツールのインターフェースが明確に定義されます。
    • サーバーは @mcp.prompt() を使用してプロンプトも定義できます。これは、構造化された会話の開始点や指示を提供します（例：システムプロンプトの設定）。
    • サーバーは、通信に使用するトランスポートメカニズムを指定して実行されます（例： mcp.run(transport="stdio") や mcp.run(transport="sse")）。stdio はサーバーを標準入出力を介して通信するサブプロセスとして実行します。これはローカルでの開発やテストに便利です。sse は通常、通信のためにシンプルなWebサーバー（多くの場合、FastAPIが内部で使用される）を実行します。これはネットワーク経由での接続に適しています。

  2. MCPクライアント (langchain-mcp-adapters):

    • クライアントの役割は、1つ以上のMCPサーバーに接続することです。
    • クライアントは、stdio や sse といった通信プロトコルの詳細を処理します。開発者は低レベルの通信を意識する必要はありません。
    • クライアントは、利用可能なツールのリストとその定義（名前、説明、入力スキーマ）をサーバーから取得します。
    • MultiServerMCPClient クラスは、特に複数のツールサーバーを扱う場合に、接続を管理するための主要な方法です。これは非同期コンテキストマネージャ (async with) として使用され、接続の確立とクリーンアップを自動的に処理します。

  3. ツール変換:

    • MCPツールには独自の定義形式があります。一方、LangChainは BaseTool クラス構造を使用します。これら二つの形式は互換性がありません。
    • langchain-mcp-adapters ライブラリは、このギャップを埋めるための変換機能を提供します。langchain_mcp_adapters.tools にある load_mcp_tools 関数は、アクティブな mcp.ClientSession を介してサーバーに接続し、MCPツールをリストアップし、それぞれをLangChainの StructuredTool でラップします。StructuredTool は、複雑な入力スキーマを持つツールに適したLangChainのツールクラスです。
    • このラッパーは、LangChainエージェントがツールを使用することを決定したときに、実際のMCPツール呼び出し（session.call_tool）を処理し、応答をLangChainが期待する形式（通常は文字列または ToolMessage）に正しくフォーマットします。これにより、エージェントはMCPツールの存在を意識することなく、シームレスにツールを利用できます。

  4. プロンプト変換:

    • ツールと同様に、MCPプロンプトは langchain_mcp_adapters.prompts の load_mcp_prompt を使用して取得できます。
    • この関数は、MCPサーバーからプロンプト構造を取得し、それをLangChainの HumanMessage または AIMessage オブジェクトのリストに変換します。これは、会話の初期化やエージェントへの指示（システムプロンプトなど）に適しています。

クイックスタート: シングルサーバーの例 (約400語)

簡単な例を構築しましょう：数学関数を提供するMCPサーバーと、それらの関数を使用するLangGraphエージェントです。

ステップ1: MCPサーバーの作成 (math_server.py)

# math_server.py
from mcp.server.fastmcp import FastMCP

# 名前を指定してMCPサーバーを初期化
mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """二つの数を加算します"""
    print(f"サーバー側: add({a}, {b}) を実行中") # サーバー側のログ
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """二つの数を乗算します"""
    print(f"サーバー側: multiply({a}, {b}) を実行中") # サーバー側のログ
    return a * b

# プロンプト定義の例
@mcp.prompt()
def configure_assistant(skills: str) -> list[dict]:
    """指定されたスキルでアシスタントを設定します。"""
    return [
        {
            "role": "assistant", # LangChainのAIMessageに対応
            "content": f"あなたは役立つアシスタントです。次のスキルを持っています: {skills}。常に一度に一つのツールのみを使用してください。",
        }
    ]

if __name__ == "__main__":
    # stdioトランスポートを使用してサーバーを実行
    print("算術MCPサーバーをstdio経由で起動中...")
    mcp.run(transport="stdio")

このコードを math_server.py として保存します。

ステップ2: クライアントとエージェントの作成 (client_app.py)

import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# --- 重要: このパスを更新してください ---
# math_server.py ファイルへの絶対パスを取得
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# ---

async def main():
    # モデルを初期化 (OpenAI GPT-4oを使用する例)
    model = ChatOpenAI(model="gpt-4o") # またはお好みのモデル

    # math_server.py スクリプトを実行するためのパラメータを設定
    server_params = StdioServerParameters(
        command="python", # 実行するコマンド
        args=[math_server_script_path], # コマンドライン引数（スクリプトパス）
        # cwd=..., env=... # オプション: 作業ディレクトリや環境変数
    )

    print("MCPサーバーに接続中...")
    # stdio_client コンテキストマネージャを使用して接続を確立
    async with stdio_client(server_params) as (read, write):
        # read/writeストリームを使用してClientSessionを作成
        # session_kwargs で ClientSession の挙動をカスタマイズ可能
        async with ClientSession(read, write) as session:
            print("セッションを初期化中...")
            # サーバーとのハンドシェイクを実行
            await session.initialize()
            print("セッションが初期化されました。")

            print("MCPツールをロード中...")
            # MCPツールを取得し、LangChainツールに変換
            # この関数が内部で session.list_tools() を呼び出し、
            # 返された MCPTool オブジェクトを convert_mcp_tool_to_langchain_tool で変換
            tools = await load_mcp_tools(session)
            print(f"ロードされたツール: {[tool.name for tool in tools]}")

            # モデルとロードされたツールを使用してLangGraph ReActエージェントを作成
            # ReActは推論(Reasoning)と行動(Action)を交互に行うエージェントの一種
            agent = create_react_agent(model, tools)

            print("エージェントを呼び出し中...")
            # エージェントを実行
            inputs = {"messages": [("human", "(3 + 5) * 12 は？")]} # ユーザーからの入力

            # イベントストリームを取得して処理の流れを監視 (推奨)
            async for event in agent.astream_events(inputs, version="v1"):
                print(f"イベント: {event['event']}, データ: {event['data']}") # イベントをストリーミングして観察

            # または最終的な応答を直接取得
            # final_response = await agent.ainvoke(inputs)
            # print("エージェントの応答:", final_response['messages'][-1].content)

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

これを math_server.py と同じディレクトリに client_app.py として保存します。

実行方法:

クライアントスクリプトを実行します:

python client_app.py

クライアントスクリプトは自動的に math_server.py をサブプロセスとして起動し、それに接続し、add と multiply ツールをロードし、LangGraphエージェントを使用してMCPサーバー経由でこれらのツールを呼び出すことで数学の問題を解決します。クライアントとサーバーの両方からのログが表示され、エージェントがツールを呼び出す様子（サーバー側のログ）とエージェントの思考プロセス（クライアント側の astream_events ログ）を確認できます。

複数のMCPサーバーへの接続

多くの場合、異なる専門サーバーからのツールを組み合わせたい場合があります。MultiServerMCPClient はこれを簡単に実現します。

ステップ1: 別のサーバーを作成 (weather_server.py)

SSEトランスポートを使用して実行される天気サーバーを作成しましょう。SSEは、サーバーからクライアントへの一方向のプッシュ通知に適したプロトコルです。

# 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)

これを weather_server.py として保存します。実行には uvicorn, fastapi, sse-starlette が必要になる場合があります (pip install uvicorn fastapi sse-starlette)。

ステップ2: MultiServerMCPClient を使用するようにクライアントを更新 (multi_client_app.py)

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())

これを multi_client_app.py として保存します。

実行方法:

  1. 一つのターミナルで天気サーバーを起動します: python weather_server.py
  2. 別のターミナルでマルチクライアントアプリを実行します: python multi_client_app.py

MultiServerMCPClient は math_server.py サブプロセス (stdio) を起動し、実行中の weather_server.py (sse) に接続します。そして、ツール (add, multiply, get_weather) を集約し、それらがLangGraphエージェントで利用可能になります。エージェントは、質問の内容に応じて適切なサーバーのツールを呼び出すことができます。

LangGraph APIサーバーとの統合

MCPツールを使用するLangGraphエージェントを、langgraph deploy を使用して永続的なAPIサービスとしてデプロイできます。重要なのは、LangGraphアプリケーションコンテキスト内で MultiServerMCPClient のライフサイクル（初期化とクリーンアップ）を正しく管理することです。これにはLangGraphの lifespan 機能を使用します。

graph.py ファイルを作成します:

# graph.py
from contextlib import asynccontextmanager
import os
from langchain_mcp_adapters.client import MultiServerMCPClient, StdioConnection, SSEConnection
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI # または Anthropic など

# --- 重要: パス/URLをサーバー環境に合わせて更新 ---
# LangGraphサーバーが実行される場所からの相対パスを想定
math_server_script_path = os.path.abspath("math_server.py")
weather_server_url = "http://localhost:8000/sse" # 天気サーバーは独立して実行されている必要あり
# ---

# 接続情報を定義 (サーバー環境でのパス/URLが正しいことを確認)
server_connections: dict[str, StdioConnection | SSEConnection] = {
    "math_service": {
        "transport": "stdio",
        "command": "python",
        "args": [math_server_script_path],
    },
    "weather_service": {
        "transport": "sse",
        "url": weather_server_url,
    }
}

model = ChatOpenAI(model="gpt-4o")

# 非同期コンテキストマネージャを使用してクライアントのセットアップ/ティアダウンを処理
# LangGraph は FastAPI の lifespan イベントハンドラを利用してこれを管理する
@asynccontextmanager
async def lifespan(_app): # LangGraph は lifespan 管理のためにこの構造を期待
    # MultiServerMCPClient コンテキスト内でクライアントを初期化
    async with MultiServerMCPClient(server_connections) as client:
        print("ライフスパン内でMCPクライアントが初期化されました。")
        # クライアントがアクティブなコンテキスト*内部*でエージェントを作成
        agent = create_react_agent(model, client.get_tools())
        # LangGraph がリクエスト処理に使用できるようにエージェントを yield する
        # ここで yield された辞書のキー (例: "agent") が entrypoint で参照される
        yield {"agent": agent}
    print("ライフスパン終了、MCPクライアントがクリーンアップされました。")

# lifespan がエージェントを yield する場合、個別の main グラフ定義は不要なことが多い

langgraph.json (または pyproject.toml の [tool.langgraph] セクション) を設定して、このグラフ定義をライフスパンマネージャーと共に使用します:

// langgraph.json (例)
{
  "dependencies": ["."], // または必要なパッケージを指定
  "graphs": {
    "my_mcp_agent": { // APIエンドポイント名 (例: /my_mcp_agent/invoke)
      "entrypoint": "graph:agent", // lifespan が yield した辞書のキーを参照
      "lifespan": "graph:lifespan" // 使用する lifespan 関数を指定
    }
  }
}

これで langgraph up を実行すると、lifespan 関数が実行され、MultiServerMCPClient が起動し (そして stdio の数学サーバーも起動し)、SSE サーバーに接続します。このコンテキスト内で作成されたエージェントがLangGraphによって提供されます。リクエストがあるたびに、この初期化済みのエージェントが使用されます。サーバーがシャットダウンする際には、lifespan の async with ブロックが終了し、クライアントと stdio サーバーが適切にクリーンアップされます。SSE 天気サーバーは依然として別途実行する必要があることに注意してください。

サーバートランスポート (stdio vs. SSE)

  • stdio:
    • 通信: サーバープロセスの標準入力および出力ストリーム経由。
    • 利点: ローカル開発のための簡単なセットアップ。クライアントがサーバーのライフサイクル（起動・停止）を管理します。ネットワーク設定が不要。
    • 欠点: クライアントとサーバーが密結合。分散システムや非Pythonサーバーにはあまり適していません。クライアント側で command と args の設定が必要です。サーバープロセスがクラッシュすると、クライアントも影響を受ける可能性があります。
  • sse (Server-Sent Events):
    • 通信: SSEプロトコルを使用してHTTP経由。サーバーはWebサービスとして動作します（多くの場合、FastAPI/Uvicornが暗黙的に使用されます）。
    • 利点: 標準的なWebプロトコル。ネットワーク化された/リモートのサーバーに適しており、異なる言語で実装されている可能性もあります。サーバーは独立して実行されます。スケーラビリティが高い。
    • 欠点: サーバーを別途実行・管理する必要があります。クライアント側で url の設定が必要です。ネットワーク遅延や信頼性の影響を受けます。

デプロイメントのニーズに基づいてトランスポートを選択してください。ローカルでの迅速なプロトタイピングには stdio が便利ですが、本番環境やマイクロサービスアーキテクチャでは sse の方が適していることが多いでしょう。

高度なクライアント設定

MultiServerMCPClient 内の StdioConnection および SSEConnection 辞書は、より細かい制御のための追加のオプションパラメータを受け入れます:

  • Stdio:
    • env: サブプロセス用のカスタム環境変数。
    • cwd: サブプロセスの作業ディレクトリ。
    • encoding: テキストエンコーディング（デフォルト: "utf-8"）。
    • encoding_error_handler: エンコーディングエラーの処理方法（デフォルト: "strict"）。
    • session_kwargs: 基盤となる mcp.ClientSession に渡される追加のキーワード引数（例：タイムアウト設定など）。
  • SSE:
    • headers: SSEエンドポイントに送信するカスタムHTTPヘッダー。
    • timeout: HTTP接続タイムアウト（デフォルト: 5秒）。
    • sse_read_timeout: SSE読み取りタイムアウト（デフォルト: 300秒）。
    • session_kwargs: 基盤となる mcp.ClientSession に渡される追加のキーワード引数。

詳細については、langchain_mcp_adapters/client.py の MultiServerMCPClient、StdioConnection、SSEConnection の定義を参照してください。

まとめ

langchain-mcp-adapters ライブラリは、標準化された Model Context Protocol と柔軟な LangChain エコシステムの間のギャップを効果的に埋めます。MultiServerMCPClient と自動ツール変換機能を提供することで、開発者は多様なMCP準拠ツールをLangChainエージェントやLangGraphアプリケーションに簡単に組み込むことができます。

コアワークフロー:

  1. MCPサーバーで @mcp.tool() を使用してツール（およびオプションで @mcp.prompt() を使用してプロンプト）を定義します。
  2. 各サーバーの接続詳細（stdioまたはsse）を使用して MultiServerMCPClient を設定します。
  3. クライアントのコンテキストマネージャ (async with ...) を使用して接続し、client.get_tools() を介してツールを取得します。
  4. 取得したLangChain互換ツールをエージェント（create_react_agent またはカスタムエージェント）に渡します。

これにより、標準化されたプロトコルを通じて特殊な外部ツールを活用する、強力でモジュール化されたAIアプリケーションを構築できます。さらなる洞察については、リポジトリ内のサンプルコードやテストコード (tests/ ディレクトリ) を参照してください。これらは、ライブラリの具体的な使用方法や機能についての理解を深めるのに役立ちます。

5
5
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
5

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?